Real-Time Chat Application

Example Code

```rust
// This example uses `warp` for the web server and WebSockets,
// `tokio` for the asynchronous runtime, and `mpsc` channels for inter-task communication.
// It demonstrates a simple chat server that broadcasts messages from one client to all others.

// To run this code, add the following to your Cargo.toml:
// [dependencies]
// tokio = { version = "1", features = ["full"] }
// warp = "0.3"
// futures = "0.3"
// serde = { version = "1", features = ["derive"] }
// serde_json = "1.0"
// pretty_env_logger = "0.4"

use tokio::sync::mpsc;
use tokio_stream::wrappers::UnboundedReceiverStream;
use futures::{FutureExt, StreamExt};
use warp::ws::{Message, WebSocket};
use warp::Filter;
use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use tokio::task::JoinHandle;
use serde::{Deserialize, Serialize};

// Shared state for all connected users
// Each user has an mpsc::UnboundedSender to send messages to their WebSocket
type Users = Arc<Mutex<HashMap<usize, mpsc::UnboundedSender<Message>>>>;

// Atomic counter for assigning unique user IDs
static NEXT_USER_ID: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(1);

// Struct to represent a chat message, used for serialization/deserialization
#[derive(Debug, Serialize, Deserialize)]
struct ChatMessage {
    user_id: usize,
    text: String,
}

#[tokio::main]
async fn main() {
    // Initialize logging for better output (optional)
    pretty_env_logger::init();

    // Create a shared state for managing connected users
    let users = Users::default();
    let users_filter = warp::any().map(move || users.clone());

    // Define the WebSocket route
    // Access at ws://127.0.0.1:8000/ws
    let chat_route = warp::path("ws")
        .and(warp::ws())
        .and(users_filter)
        .map(|ws: warp::ws::Ws, users| {
            // This closure is executed for each new WebSocket connection upgrade request
            ws.on_upgrade(move |websocket| user_connected(websocket, users))
        });

    println!("Listening for WebSocket connections on 127.0.0.1:8000/ws");
    // Start the Warp server
    warp::serve(chat_route).run(([127, 0, 0, 1], 8000)).await;
}

// Handler for when a new user connects via WebSocket
async fn user_connected(ws: WebSocket, users: Users) {
    // Assign a unique ID to the new user
    let my_id = NEXT_USER_ID.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
    eprintln!("User {} connected", my_id);

    // Split the WebSocket into a sender (ws_tx) and a receiver (ws_rx)
    // ws_tx is used to send messages *to* this client.
    // ws_rx is used to receive messages *from* this client.
    let (ws_tx, mut ws_rx) = ws.split();

    // Create an unbounded MPSC (multi-producer, single-consumer) channel.
    // `tx` (sender) will be stored in `users` to send messages to this client.
    // `rx` (receiver) will feed messages into the `ws_tx` for this client.
    let (tx, rx) = mpsc::unbounded_channel();
    let rx_stream = UnboundedReceiverStream::new(rx);

    // Store the sender (tx) in our shared `Users` HashMap, associated with `my_id`.
    // This allows other parts of the server (e.g., `user_message`) to send messages to this client.
    users.lock().unwrap().insert(my_id, tx.clone());

    // Spawn a task that continuously takes messages from the `rx_stream`
    // and forwards them to the client's WebSocket sender (`ws_tx`).
    // This task runs in the background for the lifetime of the connection.
    let websocket_sender_task: JoinHandle<()> = tokio::task::spawn(rx_stream.forward(ws_tx).map(|result| {
        if let Err(e) = result {
            eprintln!("WebSocket send error for user {}: {}", my_id, e);
        }
    }));

    // Loop to continuously receive messages from this client's WebSocket (`ws_rx`).
    while let Some(result) = ws_rx.next().await {
        let msg = match result {
            Ok(msg) => msg,
            Err(e) => {
                eprintln!("WebSocket receive error for user {}: {}", my_id, e);
                break; // Exit the loop on error, indicating a disconnection
            }
        };
        // Process the received message
        user_message(my_id, msg, &users).await;
    }

    // If the loop exits, it means the client has disconnected or an error occurred.
    user_disconnected(my_id, &users);
    // Abort the sender task as the client is no longer connected.
    websocket_sender_task.abort();
}

// Handler for when a user sends a message
async fn user_message(my_id: usize, msg: Message, users: &Users) {
    if msg.is_text() {
        let text_msg_str = match msg.to_str() {
            Ok(s) => s,
            Err(_) => {
                eprintln!("User {} sent non-UTF8 text message, ignoring.", my_id);
                return; // Ignore non-UTF8 messages
            }
        };

        eprintln!("User {} sent text: {}", my_id, text_msg_str);

        // Create a structured ChatMessage to broadcast
        let chat_msg = ChatMessage {
            user_id: my_id,
            text: text_msg_str.to_string(),
        };
        let serialized_msg = match serde_json::to_string(&chat_msg) {
            Ok(s) => s,
            Err(e) => {
                eprintln!("Failed to serialize chat message for user {}: {}", my_id, e);
                return;
            }
        };

        // Broadcast the serialized message to all other connected users
        // (or all users if you want the sender to also receive their own message)
        for (&uid, tx) in users.lock().unwrap().iter() {
            if uid != my_id { // Only send to other users in this example
                // Attempt to send the message. If it fails, the user likely disconnected.
                if let Err(_disconnected) = tx.send(Message::text(&serialized_msg)) {
                    eprintln!("Failed to send message to user {}. Receiver likely disconnected.", uid);
                }
            }
        }
    } else if msg.is_binary() {
        eprintln!("User {} sent binary data (not handled in this example).", my_id);
    } else if msg.is_close() {
        eprintln!("User {} explicitly sent a close frame.", my_id);
        // The `user_connected` loop will handle the actual disconnection cleanup.
    } else if msg.is_ping() {
        eprintln!("User {} sent a ping. Warp automatically handles pongs.", my_id);
    }
}

// Handler for when a user disconnects
fn user_disconnected(my_id: usize, users: &Users) {
    eprintln!("User {} disconnected", my_id);
    // Remove the user's sender from the shared state
    users.lock().unwrap().remove(&my_id);

    // Optionally, broadcast a "user left" message to remaining users
    let chat_msg = ChatMessage {
        user_id: my_id,
        text: "has left the chat.".to_string(),
    };
    let serialized_msg = match serde_json::to_string(&chat_msg) {
        Ok(s) => s,
        Err(e) => {
            eprintln!("Failed to serialize 'user left' message for user {}: {}", my_id, e);
            return;
        }
    };

    for (&uid, tx) in users.lock().unwrap().iter() {
        // Even if we fail to send, the `user_connected` loop will eventually clean up `uid`.
        if let Err(_disconnected) = tx.send(Message::text(&serialized_msg)) {
            eprintln!("Failed to send 'user left' message to user {}. Receiver likely disconnected.", uid);
        }
    }
}
```