Add user presence support to the server

rustpad-server/src/rustpad.rs

@@ -1,5 +1,6 @@
 //! Eventually consistent server-side logic for Rustpad.
 
+use std::collections::HashMap;
 use std::sync::atomic::{AtomicU64, Ordering};
 use std::time::Duration;
 
@@ -31,6 +32,7 @@ struct State {
     operations: Vec<UserOperation>,
     text: String,
     language: Option<String>,
+    users: HashMap<u64, UserInfo>,
 }
 
 #[derive(Clone, Debug, Serialize, Deserialize)]
@@ -39,6 +41,12 @@ struct UserOperation {
     operation: OperationSeq,
 }
 
+#[derive(Clone, Debug, Serialize, Deserialize)]
+struct UserInfo {
+    name: String,
+    hue: u32,
+}
+
 /// A message received from the client over WebSocket.
 #[derive(Clone, Debug, Serialize, Deserialize)]
 enum ClientMsg {
@@ -47,8 +55,10 @@ enum ClientMsg {
         revision: usize,
         operation: OperationSeq,
     },
-    /// Set the language of the editor.
+    /// Sets the language of the editor.
     SetLanguage(String),
+    /// Sets the user's current information.
+    ClientInfo(UserInfo),
 }
 
 /// A message sent to the client over WebSocket.
@@ -63,6 +73,8 @@ enum ServerMsg {
     },
     /// Broadcasts the current language, last writer wins.
     Language(String),
+    /// Broadcasts a user's information, or `None` on disconnect.
+    UserInfo { id: u64, info: Option<UserInfo> },
 }
 
 impl From<ServerMsg> for Message {
@@ -93,6 +105,10 @@ impl Rustpad {
             warn!("connection terminated early: {}", e);
         }
         info!("disconnection, id = {}", id);
+        self.state.write().users.remove(&id);
+        self.update
+            .send(ServerMsg::UserInfo { id, info: None })
+            .ok();
     }
 
     /// Returns a snapshot of the latest text.
@@ -148,6 +164,12 @@ impl Rustpad {
             if let Some(language) = &state.language {
                 messages.push(ServerMsg::Language(language.clone()));
             }
+            for (&id, info) in &state.users {
+                messages.push(ServerMsg::UserInfo {
+                    id,
+                    info: Some(info.clone()),
+                });
+            }
         };
         for msg in messages {
             socket.send(msg.into()).await?;
@@ -191,6 +213,14 @@ impl Rustpad {
                 self.state.write().language = Some(language.clone());
                 self.update.send(ServerMsg::Language(language)).ok();
             }
+            ClientMsg::ClientInfo(info) => {
+                self.state.write().users.insert(id, info.clone());
+                let msg = ServerMsg::UserInfo {
+                    id,
+                    info: Some(info),
+                };
+                self.update.send(msg).ok();
+            }
         }
         Ok(())
     }

rustpad-server/tests/users.rs

@@ -0,0 +1,112 @@
+//! Tests for synchronization of user presence.
+
+use anyhow::Result;
+use common::*;
+use rustpad_server::server;
+use serde_json::json;
+
+pub mod common;
+
+#[tokio::test]
+async fn test_two_users() -> Result<()> {
+    pretty_env_logger::try_init().ok();
+    let filter = server();
+
+    let mut client = connect(&filter, "foobar").await?;
+    assert_eq!(client.recv().await?, json!({ "Identity": 0 }));
+
+    let alice = json!({
+        "name": "Alice",
+        "hue": 42
+    });
+    client.send(&json!({ "ClientInfo": alice })).await;
+
+    let alice_info = json!({
+        "UserInfo": {
+            "id": 0,
+            "info": alice
+        }
+    });
+    assert_eq!(client.recv().await?, alice_info);
+
+    let mut client2 = connect(&filter, "foobar").await?;
+    assert_eq!(client2.recv().await?, json!({ "Identity": 1 }));
+    assert_eq!(client2.recv().await?, alice_info);
+
+    let bob = json!({
+        "name": "Bob",
+        "hue": 96
+    });
+    client2.send(&json!({ "ClientInfo": bob })).await;
+
+    let bob_info = json!({
+        "UserInfo": {
+            "id": 1,
+            "info": bob
+        }
+    });
+    assert_eq!(client2.recv().await?, bob_info);
+    assert_eq!(client.recv().await?, bob_info);
+
+    Ok(())
+}
+
+#[tokio::test]
+async fn test_invalid_user() -> Result<()> {
+    pretty_env_logger::try_init().ok();
+    let filter = server();
+
+    let mut client = connect(&filter, "foobar").await?;
+    assert_eq!(client.recv().await?, json!({ "Identity": 0 }));
+
+    let alice = json!({ "name": "Alice" }); // no hue
+    client.send(&json!({ "ClientInfo": alice })).await;
+    client.recv_closed().await?;
+
+    Ok(())
+}
+
+#[tokio::test]
+async fn test_leave_rejoin() -> Result<()> {
+    pretty_env_logger::try_init().ok();
+    let filter = server();
+
+    let mut client = connect(&filter, "foobar").await?;
+    assert_eq!(client.recv().await?, json!({ "Identity": 0 }));
+
+    let alice = json!({
+        "name": "Alice",
+        "hue": 42
+    });
+    client.send(&json!({ "ClientInfo": alice })).await;
+
+    let alice_info = json!({
+        "UserInfo": {
+            "id": 0,
+            "info": alice
+        }
+    });
+    assert_eq!(client.recv().await?, alice_info);
+
+    client.send(&json!({ "Invalid": "please close" })).await;
+    client.recv_closed().await?;
+
+    let mut client2 = connect(&filter, "foobar").await?;
+    assert_eq!(client2.recv().await?, json!({ "Identity": 1 }));
+
+    let bob = json!({
+        "name": "Bob",
+        "hue": 96
+    });
+    client2.send(&json!({ "ClientInfo": bob })).await;
+
+    let bob_info = json!({
+        "UserInfo": {
+            "id": 1,
+            "info": bob
+        }
+    });
+    assert_eq!(client2.recv().await?, bob_info);
+
+    Ok(())
+}