how to hand tcp request concurrently

[luxl]

My problem is when a TCP connection is established, I can get a frame from the connection and do some job. But the job could be time-consuming and when the job is finished, some response is needed to be written back to the connection.
When some specific requests were received(which will start a background job and keep sending message back every second), how can I achieve this?
Below is the example(mostly inspired by tokio example: tiny-http):

   async fn run() -> Result<(), Box<dyn Error>> {
        loop{
            let (mut conn, _) = listen.accept().await.unwrap();
            let s = state.clone();
            tokio::spawn(async move {
                if let Err(e) = process(conn, p).await {
                    println!("Failed to process connection: {}", e);
                }
            });
        }
    }

    async fn process(stream: TcpStream, pet: Arc<Mutex<State>>) -> Result<(), Box<dyn Error>> {
        let mut conn = Framed::new(stream, PetCodec);

        while let Some(frm) = conn.next().await {
            match frm {
                Ok(frm) => {
                    // how to handle 'frm' concurrently here?
                    // 'handle_request' is a time-consuming fn
                    let response = handle_request(frm, pet.clone()).await;
                    conn.send(response).await?;
                },
                Err(e) => {
                    return Err(e.into())
                },
            }
        }

        Ok(())
    }

if conn could be cloned and then I can use tokio::spawn to do some job and then write response to the conn, but it can't.
Is there any ideas? I'm new to Rust and struggling with it.

[Darksonn]

There are several options, but here is one:

async fn process(stream: TcpStream, pet: Arc<Mutex<State>>) -> Result<(), Box<dyn Error>> {
    let (write, read) = Framed::new(stream, PetCodec).split(); // futures::stream::StreamExt
    let (send, recv) = tokio::sync::mpsc::channel(16);

    tokio::try_join!(
        async move {
            while let Some(frm) = read.next().await {
                match frm {
                    Ok(frm) => {
                        let send = send.clone();
                        let pet = pet.clone();
                        tokio::spawn(async move {
                            let response = handle_request(frm, pet).await;
                            let _ = send.send(response).await;
                        });
                    }
                    Err(e) => return Err(e.into()),
                }
            }
            drop(send);

            // this type annotation is necessary to help out the compiler
            Result::<(), Box<dyn Error>>::Ok(())
        },
        async move {
            while let Some(response) = recv.recv().await {
                write.send(response)?;
            }

            Ok(())
        }
    )?;
    
    Ok(())
}

[Darksonn]

This is due to your use of Box<dyn Error>, which cannot be used anywhere inside tokio::spawn because it is not Send. Change it to Box<dyn Error + Send + Sync> or use a library like anyhow.

[luxl]

oh, thanks again, Rust is really hard to learn at the beginning...

[luxl]

Hi, would you like to add this kind of pattern to the tokio-example in case newcomers would need it, I think it's comman pattern in real-world

[Darksonn]

Sure I don't mind. Generally I've been wanting to make a section with a lot of these examples on our website, but I have not had the time.

[zuston]

Thanks for your example @Darksonn

But I doubt the sequence problem, is it possible that sequence of multiple write.send(response)? is not consistent with the request sequence on the same connection.