fn on_connect_setup(
        &self,
        flow_id: FlowId,
        role: ConnectionRole,
        peer_addr: SocketAddr,
        reader: Box<TlsReadHalf>,
        writer: Box<TlsWriteHalf>,
    ) -> Result<Arc<dyn AsyncTransportEventHandler>, TransportErrorCode> {
        let mut client_map = self.client_map.write().unwrap();
        let client_state = match client_map.get_mut(&flow_id.client_type) {
            Some(client_state) => client_state,
            None => return Err(TransportErrorCode::TransportClientNotFound),
        };
        let event_handler = client_state.event_handler.clone();

        let peer_state = match client_state.peer_map.get_mut(&flow_id.peer_id) {
            Some(client_state) => client_state,
            None => return Err(TransportErrorCode::TransportClientNotFound),
        };
        let flow_state = match peer_state.flow_map.get_mut(&flow_id.flow_tag) {
            Some(flow_state) => flow_state,
            None => return Err(TransportErrorCode::FlowNotFound),
        };

        if let ConnectionState::Connected(_) = flow_state.connection_state {
            // TODO: P2P-516
            return Err(TransportErrorCode::FlowConnectionUp);
        }

        // Spawn write task
        let flow_id_cl = flow_state.flow_id;
        let flow_label_cl = flow_state.flow_label.clone();
        let send_queue_reader = flow_state.send_queue.get_reader();
        let metrics_cl = self.data_plane_metrics.clone();
        let weak_self = self.weak_self.read().unwrap().clone();
        let write_task = async move {
            Self::flow_write_task(
                flow_id_cl,
                flow_label_cl,
                send_queue_reader,
                writer,
                metrics_cl,
                weak_self,
            )
            .await;
        };

        let flow_id_cl = flow_id;
        let flow_label_cl = flow_state.flow_label.clone();
        let event_handler_cl = event_handler.clone();
        let metrics_cl = self.data_plane_metrics.clone();
        let weak_self = self.weak_self.read().unwrap().clone();
        let read_task = async move {
            Self::flow_read_task(
                flow_id_cl,
                flow_label_cl,
                event_handler_cl,
                reader,
                metrics_cl,
                weak_self,
            )
            .await;
        };

        // Spawn the tasks with abort handles so tasks can be aborted if needed.
        let (write_abort_handle, abort_registration) = AbortHandle::new_pair();
        let log_cl = self.log.clone();
        let flow_id_cl = flow_id;
        self.tokio_runtime.spawn(async move {
            if let Err(Aborted) = Abortable::new(write_task, abort_registration).await {
                warn!(
                    log_cl,
                    "DataPlane:: Send task aborted: flow = {:?}", flow_id_cl
                );
            }
        });

        let (read_abort_handle, abort_registration) = AbortHandle::new_pair();
        let log_cl = self.log.clone();
        let flow_id_cl = flow_id;
        self.tokio_runtime.spawn(async move {
            if let Err(Aborted) = Abortable::new(read_task, abort_registration).await {
                warn!(
                    log_cl,
                    "DataPlane:: Receive task aborted: flow = {:?}", flow_id_cl
                );
            }
        });

        let connected_state = Connected {
            peer_addr,
            read_task: read_abort_handle,
            write_task: write_abort_handle,
            role,
        };
        flow_state.update(ConnectionState::Connected(connected_state));
        Ok(event_handler)
    }
/// Handle connection setup. Starts flow read and write tasks.
Handle connection setup. Starts flow read and write tasks.
Handle connection setup . Starts flow read and write tasks .
fn on_connect_setup(
        &self,
        flow_id: FlowId,
        role: ConnectionRole,
        peer_addr: SocketAddr,
        reader: Box<TlsReadHalf>,
        writer: Box<TlsWriteHalf>,
    ) -> Result<Arc<dyn AsyncTransportEventHandler>, TransportErrorCode> {
        let mut client_map = self.client_map.write().unwrap();
        let client_state = match client_map.get_mut(&flow_id.client_type) {
            Some(client_state) => client_state,
            None => return Err(TransportErrorCode::TransportClientNotFound),
        };
        let event_handler = client_state.event_handler.clone();
        let peer_state = match client_state.peer_map.get_mut(&flow_id.peer_id) {
            Some(client_state) => client_state,
            None => return Err(TransportErrorCode::TransportClientNotFound),
        };
        let flow_state = match peer_state.flow_map.get_mut(&flow_id.flow_tag) {
            Some(flow_state) => flow_state,
            None => return Err(TransportErrorCode::FlowNotFound),
        };
        if let ConnectionState::Connected(_) = flow_state.connection_state {
            // TODO: P2P-516
            return Err(TransportErrorCode::FlowConnectionUp);
        }
        // Spawn write task
        let flow_id_cl = flow_state.flow_id;
        let flow_label_cl = flow_state.flow_label.clone();
        let send_queue_reader = flow_state.send_queue.get_reader();
        let metrics_cl = self.data_plane_metrics.clone();
        let weak_self = self.weak_self.read().unwrap().clone();
        let write_task = async move {
            Self::flow_write_task(
                flow_id_cl,
                flow_label_cl,
                send_queue_reader,
                writer,
                metrics_cl,
                weak_self,
            )
            .await;
        };
        let flow_id_cl = flow_id;
        let flow_label_cl = flow_state.flow_label.clone();
        let event_handler_cl = event_handler.clone();
        let metrics_cl = self.data_plane_metrics.clone();
        let weak_self = self.weak_self.read().unwrap().clone();
        let read_task = async move {
            Self::flow_read_task(
                flow_id_cl,
                flow_label_cl,
                event_handler_cl,
                reader,
                metrics_cl,
                weak_self,
            )
            .await;
        };
        // Spawn the tasks with abort handles so tasks can be aborted if needed.
        let (write_abort_handle, abort_registration) = AbortHandle::new_pair();
        let log_cl = self.log.clone();
        let flow_id_cl = flow_id;
        self.tokio_runtime.spawn(async move {
            if let Err(Aborted) = Abortable::new(write_task, abort_registration).await {
                warn!(
                    log_cl,
                    "DataPlane:: Send task aborted: flow = {:?}", flow_id_cl
                );
            }
        });
        let (read_abort_handle, abort_registration) = AbortHandle::new_pair();
        let log_cl = self.log.clone();
        let flow_id_cl = flow_id;
        self.tokio_runtime.spawn(async move {
            if let Err(Aborted) = Abortable::new(read_task, abort_registration).await {
                warn!(
                    log_cl,
                    "DataPlane:: Receive task aborted: flow = {:?}", flow_id_cl
                );
            }
        });
        let connected_state = Connected {
            peer_addr,
            read_task: read_abort_handle,
            write_task: write_abort_handle,
            role,
        };
        flow_state.update(ConnectionState::Connected(connected_state));
        Ok(event_handler)
    }
