fn user_connected(ws: WebSocket, users: Users) -> impl Future<Item = (), Error = ()> {
    // 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 websocket user: {}", my_id);

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

    // Use an unbounded channel to handle buffering and flushing of messages
    // to the websocket...
    let (tx, rx) = mpsc::unbounded();
    warp::spawn(
        rx.map_err(|()| -> warp::Error { unreachable!("unbounded rx never errors") })
            .forward(user_ws_tx)
            .map(|_tx_rx| ())
            .map_err(|ws_err| info!("websocket send error: {}", ws_err)),
    );

    // Save the sender in our list of connected users.
    users.lock().expect("error uses.lock()").insert(my_id, tx);

    // Return a `Future` that is basically a state machine managing
    // this specific user's connection.
    // Make an extra clone to give to our disconnection handler...
    //let users2 = users.clone();
    //Clippy reccomands this crazyness insted of users.clone()
    let users2 = Arc::<
        std::sync::Mutex<
            std::collections::HashMap<
                usize,
                futures::sync::mpsc::UnboundedSender<warp::ws::Message>,
            >,
        >,
    >::clone(&users);

    user_ws_rx
        // Every time the user sends a message, broadcast it to
        // all other users...
        .for_each(move |msg| {
            user_message(my_id, &msg, &users);
            Ok(())
        })
        // for_each will keep processing as long as the user stays
        // connected. Once they disconnect, then...
        .then(move |result| {
            user_disconnected(my_id, &users2);
            result
        })
        // If at any time, there was a websocket error, log here...
        .map_err(move |e| {
            info!("websocket error(uid={}): {}", my_id, e);
        })
}
//region: websocket callbacks: connect, msg, disconnect
///new user connects
websocket callbacks: connect, msg, disconnect
new user connects
websocket callbacks : connect msg disconnect new user connects
fn user_connected(ws: WebSocket, users: Users) -> impl Future<Item = (), Error = ()> {
    // 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 websocket user: {}", my_id);
    // Split the socket into a sender and receive of messages.
    let (user_ws_tx, user_ws_rx) = ws.split();
    // Use an unbounded channel to handle buffering and flushing of messages
    // to the websocket...
    let (tx, rx) = mpsc::unbounded();
    warp::spawn(
        rx.map_err(|()| -> warp::Error { unreachable!("unbounded rx never errors") })
            .forward(user_ws_tx)
            .map(|_tx_rx| ())
            .map_err(|ws_err| info!("websocket send error: {}", ws_err)),
    );
    // Save the sender in our list of connected users.
    users.lock().expect("error uses.lock()").insert(my_id, tx);
    // Return a `Future` that is basically a state machine managing
    // this specific user's connection.
    // Make an extra clone to give to our disconnection handler...
    //let users2 = users.clone();
    //Clippy reccomands this crazyness insted of users.clone()
    let users2 = Arc::<
        std::sync::Mutex<
            std::collections::HashMap<
                usize,
                futures::sync::mpsc::UnboundedSender<warp::ws::Message>,
            >,
        >,
    >::clone(&users);
    user_ws_rx
        // Every time the user sends a message, broadcast it to
        // all other users...
        .for_each(move |msg| {
            user_message(my_id, &msg, &users);
            Ok(())
        })
        // for_each will keep processing as long as the user stays
        // connected. Once they disconnect, then...
        .then(move |result| {
            user_disconnected(my_id, &users2);
            result
        })
        // If at any time, there was a websocket error, log here...
        .map_err(move |e| {
            info!("websocket error(uid={}): {}", my_id, e);
        })
}
fn user_connected ( ws : WebSocket , users : Users ) -> impl Future < Item = ( ) , Error = ( ) > { // Use a counter to assign a new unique ID for this user. let my_id = NEXT_USER_ID . fetch_add ( 1 , Ordering :: Relaxed ) ; info ! ( " " my_id ) ; // Split the socket into a sender and receive of messages. let ( user_ws_tx , user_ws_rx ) = ws . split ( ) ; // Use an unbounded channel to handle buffering and flushing of messages // to the websocket... let ( tx , rx ) = mpsc :: unbounded ( ) ; warp :: spawn ( rx . map_err ( | ( ) | -> warp :: Error { unreachable ! ( " " ) } ) . forward ( user_ws_tx ) . map ( | _tx_rx | ( ) ) . map_err ( | ws_err | info ! ( " " ws_err ) ) , ) ; // Save the sender in our list of connected users. users . lock ( ) . expect ( " " ) . insert ( my_id , tx ) ; // Return a `Future` that is basically a state machine managing // this specific user's connection. // Make an extra clone to give to our disconnection handler... //let users2 = users.clone(); //Clippy reccomands this crazyness insted of users.clone() let users2 = Arc :: < std :: sync :: Mutex < std :: collections :: HashMap < usize , futures :: sync :: mpsc :: UnboundedSender < warp :: ws :: Message > , > , > , > :: clone ( & users ) ; user_ws_rx // Every time the user sends a message, broadcast it to // all other users... . for_each ( move | msg | { user_message ( my_id , & msg , & users ) ; Ok ( ( ) ) } ) // for_each will keep processing as long as the user stays // connected. Once they disconnect, then... . then ( move | result | { user_disconnected ( my_id , & users2 ) ; result } ) // If at any time, there was a websocket error, log here... . map_err ( move | e | { info ! ( " " my_id e ) ; } ) }
region: websocket callbacks: connect, msg, disconnect
new user connects
region : websocket callbacks : connect msg disconnect new user connects
ws
WebSocket
users
Users