/// Starts the async task to accept the incoming TcpStreams in server mode.
Starts the async task to accept the incoming TcpStreams in server mode.
Starts the async task to accept the incoming TcpStreams in server mode .
fn spawn_accept_task(
        &self,
        client_type: TransportClientType,
        flow_tag: FlowTag,
        tcp_listener: TcpListener,
    ) -> AbortHandle {
        let weak_self = self.weak_self.read().unwrap().clone();
        let tokio_runtime = self.tokio_runtime.clone();
        let metrics = self.control_plane_metrics.clone();
        let accept_task = async move {
            let local_addr = match tcp_listener.local_addr() {
                Ok(addr) => addr,
                _ => return,
            };
            loop {
                // If the TransportImpl has been deleted, abort.
                let arc_self = match weak_self.upgrade() {
                    Some(arc_self) => arc_self,
                    _ => return,
                };
                match tcp_listener.accept().await {
                    Ok((stream, _)) => {
                        let metrics = metrics.clone();
                        metrics
                            .tcp_accepts
                            .with_label_values(&[&flow_tag.to_string()])
                            .inc();
                        tokio_runtime.spawn(async move {
                            // Errors are reported in set_sockopts
                            let stream = match Self::set_send_sockopts(stream, &arc_self.log) {
                                Ok(stream) => stream,
                                Err(_) => return,
                            };
                            // Errors are reported in tls_server_handshake
                            if let Ok(()) = arc_self
                                .tls_server_handshake(client_type, flow_tag, stream)
                                .await
                            {
                                metrics
                                    .tcp_accept_conn_success
                                    .with_label_values(&[&flow_tag.to_string()])
                                    .inc();
                            }
                        });
                    }
                    Err(e) => {
                        metrics
                            .tcp_accept_conn_err
                            .with_label_values(&[&flow_tag.to_string()])
                            .inc();
                        warn!(
                            arc_self.log,
                            "ControlPlane::accept(): local_addr = {:?} flow = {:?}, err = {:?}",
                            local_addr,
                            flow_tag,
                            e,
                        );
                    }
                }
            }
        };
        let (abort_handle, abort_registration) = AbortHandle::new_pair();
        let log_cl = self.log.clone();
        self.tokio_runtime.spawn(async move {
            if let Err(Aborted) = Abortable::new(accept_task, abort_registration).await {
                warn!(log_cl, "ControlPlane: accept task aborted");
            }
        });
        abort_handle
    }
fn spawn_accept_task ( & self , client_type : TransportClientType , flow_tag : FlowTag , tcp_listener : TcpListener , ) -> AbortHandle { let weak_self = self . weak_self . read ( ) . unwrap ( ) . clone ( ) ; let tokio_runtime = self . tokio_runtime . clone ( ) ; let metrics = self . control_plane_metrics . clone ( ) ; let accept_task = async move { let local_addr = match tcp_listener . local_addr ( ) { Ok ( addr ) => addr , _ => return , } ; loop { // If the TransportImpl has been deleted, abort. let arc_self = match weak_self . upgrade ( ) { Some ( arc_self ) => arc_self , _ => return , } ; match tcp_listener . accept ( ) . await { Ok ( ( stream , _ ) ) => { let metrics = metrics . clone ( ) ; metrics . tcp_accepts . with_label_values ( & [ & flow_tag . to_string ( ) ] ) . inc ( ) ; tokio_runtime . spawn ( async move { // Errors are reported in set_sockopts let stream = match Self :: set_send_sockopts ( stream , & arc_self . log ) { Ok ( stream ) => stream , Err ( _ ) => return , } ; // Errors are reported in tls_server_handshake if let Ok ( ( ) ) = arc_self . tls_server_handshake ( client_type , flow_tag , stream ) . await { metrics . tcp_accept_conn_success . with_label_values ( & [ & flow_tag . to_string ( ) ] ) . inc ( ) ; } } ) ; } Err ( e ) => { metrics . tcp_accept_conn_err . with_label_values ( & [ & flow_tag . to_string ( ) ] ) . inc ( ) ; warn ! ( arc_self log " " local_addr flow_tag e ) ; } } } } ; let ( abort_handle , abort_registration ) = AbortHandle :: new_pair ( ) ; let log_cl = self . log . clone ( ) ; self . tokio_runtime . spawn ( async move { if let Err ( Aborted ) = Abortable :: new ( accept_task , abort_registration ) . await { warn ! ( log_cl " " ) ; } } ) ; abort_handle }
Starts the async task to accept the incoming TcpStreams in server mode.
Starts the async task to accept the incoming TcpStreams in server mode .
&self
client_type
TransportClientType
flow_tag
FlowTag
tcp_listener
TcpListener