fn on_connect_setup ( & self , flow_id : FlowId , role : ConnectionRole , peer_addr : SocketAddr , reader : Box < TlsReadHalf > , writer : Box < TlsWriteHalf > , ) -> Result < Arc < dyn AsyncTransportEventHandler > , TransportErrorCode > { let mut client_map = self . client_map . write ( ) . unwrap ( ) ; let client_state = match client_map . get_mut ( & flow_id . client_type ) { Some ( client_state ) => client_state , None => return Err ( TransportErrorCode :: TransportClientNotFound ) , } ; let event_handler = client_state . event_handler . clone ( ) ; let peer_state = match client_state . peer_map . get_mut ( & flow_id . peer_id ) { Some ( client_state ) => client_state , None => return Err ( TransportErrorCode :: TransportClientNotFound ) , } ; let flow_state = match peer_state . flow_map . get_mut ( & flow_id . flow_tag ) { Some ( flow_state ) => flow_state , None => return Err ( TransportErrorCode :: FlowNotFound ) , } ; if let ConnectionState :: Connected ( _ ) = flow_state . connection_state { // TODO: P2P-516 return Err ( TransportErrorCode :: FlowConnectionUp ) ; } // Spawn write task let flow_id_cl = flow_state . flow_id ; let flow_label_cl = flow_state . flow_label . clone ( ) ; let send_queue_reader = flow_state . send_queue . get_reader ( ) ; let metrics_cl = self . data_plane_metrics . clone ( ) ; let weak_self = self . weak_self . read ( ) . unwrap ( ) . clone ( ) ; let write_task = async move { Self :: flow_write_task ( flow_id_cl , flow_label_cl , send_queue_reader , writer , metrics_cl , weak_self , ) . await ; } ; let flow_id_cl = flow_id ; let flow_label_cl = flow_state . flow_label . clone ( ) ; let event_handler_cl = event_handler . clone ( ) ; let metrics_cl = self . data_plane_metrics . clone ( ) ; let weak_self = self . weak_self . read ( ) . unwrap ( ) . clone ( ) ; let read_task = async move { Self :: flow_read_task ( flow_id_cl , flow_label_cl , event_handler_cl , reader , metrics_cl , weak_self , ) . await ; } ; // Spawn the tasks with abort handles so tasks can be aborted if needed. let ( write_abort_handle , abort_registration ) = AbortHandle :: new_pair ( ) ; let log_cl = self . log . clone ( ) ; let flow_id_cl = flow_id ; self . tokio_runtime . spawn ( async move { if let Err ( Aborted ) = Abortable :: new ( write_task , abort_registration ) . await { warn ! ( log_cl " " flow_id_cl ) ; } } ) ; let ( read_abort_handle , abort_registration ) = AbortHandle :: new_pair ( ) ; let log_cl = self . log . clone ( ) ; let flow_id_cl = flow_id ; self . tokio_runtime . spawn ( async move { if let Err ( Aborted ) = Abortable :: new ( read_task , abort_registration ) . await { warn ! ( log_cl " " flow_id_cl ) ; } } ) ; let connected_state = Connected { peer_addr , read_task : read_abort_handle , write_task : write_abort_handle , role , } ; flow_state . update ( ConnectionState :: Connected ( connected_state ) ) ; Ok ( event_handler ) }
Handle connection setup.
Handle connection setup .
&self
flow_id
FlowId
role
ConnectionRole
peer_addr
SocketAddr
reader
Box<TlsReadHalf>
writer
Box<TlsWriteHalf>
&self
flow_id
FlowId
role
ConnectionRole
peer_addr
SocketAddr
reader
Box<TlsReadHalf>
writer
Box<TlsWriteHalf>