fn spawn_accept_task(
        &self,
        client_type: TransportClientType,
        flow_tag: FlowTag,
        mut tcp_listener: TcpListener,
        canceler: Arc<AtomicBool>,
    ) {
        let weak_self = self.weak_self.read().unwrap().clone();
        let tokio_runtime = self.tokio_runtime.clone();
        let metrics = self.control_plane_metrics.clone();
        self.tokio_runtime.spawn(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,
                };
                let timeout = delay_for(Duration::from_millis(CANCEL_CHECK_PERIOD_MILLISECONDS));
                // TODO: P2P-515
                match future::select(tcp_listener.accept().boxed(), timeout).await {
                    Either::Left((result, _)) => {
                        match result {
                            Ok((stream, _)) => {
                                let metrics = metrics.clone();
                                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,
                                );
                            }
                        }
                    }
                    Either::Right((_, _)) => {
                        if canceler.load(Ordering::SeqCst) {
                            return;
                        }
                    }
                }
            }
        });
    }
/// 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,
        mut tcp_listener: TcpListener,
        canceler: Arc<AtomicBool>,
    ) {
        let weak_self = self.weak_self.read().unwrap().clone();
        let tokio_runtime = self.tokio_runtime.clone();
        let metrics = self.control_plane_metrics.clone();
        self.tokio_runtime.spawn(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,
                };
                let timeout = delay_for(Duration::from_millis(CANCEL_CHECK_PERIOD_MILLISECONDS));
                // TODO: P2P-515
                match future::select(tcp_listener.accept().boxed(), timeout).await {
                    Either::Left((result, _)) => {
                        match result {
                            Ok((stream, _)) => {
                                let metrics = metrics.clone();
                                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,
                                );
                            }
                        }
                    }
                    Either::Right((_, _)) => {
                        if canceler.load(Ordering::SeqCst) {
                            return;
                        }
                    }
                }
            }
        });
    }
fn spawn_accept_task ( & self , client_type : TransportClientType , flow_tag : FlowTag , mut tcp_listener : TcpListener , canceler : Arc < AtomicBool > , ) { let weak_self = self . weak_self . read ( ) . unwrap ( ) . clone ( ) ; let tokio_runtime = self . tokio_runtime . clone ( ) ; let metrics = self . control_plane_metrics . clone ( ) ; self . tokio_runtime . spawn ( 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 , } ; let timeout = delay_for ( Duration :: from_millis ( CANCEL_CHECK_PERIOD_MILLISECONDS ) ) ; // TODO: P2P-515 match future :: select ( tcp_listener . accept ( ) . boxed ( ) , timeout ) . await { Either :: Left ( ( result , _ ) ) => { match result { Ok ( ( stream , _ ) ) => { let metrics = metrics . clone ( ) ; 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 ) ; } } } Either :: Right ( ( _ , _ ) ) => { if canceler . load ( Ordering :: SeqCst ) { return ; } } } } } ) ; }
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
canceler
Arc<AtomicBool>
&self
client_type
TransportClientType
flow_tag
FlowTag
tcp_listener
TcpListener
canceler
Arc<AtomicBool>