Question: Multiple concurrent writes to a SplitSink

[JuxhinDB]

Following from the websocket chat example provided, we are able to receive messages and send messages independent of each other. My currently use-case would need to take this a step further, where multiple tasks would need to send messages through the same stream.

async fn handle_socket(socket: WebSocket, state: Arc<State>) {
    // Split our sender and receiver
    let (mut sender,  mut recv) = socket.split();

    while let Some(Ok(msg)) = recv.next().await {
        match msg {
            Message::Text(msg) => {
                println!("client sent str: {:?}", &msg);

                loop {
                    tokio::spawn(async move {
                        // Some IO-bound processing
                        if sender
                            .send(Message::Text("Result!")).collect())
                            .is_err()
                            {
                                println!("client disconnected");
                                return;
                            }
                    });
                }
            }
            // Other message variants
        }
    }
}

This will fail as sender is moved from the previous loop iteration, I'm not sure as to how you can clone multiple transmitters per-task from a single sender SplitSink. Is this a common pattern or should another pattern be used entirely (i.e. channels)?

[davidpdrsn]

I'm not sure actually. It could probably work to spawn a task that holds the sender and send messages to that via an mpsc channel. Then the task would receive the messages and forward them to the real sender. Then you'd only be cloning the mpsc sender half.

However I'm not sure if there is a better way. I recommend asking in #tokio-users in the tokio discord.

[JuxhinDB]

Thanks @davidpdrsn -- infact that is similar to what I had created in an old websocket example - https://github.com/JuxhinDB/twistrs/blob/master/examples/twistrs-ws/src/main.rs

So likely will try to re-create this example using axum. Will keep you updated, if you find it valuable I can look into contributing an example to the repository.

[davidpdrsn]

I'm not sure we need an example for it 🤔 Its more about futures/streams/async rust in general than axum and we already have two websocket examples.

[JuxhinDB]

Good point. Will likely write a small piece on it independently. Thanks for the help @davidpdrsn!

[davidpdrsn]

np (:

[imokoi]

hello, I got another way to get multiple senders of SplitSink, here is the example.

        let (tx, mut rx) = stream.split();
// we can define a Arc<Mutext<SplitSink>>
        let arc_tx = Arc::new(Mutex::new(tx));
        let tx1 = arc_tx.clone();
        tokio::spawn(async move {
            loop {
                thread::sleep(Duration::from_secs(2));
                let message = Message::Ping("ping".into());
                let mut sender = tx1.lock().await;
                match sender.send(message).await {
                    Ok(_) => {
                        println!("sent ping");
                    }
                    Err(e) => {
                        println!("send error: {}", e);
                        break;
                    }
                };
            }
        });

        let tx2 = arc_tx.clone();
        while let Some(Ok(message)) = rx.next().await {
            let mut sender = tx2.lock().await;
            // handle messages from the client
            match message {
                Message::Text(text) => {
                    let mut user_set = app_state.user_set.lock().await;
                    user_set.insert(text.clone());
                    let message = Message::Text(text.into());
                    sender.send(message).await.unwrap();
                }
                Message::Pong(_) => {
                    println!("received pong");
                }
                Message::Close(_) => {
                    println!("Websocket connection closed");
                    break;
                }
                _ => {}
            }
        }

[physics515]

I just found this and it's important to point out the use tokio::sync::Mutex;. I was using a std::sync::Mutex; and just caught on the the .await on yours.

[hexcowboy]

I've created a working example here, since the Arc/Mutex method was still giving me errors about thread safety.

https://gist.github.com/hexcowboy/8ebcf13a5d3b681aa6c684ad51dd6e0c