I am having trouble with tokio::sync::mpsc::UnboundedSender

[TheDarkUndoing]

Source:

use log::{info,error,warn};
use futures_util::{SinkExt, StreamExt, TryFutureExt};
use prost::Message;
use tinytown::networking::{Client, Invite, Sender};

use tokio::sync::{mpsc, RwLock};
use tokio_stream::wrappers::UnboundedReceiverStream;
use warp::ws::{Message as WS_Message, WebSocket};
use warp::Filter;

use std::sync::atomic::{AtomicUsize, Ordering};
use std::{collections::HashMap, sync::Arc};
use tinytown::networking::sender::Msg;
use uuid::Uuid;

type Users = Arc<RwLock<HashMap<usize, mpsc::UnboundedSender<WS_Message>>>>;
type Lobbies = Arc<RwLock<HashMap<String, Vec<mpsc::UnboundedSender<WS_Message>>>>>;
type Games = Arc<RwLock<HashMap<String, Vec<game::GameState>>>>;

async fn get_tx(user_db: &Users, id: &usize) -> mpsc::UnboundedSender<WS_Message> {
    user_db
        .read()
        .await
        .get(&id)
        .expect("No such transactions found for that user. is it a valid user?")
        .clone()
        

}

static NEXT_USER_ID: AtomicUsize = AtomicUsize::new(1);

pub async fn start() {

    let users = Users::default();
    let users = warp::any().map(move || users.clone());

    let lobbies = Lobbies::default();
    let lobbies = warp::any().map(move || lobbies.clone());

    let games = Games::default();
    let games = warp::any().map(move || games.clone());

    let routes = warp::path("connect")
        // The `ws()` filter will prepare the Websocket handshake.
        .and(warp::ws())
        .and(users)
        .and(lobbies)
        .and(games)
        .map(|ws: warp::ws::Ws, users, lobbies, games| {
            // And then our closure will be called when it completes...
            ws.on_upgrade(|websocket| user_connected(websocket, users, lobbies))
        });

    // info!("Started on 127.0.0.1:30303");
    warp::serve(routes).run(([127, 0, 0, 1], 3030)).await;
}

async fn user_connected(ws: WebSocket, users: Users, lobbies: Lobbies) {
    // Use a counter to assign a new unique ID for this user.
    let my_id = NEXT_USER_ID.fetch_add(1, Ordering::Relaxed);

    info!("new chat user: {}", my_id);

    // Split the socket into a sender and receive of messages.
    let (mut user_ws_tx, mut user_ws_rx) = ws.split();

    // Use an unbounded channel to handle buffering and flushing of messages
    // to the websocket...
    let (tx, rx) = mpsc::unbounded_channel();
    let mut rx = UnboundedReceiverStream::new(rx);

    // let mut tx = UnboundedSender::new(tx);

    // Closure for When user sends a message//
    tokio::task::spawn(async move {
        while let Some(message) = rx.next().await {
            info!("{:?}", message);
            //Send User its client ID

            user_ws_tx
                .send(message)
                .unwrap_or_else(|e| {
                    error!("websocket send error: {}", e);
                })
                .await;
        }
        info!("Ran out of rx")
    });
    let id_msg = Client::new(my_id.to_string());
    let wrapper_msg = Sender::new(Msg::Client(id_msg));

    tx.send(WS_Message::binary(wrapper_msg.encode_to_vec())).expect("Failed to send ");
    // Save the sender in our list of connected users.
    users.write().await.insert(my_id, tx);

    // Every time the user do...
    while let Some(result) = user_ws_rx.next().await {
        let msg = match result {
            Ok(msg) => msg,
            Err(e) => {
                error!("websocket error(uid={}): {}", my_id, e);
                break;
            }
        };
        // If Client sends no bytes then move on and disconnect client
        if msg.as_bytes().len() == 0 {
            continue;
        }
        // println!("message bytes from {}: {:?}", my_id, msg.as_bytes());
        let message = msg.to_str().expect("could not convert to str");
        info!("{} says {}", my_id, &message);
        user_message(my_id, msg.clone(), &users).await;
        // print
        let command = message.split(" ").collect::<Vec<&str>>();
        match command[0] {
            "play" => {
                info!("{} wants to PLAY the game", my_id);
                matchmaking(my_id, msg.clone(), &users, &lobbies).await;
            }
            "approve" => {
                info!("{} wants to Approve the join request", my_id);
                approve(my_id, msg.clone(), &users, &lobbies).await;
            }
            &_ => error!("Invalid Command: {}",command[0]),
        }
        //Print State of lobbies
        info!("Lobbies: {:?}", lobbies.read().await);
    }

    // user_ws_rx stream will keep processing as long as the user stays
    // connected. Once they disconnect, then...
    user_disconnected(my_id, &users).await;
}

