async fn run_tcp_listeners(
    bindings: Vec<TcpListener>,
    target_addr: (Option<String>, u16),
    weak_tunnel: Weak<dyn Tunnel + Send + Sync + Unpin>,
    request_client_handler: Weak<RequestClientHandler>,
    no_new_requests_listener: CancellationToken,
  ) -> Result<(), ServiceError> {
    let span =
      tracing::span!(tracing::Level::DEBUG, "demand_proxy_forwarding", target = ?target_addr);
    use futures::stream::TryStreamExt;
    let parsed_addr = Arc::new(target_addr);
    let fut = futures::stream::iter(
      bindings
        .into_iter()
        .map(move |listener| {
          (
            listener,
            weak_tunnel.clone(),
            request_client_handler.clone(),
            no_new_requests_listener.clone(),
            Arc::clone(&parsed_addr),
          )
        })
        .map(Result::<_, ServiceError>::Ok),
    )
    .try_for_each_concurrent(
      None,
      |(listener, weak_tunnel, request_client_handler, no_new_requests_listener, parsed_addr)| {
        Self::run_tcp_listener(
          parsed_addr,
          listener,
          weak_tunnel,
          request_client_handler,
          no_new_requests_listener,
        )
      },
    )
    .instrument(span);
    // Run as a tokio::task to ensure scheduling across tunnels
    tokio::task::spawn(fut)
      // Treat JoinErrors as just another ServiceError
      .map(|res| match res {
        Err(_join_error) => Err(ServiceError::DependencyFailure),
        Ok(x) => x,
      })
      .await
  }
/// Handle forwarding concurrently across all streams requested by the TCP ports bound for each listener
Handle forwarding concurrently across all streams requested by the TCP ports bound for each listener
Handle forwarding concurrently across all streams requested by the TCP ports bound for each listener
async fn run_tcp_listeners(
    bindings: Vec<TcpListener>,
    target_addr: (Option<String>, u16),
    weak_tunnel: Weak<dyn Tunnel + Send + Sync + Unpin>,
    request_client_handler: Weak<RequestClientHandler>,
    no_new_requests_listener: CancellationToken,
  ) -> Result<(), ServiceError> {
    let span =
      tracing::span!(tracing::Level::DEBUG, "demand_proxy_forwarding", target = ?target_addr);
    use futures::stream::TryStreamExt;
    let parsed_addr = Arc::new(target_addr);
    let fut = futures::stream::iter(
      bindings
        .into_iter()
        .map(move |listener| {
          (
            listener,
            weak_tunnel.clone(),
            request_client_handler.clone(),
            no_new_requests_listener.clone(),
            Arc::clone(&parsed_addr),
          )
        })
        .map(Result::<_, ServiceError>::Ok),
    )
    .try_for_each_concurrent(
      None,
      |(listener, weak_tunnel, request_client_handler, no_new_requests_listener, parsed_addr)| {
        Self::run_tcp_listener(
          parsed_addr,
          listener,
          weak_tunnel,
          request_client_handler,
          no_new_requests_listener,
        )
      },
    )
    .instrument(span);
    // Run as a tokio::task to ensure scheduling across tunnels
    tokio::task::spawn(fut)
      // Treat JoinErrors as just another ServiceError
      .map(|res| match res {
        Err(_join_error) => Err(ServiceError::DependencyFailure),
        Ok(x) => x,
      })
      .await
  }
async fn run_tcp_listeners ( bindings : Vec < TcpListener > , target_addr : ( Option < String > , u16 ) , weak_tunnel : Weak < dyn Tunnel + Send + Sync + Unpin > , request_client_handler : Weak < RequestClientHandler > , no_new_requests_listener : CancellationToken , ) -> Result < ( ) , ServiceError > { let span = tracing :: span ! ( tracing Level DEBUG " " target target_addr ) ; use futures :: stream :: TryStreamExt ; let parsed_addr = Arc :: new ( target_addr ) ; let fut = futures :: stream :: iter ( bindings . into_iter ( ) . map ( move | listener | { ( listener , weak_tunnel . clone ( ) , request_client_handler . clone ( ) , no_new_requests_listener . clone ( ) , Arc :: clone ( & parsed_addr ) , ) } ) . map ( Result :: < _ , ServiceError > :: Ok ) , ) . try_for_each_concurrent ( None , | ( listener , weak_tunnel , request_client_handler , no_new_requests_listener , parsed_addr ) | { Self :: run_tcp_listener ( parsed_addr , listener , weak_tunnel , request_client_handler , no_new_requests_listener , ) } , ) . instrument ( span ) ; // Run as a tokio::task to ensure scheduling across tunnels tokio :: task :: spawn ( fut ) // Treat JoinErrors as just another ServiceError . map ( | res | match res { Err ( _join_error ) => Err ( ServiceError :: DependencyFailure ) , Ok ( x ) => x , } ) . await }
Handle forwarding concurrently across all streams requested by the TCP ports bound for each listener
Handle forwarding concurrently across all streams requested by the TCP ports bound for each listener
bindings
Vec<TcpListener>
target_addr
(Option<String>, u16)
weak_tunnel
Weak<dyn Tunnel + Send + Sync + Unpin>
request_client_handler
Weak<RequestClientHandler>
no_new_requests_listener
CancellationToken
bindings
Vec<TcpListener>
target_addr
(Option<String>, u16)
weak_tunnel
Weak<dyn Tunnel + Send + Sync + Unpin>
request_client_handler
Weak<RequestClientHandler>
no_new_requests_listener
CancellationToken