async fn on_new_conn<T, U>(
    transport: Transport<T, U>,
    conn_id: usize,
    state: Arc<Mutex<State>>,
    tracker: Option<Arc<Mutex<ConnTracker>>>,
    max_msg_capacity: usize,
) -> io::Result<JoinHandle<()>>
where
    T: DataStream,
    U: Codec + Send + 'static,
{
    // Update our tracker to reflect the new connection
    if let Some(ct) = tracker.as_ref() {
        ct.lock().await.increment();
    }

    // Split the transport into read and write halves so we can handle input
    // and output concurrently
    let (t_read, t_write) = transport.into_split();
    let (tx, rx) = mpsc::channel(max_msg_capacity);

    // Spawn a new task that loops to handle requests from the client
    let state_2 = Arc::clone(&state);
    let req_task = tokio::spawn(async move {
        request_loop(conn_id, state_2, t_read, tx).await;
    });

    // Spawn a new task that loops to handle responses to the client
    let res_task = tokio::spawn(async move { response_loop(conn_id, t_write, rx).await });

    // Spawn cleanup task that waits on our req & res tasks to complete
    let cleanup_task = tokio::spawn(async move {
        // Wait for both receiving and sending tasks to complete before marking
        // the connection as complete
        let _ = tokio::join!(req_task, res_task);

        state.lock().await.cleanup_connection(conn_id).await;
        if let Some(ct) = tracker.as_ref() {
            ct.lock().await.decrement();
        }
    });

    Ok(cleanup_task)
}
/// Processes a new connection, performing a handshake, and then spawning two tasks to handle
/// input and output, returning join handles for the input and output tasks respectively
Processes a new connection, performing a handshake, and then spawning two tasks to handle
input and output, returning join handles for the input and output tasks respectively
Processes a new connection performing a handshake and then spawning two tasks to handle input and output returning join handles for the input and output tasks respectively
async fn on_new_conn<T, U>(
    transport: Transport<T, U>,
    conn_id: usize,
    state: Arc<Mutex<State>>,
    tracker: Option<Arc<Mutex<ConnTracker>>>,
    max_msg_capacity: usize,
) -> io::Result<JoinHandle<()>>
where
    T: DataStream,
    U: Codec + Send + 'static,
{
    // Update our tracker to reflect the new connection
    if let Some(ct) = tracker.as_ref() {
        ct.lock().await.increment();
    }
    // Split the transport into read and write halves so we can handle input
    // and output concurrently
    let (t_read, t_write) = transport.into_split();
    let (tx, rx) = mpsc::channel(max_msg_capacity);
    // Spawn a new task that loops to handle requests from the client
    let state_2 = Arc::clone(&state);
    let req_task = tokio::spawn(async move {
        request_loop(conn_id, state_2, t_read, tx).await;
    });
    // Spawn a new task that loops to handle responses to the client
    let res_task = tokio::spawn(async move { response_loop(conn_id, t_write, rx).await });
    // Spawn cleanup task that waits on our req & res tasks to complete
    let cleanup_task = tokio::spawn(async move {
        // Wait for both receiving and sending tasks to complete before marking
        // the connection as complete
        let _ = tokio::join!(req_task, res_task);
        state.lock().await.cleanup_connection(conn_id).await;
        if let Some(ct) = tracker.as_ref() {
            ct.lock().await.decrement();
        }
    });
    Ok(cleanup_task)
}
async fn on_new_conn < T , U > ( transport : Transport < T , U > , conn_id : usize , state : Arc < Mutex < State > > , tracker : Option < Arc < Mutex < ConnTracker > > > , max_msg_capacity : usize , ) -> io :: Result < JoinHandle < ( ) > > where T : DataStream , U : Codec + Send + ' static , { // Update our tracker to reflect the new connection if let Some ( ct ) = tracker . as_ref ( ) { ct . lock ( ) . await . increment ( ) ; } // Split the transport into read and write halves so we can handle input // and output concurrently let ( t_read , t_write ) = transport . into_split ( ) ; let ( tx , rx ) = mpsc :: channel ( max_msg_capacity ) ; // Spawn a new task that loops to handle requests from the client let state_2 = Arc :: clone ( & state ) ; let req_task = tokio :: spawn ( async move { request_loop ( conn_id , state_2 , t_read , tx ) . await ; } ) ; // Spawn a new task that loops to handle responses to the client let res_task = tokio :: spawn ( async move { response_loop ( conn_id , t_write , rx ) . await } ) ; // Spawn cleanup task that waits on our req & res tasks to complete let cleanup_task = tokio :: spawn ( async move { // Wait for both receiving and sending tasks to complete before marking // the connection as complete let _ = tokio :: join ! ( req_task res_task ) ; state . lock ( ) . await . cleanup_connection ( conn_id ) . await ; if let Some ( ct ) = tracker . as_ref ( ) { ct . lock ( ) . await . decrement ( ) ; } } ) ; Ok ( cleanup_task ) }
Processes a new connection, performing a handshake, and then spawning two tasks to handle
input and output, returning join handles for the input and output tasks respectively
Processes a new connection performing a handshake and then spawning two tasks to handle input and output returning join handles for the input and output tasks respectively
transport
Transport<T, U>
conn_id
usize
state
Arc<Mutex<State>>
tracker
Option<Arc<Mutex<ConnTracker>>>
max_msg_capacity
usize