Repository: seanmonstar/warp (language: rust)

Issue:
SSE Chat Example is incorrect
So the [SSE example](https://github.com/seanmonstar/warp/blob/master/examples/sse_chat.rs) has the following code:
```rust
    // Create channel to track disconnecting the receiver side of events.
    // This is little bit tricky.
    let (mut dtx, mut drx) = oneshot::channel::<()>();

    // When `drx` will dropped then `dtx` will be canceled.
    // We can track it to make sure when the user leaves chat.
    tokio::task::spawn(async move {
        dtx.closed().await;
        drx.close();
        user_disconnected(my_id, &users2);
    });
```
Which does not actually do anything - `dtx` and `drx` are both moved into the task, which runs and never completes. As a result, no clients are ever disconnected from the server (I tested this).

I think the SSE example needs major changes in order to allow clients to be disconnected, as far as I can tell from the docs.


Discussion / hints:
yeah, it also seems outdated regarding the `Mutex` use, can be switched to the `tokio::sync::Mutex` would you like to try and PR an update to it?

Fix patch:
diff --git a/examples/sse_chat.rs b/examples/sse_chat.rs
index 139defdc7..ffb3064ed 100644
--- a/examples/sse_chat.rs
+++ b/examples/sse_chat.rs
@@ -4,28 +4,9 @@ use std::sync::{
     atomic::{AtomicUsize, Ordering},
     Arc, Mutex,
 };
-use tokio::sync::{mpsc, oneshot};
+use tokio::sync::mpsc;
 use warp::{sse::ServerSentEvent, Filter};
 
-/// Our global unique user id counter.
-static NEXT_USER_ID: AtomicUsize = AtomicUsize::new(1);
-
-/// Message variants.
-enum Message {
-    UserId(usize),
-    Reply(String),
-}
-
-#[derive(Debug)]
-struct NotUtf8;
-impl warp::reject::Reject for NotUtf8 {}
-
-/// Our state of currently connected users.
-///
-/// - Key is their id
-/// - Value is a sender of `Message`
-type Users = Arc<Mutex<HashMap<usize, mpsc::UnboundedSender<Message>>>>;
-
 #[tokio::main]
 async fn main() {
     pretty_env_logger::init();
@@ -73,6 +54,26 @@ async fn main() {
     warp::serve(routes).run(([127, 0, 0, 1], 3030)).await;
 }
 
+/// Our global unique user id counter.
+static NEXT_USER_ID: AtomicUsize = AtomicUsize::new(1);
+
+/// Message variants.
+#[derive(Debug)]
+enum Message {
+    UserId(usize),
+    Reply(String),
+}
+
+#[derive(Debug)]
+struct NotUtf8;
+impl warp::reject::Reject for NotUtf8 {}
+
+/// Our state of currently connected users.
+///
+/// - Key is their id
+/// - Value is a sender of `Message`
+type Users = Arc<Mutex<HashMap<usize, mpsc::UnboundedSender<Message>>>>;
+
 fn user_connected(
     users: Users,
 ) -> impl Stream<Item = Result<impl ServerSentEvent + Send + 'static, warp::Error>> + Send + 'static
@@ -86,33 +87,13 @@ fn user_connected(
     // to the event source...
     let (tx, rx) = mpsc::unbounded_channel();
 
-    match tx.send(Message::UserId(my_id)) {
-        Ok(()) => (),
-        Err(_disconnected) => {
-            // The tx is disconnected, our `user_disconnected` code
-            // should be happening in another task, nothing more to
-            // do here.
-        }
-    }
-
-    // Make an extra clone of users list to give to our disconnection handler...
-    let users2 = users.clone();
+    tx.send(Message::UserId(my_id))
+        // rx is right above, so this cannot fail
+        .unwrap();
 
     // Save the sender in our list of connected users.
     users.lock().unwrap().insert(my_id, tx);
 
-    // Create channel to track disconnecting the receiver side of events.
-    // This is little bit tricky.
-    let (mut dtx, mut drx) = oneshot::channel::<()>();
-
-    // When `drx` will dropped then `dtx` will be canceled.
-    // We can track it to make sure when the user leaves chat.
-    tokio::task::spawn(async move {
-        dtx.closed().await;
-        drx.close();
-        user_disconnected(my_id, &users2);
-    });
-
     // Convert messages into Server-Sent Events and return resulting stream.
     rx.map(|msg| match msg {
         Message::UserId(my_id) => Ok((warp::sse::event("user"), warp::sse::data(my_id)).into_a()),
@@ -127,25 +108,15 @@ fn user_message(my_id: usize, msg: String, users: &Users) {
     //
     // We use `retain` instead of a for loop so that we can reap any user that
     // appears to have disconnected.
-    for (&uid, tx) in users.lock().unwrap().iter_mut() {
-        if my_id != uid {
-            match tx.send(Message::Reply(new_msg.clone())) {
-                Ok(()) => (),
-                Err(_disconnected) => {
-                    // The tx is disconnected, our `user_disconnected` code
-                    // should be happening in another task, nothing more to
-                    // do here.
-                }
-            }
+    users.lock().unwrap().retain(|uid, tx| {
+        if my_id == *uid {
+            // don't send to same user, but do retain
+            true
+        } else {
+            // If not `is_ok`, the SSE stream is gone, and so don't retain
+            tx.send(Message::Reply(new_msg.clone())).is_ok()
         }
-    }
-}
-
-fn user_disconnected(my_id: usize, users: &Users) {
-    eprintln!("good bye user: {}", my_id);
-
-    // Stream closed up, so remove from the user list
-    users.lock().unwrap().remove(&my_id);
+    });
 }
 
 static INDEX_HTML: &str = r#"


Test patch: