connect to multiple hosts simultaneously and return the first successful connection along with the instance_id of the winning address.
(
addresses: AddressGroup,
port: u16,
max_connections: u64,
app_id: &str,
)
| 142 | /// connect to multiple hosts simultaneously and return the first successful connection |
| 143 | /// along with the instance_id of the winning address. |
| 144 | pub(crate) async fn connect_multiple_hosts( |
| 145 | addresses: AddressGroup, |
| 146 | port: u16, |
| 147 | max_connections: u64, |
| 148 | app_id: &str, |
| 149 | ) -> Result<(TcpStream, EnteredCounter, String)> { |
| 150 | check_connection_limit(&addresses, max_connections, app_id)?; |
| 151 | |
| 152 | let mut join_set = JoinSet::new(); |
| 153 | for addr in addresses { |
| 154 | let counter = addr.counter.enter(); |
| 155 | let ip = addr.ip; |
| 156 | let instance_id = addr.instance_id; |
| 157 | debug!("connecting to {ip}:{port}"); |
| 158 | let future = TcpStream::connect((ip, port)); |
| 159 | join_set.spawn(async move { |
| 160 | ( |
| 161 | future.await.map_err(|e| (e, ip, port)), |
| 162 | counter, |
| 163 | instance_id, |
| 164 | ) |
| 165 | }); |
| 166 | } |
| 167 | // select the first successful connection |
| 168 | let (connection, counter, instance_id) = loop { |
| 169 | let (result, counter, instance_id) = join_set |
| 170 | .join_next() |
| 171 | .await |
| 172 | .context("No connection success")? |
| 173 | .context("Failed to join the connect task")?; |
| 174 | match result { |
| 175 | Ok(connection) => break (connection, counter, instance_id), |
| 176 | Err((e, addr, port)) => { |
| 177 | info!("failed to connect to app@{addr}:{port}: {e}"); |
| 178 | } |
| 179 | } |
| 180 | }; |
| 181 | debug!("connected to {:?}", connection.peer_addr()); |
| 182 | Ok((connection, counter, instance_id)) |
| 183 | } |
| 184 | |
| 185 | pub(crate) async fn proxy_to_app( |
| 186 | state: Proxy, |
no test coverage detected