how can i use a channel in tokio::select!

[xiashaohua]

I'm learning about Rust, and I tried to use a select! to handle both TCP Stream and mpsc ::Receiver .But when I add the Receiver's consumption function, the select! doesn't work at all. Why, or how, should I consume the Receiver in a select!.

the code is:

    let mut transport =Framed::new(stream, BytesCodec::new());

    let (message_sender, message_receiver) = mpsc::channel();
    let receiver = Arc::new(Mutex::new(message_receiver));

    loop {
        println!("into loop");

        tokio::select! {
            Some(data) = transport.next() => {
                println!("into stream");
                match data {
                    Ok(request) => {
                        println!("bytes: {:?}", request[27]);
                        let tmp = mpsc::Sender::clone(&message_sender);
                        select_task_by_msg(request, tmp).await;
            
                    }
                    Err(e) => return Err(e.into()),
                
                }
            },
            res = receive_write_data(Arc::clone(&receiver)) => {
                println!("into ch");
                match res {
                    Some(v) => {
                        let response = create_message_by_struct(v);
                        transport.send(Bytes::copy_from_slice(&response.as_bytes())).await?;

                    }
                    None => {}
                }
                
            },
            else =>{
                println!("not match");
                break;
            }

        }

    }

receive_write_data is:

pub async fn receive_write_data(receiver: Arc<Mutex<mpsc::Receiver<CascadeMessage>>>) -> Option<CascadeMessage> {
    let recv = receiver.lock();
    match recv {
        Ok(V) => {
            let data = V.recv();
            match data {
                Ok(real) => Some(real),
                Err(_) => None
            }
        }
        Err(_) => None
    }

}

[Darksonn]

This is because you are blocking the thread by using a channel whose recv method does not involve an .await. You should be using tokio::sync::mpsc instead, which is a channel designed for async code. For more info, see the channels chapter in the Tokio tutorial.

You can receive on it like this:

use tokio::sync::mpsc;

let mut transport = Framed::new(stream, BytesCodec::new());

let (message_sender, message_receiver) = mpsc::channel();

loop {
    println!("into loop");

    tokio::select! {
        Some(data) = transport.next() => {
            ...
        },
        Some(res) = message_receiver.recv() => {
            let response = create_message_by_struct(v);
            transport.send(Bytes::copy_from_slice(&response.as_bytes())).await?;
        },
        else => {
            println!("channel and connection both closed");
            break;
        }
    }
}

Additionally, please note that it pretty much never makes sense to put a channel receiver into a Mutex.

[xiashaohua]

thanks a lot. That solved my problem successfully