(
&self,
inbound: TcpStream,
pp_header: ProxyHeader,
buffer: Vec<u8>,
app_id: &str,
port: u16,
h2: bool,
)
| 271 | } |
| 272 | |
| 273 | pub(super) async fn proxy( |
| 274 | &self, |
| 275 | inbound: TcpStream, |
| 276 | pp_header: ProxyHeader, |
| 277 | buffer: Vec<u8>, |
| 278 | app_id: &str, |
| 279 | port: u16, |
| 280 | h2: bool, |
| 281 | ) -> Result<()> { |
| 282 | if app_id == "health" { |
| 283 | return self.handle_health_check(inbound, buffer, port, h2).await; |
| 284 | } |
| 285 | if app_id == "gateway" { |
| 286 | return self.handle_this_node(inbound, buffer, port, h2).await; |
| 287 | } |
| 288 | let addresses = self |
| 289 | .lock() |
| 290 | .select_top_n_hosts(app_id) |
| 291 | .with_context(|| format!("app <{app_id}> not found"))?; |
| 292 | let addresses = filter_allowed_addresses(self, addresses, app_id, port)?; |
| 293 | debug!("selected top n hosts: {addresses:?}"); |
| 294 | let tls_stream = self.tls_accept(inbound, buffer, h2).await?; |
| 295 | let max_connections = self.config.proxy.max_connections_per_app; |
| 296 | let (mut outbound, _counter, instance_id) = timeout( |
| 297 | self.config.proxy.timeouts.connect, |
| 298 | connect_multiple_hosts(addresses, port, max_connections, app_id), |
| 299 | ) |
| 300 | .await |
| 301 | .map_err(|_| anyhow!("connecting timeout"))? |
| 302 | .context("failed to connect to app")?; |
| 303 | if should_send_pp(self, &instance_id, port) { |
| 304 | let pp_header_bin = |
| 305 | proxy_protocol::encode(pp_header).context("failed to encode pp header")?; |
| 306 | outbound.write_all(&pp_header_bin).await?; |
| 307 | } |
| 308 | bridge( |
| 309 | IgnoreUnexpectedEofStream::new(tls_stream), |
| 310 | outbound, |
| 311 | &self.config.proxy, |
| 312 | ) |
| 313 | .await |
| 314 | .context("bridge error")?; |
| 315 | Ok(()) |
| 316 | } |
| 317 | } |
| 318 | |
| 319 | #[pin_project::pin_project] |
no test coverage detected