| 130 | } |
| 131 | |
| 132 | pub async fn initialize_server( |
| 133 | gsettings: &GlobalSettings, |
| 134 | mut server_cfg: ServerConfig, |
| 135 | worker_id_initial_value: WorkerId, |
| 136 | queue_id_initial_value: QueueId, |
| 137 | truncate_log: Option<u64>, |
| 138 | ) -> anyhow::Result<( |
| 139 | impl Future<Output = anyhow::Result<()>> + use<>, |
| 140 | Arc<Notify>, |
| 141 | StateRef, |
| 142 | Senders, |
| 143 | )> { |
| 144 | let server_directory = gsettings.server_directory(); |
| 145 | |
| 146 | let client_listener = TcpListener::bind(SocketAddr::new( |
| 147 | Ipv4Addr::UNSPECIFIED.into(), |
| 148 | server_cfg.client_port.unwrap_or(0), |
| 149 | )) |
| 150 | .await |
| 151 | .with_context(|| "Cannot create HQ server socket".to_string())?; |
| 152 | let client_port = client_listener.local_addr()?.port(); |
| 153 | |
| 154 | let worker_listen_address = SocketAddr::new( |
| 155 | Ipv4Addr::UNSPECIFIED.into(), |
| 156 | server_cfg.worker_port.unwrap_or(0), |
| 157 | ); |
| 158 | log::debug!("Waiting for workers on {worker_listen_address:?}"); |
| 159 | let worker_listener = TcpListener::bind(worker_listen_address).await?; |
| 160 | |
| 161 | let server_uid = server_cfg |
| 162 | .server_uid |
| 163 | .take() |
| 164 | .unwrap_or_else(generate_server_uid); |
| 165 | let worker_key = server_cfg.worker_secret_key.take(); |
| 166 | let client_key = server_cfg.client_secret_key.take(); |
| 167 | |
| 168 | if worker_key.is_none() { |
| 169 | log::warn!("!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!"); |
| 170 | log::warn!("Server is started with unprotected worker connections"); |
| 171 | log::warn!("Anyone can connect as a worker"); |
| 172 | log::warn!("!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!"); |
| 173 | } |
| 174 | |
| 175 | if client_key.is_none() { |
| 176 | log::warn!("!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!"); |
| 177 | log::warn!("Server is started with unprotected client connections"); |
| 178 | log::warn!("Anyone can connect as a client"); |
| 179 | log::warn!("!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!"); |
| 180 | } |
| 181 | |
| 182 | let state_ref = StateRef::new(ServerInfo { |
| 183 | version: HQ_VERSION.to_string(), |
| 184 | server_uid: server_uid.clone(), |
| 185 | client_host: server_cfg.client_host.clone(), |
| 186 | worker_host: server_cfg.worker_host.clone(), |
| 187 | client_port, |
| 188 | worker_port: 0, // Will be set later |
| 189 | pid: std::process::id(), |