/// Per-flow receive task. Reads the messages from the socket and passes to
/// the client.
Per-flow receive task. Reads the messages from the socket and passes to
the client.
Per - flow receive task . Reads the messages from the socket and passes to the client .
async fn flow_read_task(
        flow_id: FlowId,
        flow_label: String,
        event_handler: Arc<dyn AsyncTransportEventHandler>,
        mut reader: Box<TlsReadHalf>,
        metrics: DataPlaneMetrics,
        state: Weak<TransportImpl>,
    ) {
        let heartbeat_timeout = Duration::from_millis(TRANSPORT_HEARTBEAT_WAIT_INTERVAL_MS);
        let _updater = MetricsUpdater::new(metrics.clone(), false);
        let flow_tag = flow_id.flow_tag.to_string();
        loop {
            // If the TransportImpl has been deleted, abort.
            let state = match state.upgrade() {
                Some(transport) => transport,
                _ => return,
            };
            // Read the next message from the socket
            let ret = Self::read_one_message(&mut reader, heartbeat_timeout).await;
            if ret.is_err() {
                warn!(
                    state.log,
                    "DataPlane::flow_read_task(): failed to receive message: flow: {:?}, {:?}",
                    flow_id,
                    ret.as_ref().err(),
                );
                if let Err(ReadError::SocketReadTimeOut) = ret {
                    event_handler
                        .error(flow_id, TransportErrorCode::TimeoutExpired)
                        .await;
                    metrics
                        .socket_heart_beat_timeouts
                        .with_label_values(&[&flow_label, &flow_tag])
                        .inc();
                }
                state.on_disconnect(flow_id).await;
                return;
            }
            // Process the received message
            let (header, payload) = ret.unwrap();
            if header.flags & TRANSPORT_FLAGS_IS_HEARTBEAT != 0 {
                // It's an empty heartbeat message -- do nothing
                metrics
                    .heart_beats_received
                    .with_label_values(&[&flow_label, &flow_tag])
                    .inc();
                continue;
            }
            // Pass up sender indicated error
            if header.flags & TRANSPORT_FLAGS_SENDER_ERROR != 0 {
                event_handler
                    .error(flow_id, TransportErrorCode::SenderErrorIndicated)
                    .await;
                metrics
                    .send_errors_received
                    .with_label_values(&[&flow_label, &flow_tag])
                    .inc();
            }
            // Pass up the received message
            // Errors out for unsolicited messages, decoding errors and p2p
            // shutdowns.
            let payload = payload.unwrap();
            metrics
                .socket_read_bytes
                .with_label_values(&[&flow_label, &flow_tag])
                .inc_by(payload.0.len() as u64);
            let start_time = Instant::now();
            let _ = event_handler.send_message(flow_id, payload).await;
            metrics
                .client_send_time_msec
                .with_label_values(&[&flow_label, &flow_tag])
                .observe(start_time.elapsed().as_millis() as f64);
        }
    }
async fn flow_read_task ( flow_id : FlowId , flow_label : String , event_handler : Arc < dyn AsyncTransportEventHandler > , mut reader : Box < TlsReadHalf > , metrics : DataPlaneMetrics , state : Weak < TransportImpl > , ) { let heartbeat_timeout = Duration :: from_millis ( TRANSPORT_HEARTBEAT_WAIT_INTERVAL_MS ) ; let _updater = MetricsUpdater :: new ( metrics . clone ( ) , false ) ; let flow_tag = flow_id . flow_tag . to_string ( ) ; loop { // If the TransportImpl has been deleted, abort. let state = match state . upgrade ( ) { Some ( transport ) => transport , _ => return , } ; // Read the next message from the socket let ret = Self :: read_one_message ( & mut reader , heartbeat_timeout ) . await ; if ret . is_err ( ) { warn ! ( state log " " flow_id ret as_ref ( ) err ( ) ) ; if let Err ( ReadError :: SocketReadTimeOut ) = ret { event_handler . error ( flow_id , TransportErrorCode :: TimeoutExpired ) . await ; metrics . socket_heart_beat_timeouts . with_label_values ( & [ & flow_label , & flow_tag ] ) . inc ( ) ; } state . on_disconnect ( flow_id ) . await ; return ; } // Process the received message let ( header , payload ) = ret . unwrap ( ) ; if header . flags & TRANSPORT_FLAGS_IS_HEARTBEAT != 0 { // It's an empty heartbeat message -- do nothing metrics . heart_beats_received . with_label_values ( & [ & flow_label , & flow_tag ] ) . inc ( ) ; continue ; } // Pass up sender indicated error if header . flags & TRANSPORT_FLAGS_SENDER_ERROR != 0 { event_handler . error ( flow_id , TransportErrorCode :: SenderErrorIndicated ) . await ; metrics . send_errors_received . with_label_values ( & [ & flow_label , & flow_tag ] ) . inc ( ) ; } // Pass up the received message // Errors out for unsolicited messages, decoding errors and p2p // shutdowns. let payload = payload . unwrap ( ) ; metrics . socket_read_bytes . with_label_values ( & [ & flow_label , & flow_tag ] ) . inc_by ( payload . 0 . len ( ) as u64 ) ; let start_time = Instant :: now ( ) ; let _ = event_handler . send_message ( flow_id , payload ) . await ; metrics . client_send_time_msec . with_label_values ( & [ & flow_label , & flow_tag ] ) . observe ( start_time . elapsed ( ) . as_millis ( ) as f64 ) ; } }
Per-flow receive task.
Per - flow receive task .
flow_id
FlowId
flow_label
String
event_handler
Arc<dyn AsyncTransportEventHandler>
reader
Box<TlsReadHalf>
metrics
DataPlaneMetrics
state
Weak<TransportImpl>