async fn request_loop<T, U>(
    conn_id: usize,
    state: Arc<Mutex<State>>,
    mut transport: TransportReadHalf<T, U>,
    tx: mpsc::Sender<Response>,
) where
    T: AsyncRead + Send + Unpin + 'static,
    U: Codec,
{
    loop {
        match transport.receive::<Request>().await {
            Ok(Some(req)) => {
                debug!(
                    "<Conn @ {}> Received request of type{} {}",
                    conn_id,
                    if req.payload.len() > 1 { "s" } else { "" },
                    req.to_payload_type_string()
                );

                if let Err(x) = handler::process(conn_id, Arc::clone(&state), req, tx.clone()).await
                {
                    error!("<Conn @ {}> {}", conn_id, x);
                    break;
                }
            }
            Ok(None) => {
                trace!("<Conn @ {}> Input from connection closed", conn_id);
                break;
            }
            Err(x) => {
                error!("<Conn @ {}> {}", conn_id, x);
                break;
            }
        }
    }

    // Properly close off any associated process' stdin given that we can't get new
    // requests to send more stdin to them
    state.lock().await.close_stdin_for_connection(conn_id);
}
/// Repeatedly reads in new requests, processes them, and sends their responses to the
/// response loop
Repeatedly reads in new requests, processes them, and sends their responses to the
response loop
Repeatedly reads in new requests processes them and sends their responses to the response loop
async fn request_loop<T, U>(
    conn_id: usize,
    state: Arc<Mutex<State>>,
    mut transport: TransportReadHalf<T, U>,
    tx: mpsc::Sender<Response>,
) where
    T: AsyncRead + Send + Unpin + 'static,
    U: Codec,
{
    loop {
        match transport.receive::<Request>().await {
            Ok(Some(req)) => {
                debug!(
                    "<Conn @ {}> Received request of type{} {}",
                    conn_id,
                    if req.payload.len() > 1 { "s" } else { "" },
                    req.to_payload_type_string()
                );
                if let Err(x) = handler::process(conn_id, Arc::clone(&state), req, tx.clone()).await
                {
                    error!("<Conn @ {}> {}", conn_id, x);
                    break;
                }
            }
            Ok(None) => {
                trace!("<Conn @ {}> Input from connection closed", conn_id);
                break;
            }
            Err(x) => {
                error!("<Conn @ {}> {}", conn_id, x);
                break;
            }
        }
    }
    // Properly close off any associated process' stdin given that we can't get new
    // requests to send more stdin to them
    state.lock().await.close_stdin_for_connection(conn_id);
}
async fn request_loop < T , U > ( conn_id : usize , state : Arc < Mutex < State > > , mut transport : TransportReadHalf < T , U > , tx : mpsc :: Sender < Response > , ) where T : AsyncRead + Send + Unpin + ' static , U : Codec , { loop { match transport . receive :: < Request > ( ) . await { Ok ( Some ( req ) ) => { debug ! ( " " conn_id if req payload len ( ) 1 { " " } else { " " } req to_payload_type_string ( ) ) ; if let Err ( x ) = handler :: process ( conn_id , Arc :: clone ( & state ) , req , tx . clone ( ) ) . await { error ! ( " " conn_id x ) ; break ; } } Ok ( None ) => { trace ! ( " " conn_id ) ; break ; } Err ( x ) => { error ! ( " " conn_id x ) ; break ; } } } // Properly close off any associated process' stdin given that we can't get new // requests to send more stdin to them state . lock ( ) . await . close_stdin_for_connection ( conn_id ) ; }
Repeatedly reads in new requests, processes them, and sends their responses to the
response loop
Repeatedly reads in new requests processes them and sends their responses to the response loop
conn_id
usize
state
Arc<Mutex<State>>
transport
TransportReadHalf<T, U>
tx
mpsc::Sender<Response>