async fn user_message(my_id: usize, msg: WS_Message, users: &Users) {
    // Skip any non-Text messages...
    let msg = if let Ok(s) = msg.to_str() {
        s
    } else {
        return;
    };

    let new_msg = format!("<User#{}>: {}", &my_id, msg);

    let db = users.read().await;
    let tx = db.get(&my_id);
    match tx
        .expect("failed to get sender")
        .send(WS_Message::text(new_msg.clone()))
    {
        Ok(a) => a,
        Err(e) => eprintln!("{:?}", e),
    };
    // tx.send(Message::text(new_msg.clone()));
    drop(db);
}
async fn user_disconnected(my_id: usize, users: &Users) {
    info!(" user: {} disconnected", my_id);

    // Stream closed up, so remove from the user list
    users.write().await.remove(&my_id);
}

async fn matchmaking(my_id: usize, msg: WS_Message, users: &Users, lobbies: &Lobbies) {
    let command = msg.to_str().unwrap().split(" ").collect::<Vec<&str>>();

    let lobby_id = Uuid::new_v4();
    //Create Invite Protobuf message
    let invite = Invite::new(lobby_id.to_string(), my_id.to_string());

    let invite_msg = Sender::new(Msg::Invite(invite));

    let other_user: usize;
    if command.len() > 1 {
        let parse = command[1].parse::<usize>();
        match parse {
            Ok(a) => {
                other_user = a;
            }
            Err(e) => {
                error!("{}", e);
                return;
            }
        }
    } else {
        return;
    }

    info!("Matchmaking for {}", my_id);
    info!(
        "{} invites {} to lobby ID: {}",
        my_id,
        other_user,
        lobby_id.to_string()
    );

    // Notify other user of matchmaking
    let new_msg = format!("<User#{}> wants to play with you!", &my_id);
    let game_array: Vec<mpsc::UnboundedSender<WS_Message>>;
    let my_tx = get_tx(users, &my_id).await;
    // let invitee_tx = get_tx(users, &other_user).await;

    game_array = vec![my_tx];
    // Add initiating user to lobby
    lobbies
        .write()
        .await
        .insert(lobby_id.to_string(), game_array);
    // let lobby:String = format("",);

    let invitee_tx = get_tx(users, &other_user).await;
   
    invitee_tx.send(WS_Message::text(&new_msg)).unwrap();
    invitee_tx.send(WS_Message::binary(invite_msg.encode_to_vec())).unwrap();

    
    
}

async fn approve(my_id: usize, msg: WS_Message, users: &Users, lobbies: &Lobbies) {
    let command = msg.to_str().unwrap().split(" ").collect::<Vec<&str>>();
    // let mut g_lobby_id = String::new();
    let my_tx = get_tx(users, &my_id).await;
    if command.len() > 1 {
        let parse = command[1].parse::<String>();
        match parse {
            Ok(lobby_id) => {
                info!("Awaiting Lobby read...");
                let temp_lobbies = lobbies.read().await;
                
                info!("Looking for specific Lobby...");
                let client_connection_array = temp_lobbies.get(lobby_id.as_str());

                match client_connection_array {
                    Some(valid_array) => {
                        info!("Cloning Lobby...");
                        let mut cloned_array = valid_array.clone();
                        info!("Adding this user to Lobby");
                        cloned_array.push(my_tx);
                        
                        info!("{:#?}",cloned_array);
                        lobbies
                            .write()
                            .await
                            .insert(lobby_id.to_string(), cloned_array);
                        info!("Joining lobby id {}", lobby_id);
                    },
                    None => error!("No such Lobby in existance for {}",lobby_id),
                }
                drop(temp_lobbies);
               
               
           
            }
            Err(e) => {
                error!("{}", e);
                return;
            }
        }
        // return ;
    } else {
        return;
    }
}

Expected Behavior:
Once connected, a client should be able to approve an invite to a lobby and continue receiving messages

Current Behavior:
After approving an invite, the invitee ans the inviter receive one more message in total from the server and then communication halts between the two and the server.

At first I thought it was the RWLocks were not being handled correctly but I don't think that is the issue. My current theory is that when cloning an UnboundedSender it messes with the UnboundedReciever or changes the state of the UnboundedSender in the users Hashmap. Any help on this would be greatly appreciated. I've been banging my head against this for a few weeks now.

[Darksonn]

In general, putting your channels in locks like that is generally not the best way to approach it. Consider letting the channels be owned by a single task via the actor pattern, see also this example.

If you need further help, please post the question on URLO. A lot more people view that forum than answer here.