Merge pull request #293 from actix/fix/chat-broker

websocket chat broker improvements

.gitignore

@@ -16,5 +16,8 @@ Cargo.lock
 
 .history/
 
+# VS Code workspace config
+.vscode
+
 # For multipart example
 upload.png

Cargo.toml

@@ -47,6 +47,6 @@ members = [
    "web-cors/backend",
    "websocket",
    "websocket-chat",
-#   "websocket-chat-broker",
+   "websocket-chat-broker",
    "websocket-tcp-chat",
 ]

websocket-chat-broker/Cargo.toml

@@ -1,7 +1,10 @@
 [package]
-name = "websocket-broker-example"
+name = "websocket-chat-broker"
 version = "0.1.0"
-authors = ["Chris Ricketts <chris.ricketts@steribar.com>"]
+authors = [
+  "Chris Ricketts <chris.ricketts@steribar.com>",
+  "Rob Ede <robjtede@icloud.com>",
+]
 edition = "2018"
 
 [[bin]]
@@ -9,12 +12,13 @@ name = "server"
 path = "src/main.rs"
 
 [dependencies]
-rand = "0.6"
+actix = "0.9"
-futures = "0.1.24"
+actix-broker = "0.3.0"
-actix = "0.8.2"
+actix-files = "0.2"
-actix-web = "1.0"
+actix-rt = "1.1"
-actix-files = "0.1"
+actix-web = "2.0"
-actix-web-actors = "1.0"
+actix-web-actors = "2.0"
-actix-broker = "0.2.0"
+env_logger = "0.7"
-log = "0.4.5"
+futures = "0.3"
-simple_logger = "0.5.0"
+log = "0.4"
+rand = "0.7"

websocket-chat-broker/src/main.rs

@@ -1,159 +1,36 @@
-#[macro_use]
+use log::info;
-extern crate log;
 
-use actix::fut;
-use actix::prelude::*;
-use actix_broker::BrokerIssue;
 use actix_files::Files;
 use actix_web::{web, App, Error, HttpRequest, HttpResponse, HttpServer};
 use actix_web_actors::ws;
 
+mod message;
 mod server;
-use server::*;
+mod session;
 
-fn chat_route(req: HttpRequest, stream: web::Payload) -> Result<HttpResponse, Error> {
+use session::WsChatSession;
+
+async fn chat_route(
+    req: HttpRequest,
+    stream: web::Payload,
+) -> Result<HttpResponse, Error> {
     ws::start(WsChatSession::default(), &req, stream)
 }
 
-#[derive(Default)]
+#[actix_rt::main]
-struct WsChatSession {
+async fn main() -> std::io::Result<()> {
-    id: usize,
+    env_logger::from_env(env_logger::Env::default().default_filter_or("info")).init();
-    room: String,
-    name: Option<String>,
-}
 
-impl WsChatSession {
+    let addr = "127.0.0.1:8080";
-    fn join_room(&mut self, room_name: &str, ctx: &mut ws::WebsocketContext<Self>) {
-        let room_name = room_name.to_owned();
-        // First send a leave message for the current room
-        let leave_msg = LeaveRoom(self.room.clone(), self.id);
-        // issue_sync comes from having the `BrokerIssue` trait in scope.
-        self.issue_system_sync(leave_msg, ctx);
-        // Then send a join message for the new room
-        let join_msg = JoinRoom(
-            room_name.to_owned(),
-            self.name.clone(),
-            ctx.address().recipient(),
-        );
 
-        WsChatServer::from_registry()
+    let srv = HttpServer::new(move || {
-            .send(join_msg)
-            .into_actor(self)
-            .then(|id, act, _ctx| {
-                if let Ok(id) = id {
-                    act.id = id;
-                    act.room = room_name;
-                }
-
-                fut::ok(())
-            })
-            .spawn(ctx);
-    }
-
-    fn list_rooms(&mut self, ctx: &mut ws::WebsocketContext<Self>) {
-        WsChatServer::from_registry()
-            .send(ListRooms)
-            .into_actor(self)
-            .then(|res, _, ctx| {
-                if let Ok(rooms) = res {
-                    for room in rooms {
-                        ctx.text(room);
-                    }
-                }
-                fut::ok(())
-            })
-            .spawn(ctx);
-    }
-
-    fn send_msg(&self, msg: &str) {
-        let content = format!(
-            "{}: {}",
-            self.name.clone().unwrap_or_else(|| "anon".to_string()),
-            msg
-        );
-        let msg = SendMessage(self.room.clone(), self.id, content);
-        // issue_async comes from having the `BrokerIssue` trait in scope.
-        self.issue_system_async(msg);
-    }
-}
-
-impl Actor for WsChatSession {
-    type Context = ws::WebsocketContext<Self>;
-
-    fn started(&mut self, ctx: &mut Self::Context) {
-        self.join_room("Main", ctx);
-    }
-
-    fn stopped(&mut self, _ctx: &mut Self::Context) {
-        info!(
-            "WsChatSession closed for {}({}) in room {}",
-            self.name.clone().unwrap_or_else(|| "anon".to_string()),
-            self.id,
-            self.room
-        );
-    }
-}
-
-impl Handler<ChatMessage> for WsChatSession {
-    type Result = ();
-
-    fn handle(&mut self, msg: ChatMessage, ctx: &mut Self::Context) {
-        ctx.text(msg.0);
-    }
-}
-
-impl StreamHandler<ws::Message, ws::ProtocolError> for WsChatSession {
-    fn handle(&mut self, msg: ws::Message, ctx: &mut Self::Context) {
-        debug!("WEBSOCKET MESSAGE: {:?}", msg);
-        match msg {
-            ws::Message::Text(text) => {
-                let msg = text.trim();
-                if msg.starts_with('/') {
-                    let mut command = msg.splitn(2, ' ');
-                    match command.next() {
-                        Some("/list") => self.list_rooms(ctx),
-                        Some("/join") => {
-                            if let Some(room_name) = command.next() {
-                                self.join_room(room_name, ctx);
-                            } else {
-                                ctx.text("!!! room name is required");
-                            }
-                        }
-                        Some("/name") => {
-                            if let Some(name) = command.next() {
-                                self.name = Some(name.to_owned());
-                                ctx.text(format!("name changed to: {}", name));
-                            } else {
-                                ctx.text("!!! name is required");
-                            }
-                        }
-                        _ => ctx.text(format!("!!! unknown command: {:?}", msg)),
-                    }
-                    return;
-                }
-                self.send_msg(msg);
-            }
-            ws::Message::Close(_) => {
-                ctx.stop();
-            }
-            _ => {}
-        }
-    }
-}
-
-fn main() -> std::io::Result<()> {
-    let sys = actix::System::new("websocket-broker-example");
-    simple_logger::init_with_level(log::Level::Info).unwrap();
-
-    HttpServer::new(move || {
         App::new()
             .service(web::resource("/ws/").to(chat_route))
             .service(Files::new("/", "./static/").index_file("index.html"))
     })
-    .bind("127.0.0.1:8080")
+    .bind(&addr)?;
-    .unwrap()
-    .start();
 
-    info!("Started http server: 127.0.0.1:8080");
+    info!("Starting http server: {}", &addr);
-    sys.run()
+
+    srv.run().await
 }

websocket-chat-broker/src/message.rs

@@ -0,0 +1,21 @@
+use actix::prelude::*;
+
+#[derive(Clone, Message)]
+#[rtype(result = "()")]
+pub struct ChatMessage(pub String);
+
+#[derive(Clone, Message)]
+#[rtype(result = "usize")]
+pub struct JoinRoom(pub String, pub Option<String>, pub Recipient<ChatMessage>);
+
+#[derive(Clone, Message)]
+#[rtype(result = "()")]
+pub struct LeaveRoom(pub String, pub usize);
+
+#[derive(Clone, Message)]
+#[rtype(result = "Vec<String>")]
+pub struct ListRooms;
+
+#[derive(Clone, Message)]
+#[rtype(result = "()")]
+pub struct SendMessage(pub String, pub usize, pub String);

websocket-chat-broker/src/server.rs

@@ -1,29 +1,14 @@
 use actix::prelude::*;
 use actix_broker::BrokerSubscribe;
-use rand;
 
 use std::collections::HashMap;
 use std::mem;
 
-#[derive(Clone, Message)]
+use crate::message::{ChatMessage, JoinRoom, LeaveRoom, ListRooms, SendMessage};
-pub struct ChatMessage(pub String);
-
-#[derive(Clone, Message)]
-#[rtype(result = "usize")]
-pub struct JoinRoom(pub String, pub Option<String>, pub Recipient<ChatMessage>);
-
-#[derive(Clone, Message)]
-pub struct LeaveRoom(pub String, pub usize);
-
-#[derive(Clone, Message)]
-#[rtype(result = "Vec<String>")]
-pub struct ListRooms;
-
-#[derive(Clone, Message)]
-pub struct SendMessage(pub String, pub usize, pub String);
 
 type Client = Recipient<ChatMessage>;
 type Room = HashMap<usize, Client>;
+
 #[derive(Default)]
 pub struct WsChatServer {
     rooms: HashMap<String, Room>,
@@ -43,6 +28,7 @@ impl WsChatServer {
         client: Client,
     ) -> usize {
         let mut id = id.unwrap_or_else(rand::random::<usize>);
+
         if let Some(room) = self.rooms.get_mut(room_name) {
             loop {
                 if room.contains_key(&id) {
@@ -51,13 +37,17 @@ impl WsChatServer {
                     break;
                 }
             }
+
             room.insert(id, client);
             return id;
         }
+
         // Create a new room for the first client
         let mut room: Room = HashMap::new();
+
         room.insert(id, client);
         self.rooms.insert(room_name.to_owned(), room);
+
         id
     }
 
@@ -68,11 +58,13 @@ impl WsChatServer {
         _src: usize,
     ) -> Option<()> {
         let mut room = self.take_room(room_name)?;
+
         for (id, client) in room.drain() {
             if client.do_send(ChatMessage(msg.to_owned())).is_ok() {
                 self.add_client_to_room(room_name, Some(id), client);
             }
         }
+
         Some(())
     }
 }
@@ -91,12 +83,14 @@ impl Handler<JoinRoom> for WsChatServer {
 
     fn handle(&mut self, msg: JoinRoom, _ctx: &mut Self::Context) -> Self::Result {
         let JoinRoom(room_name, client_name, client) = msg;
+
         let id = self.add_client_to_room(&room_name, None, client);
         let join_msg = format!(
             "{} joined {}",
             client_name.unwrap_or_else(|| "anon".to_string()),
             room_name
         );
+
         self.send_chat_message(&room_name, &join_msg, id);
         MessageResult(id)
     }

websocket-chat-broker/src/session.rs

@@ -0,0 +1,162 @@
+use log::{debug, info};
+
+use actix::fut;
+use actix::prelude::*;
+use actix_broker::BrokerIssue;
+use actix_web_actors::ws;
+
+use crate::message::{ChatMessage, JoinRoom, LeaveRoom, ListRooms, SendMessage};
+use crate::server::WsChatServer;
+
+#[derive(Default)]
+pub struct WsChatSession {
+    id: usize,
+    room: String,
+    name: Option<String>,
+}
+
+impl WsChatSession {
+    pub fn join_room(&mut self, room_name: &str, ctx: &mut ws::WebsocketContext<Self>) {
+        let room_name = room_name.to_owned();
+
+        // First send a leave message for the current room
+        let leave_msg = LeaveRoom(self.room.clone(), self.id);
+
+        // issue_sync comes from having the `BrokerIssue` trait in scope.
+        self.issue_system_sync(leave_msg, ctx);
+
+        // Then send a join message for the new room
+        let join_msg = JoinRoom(
+            room_name.to_owned(),
+            self.name.clone(),
+            ctx.address().recipient(),
+        );
+
+        WsChatServer::from_registry()
+            .send(join_msg)
+            .into_actor(self)
+            .then(|id, act, _ctx| {
+                if let Ok(id) = id {
+                    act.id = id;
+                    act.room = room_name;
+                }
+
+                fut::ready(())
+            })
+            .wait(ctx);
+    }
+
+    pub fn list_rooms(&mut self, ctx: &mut ws::WebsocketContext<Self>) {
+        WsChatServer::from_registry()
+            .send(ListRooms)
+            .into_actor(self)
+            .then(|res, _, ctx| {
+                if let Ok(rooms) = res {
+                    for room in rooms {
+                        ctx.text(room);
+                    }
+                }
+
+                fut::ready(())
+            })
+            .wait(ctx);
+    }
+
+    pub fn send_msg(&self, msg: &str) {
+        let content = format!(
+            "{}: {}",
+            self.name.clone().unwrap_or_else(|| "anon".to_string()),
+            msg
+        );
+
+        let msg = SendMessage(self.room.clone(), self.id, content);
+
+        // issue_async comes from having the `BrokerIssue` trait in scope.
+        self.issue_system_async(msg);
+    }
+}
+
+impl Actor for WsChatSession {
+    type Context = ws::WebsocketContext<Self>;
+
+    fn started(&mut self, ctx: &mut Self::Context) {
+        self.join_room("Main", ctx);
+    }
+
+    fn stopped(&mut self, _ctx: &mut Self::Context) {
+        info!(
+            "WsChatSession closed for {}({}) in room {}",
+            self.name.clone().unwrap_or_else(|| "anon".to_string()),
+            self.id,
+            self.room
+        );
+    }
+}
+
+impl Handler<ChatMessage> for WsChatSession {
+    type Result = ();
+
+    fn handle(&mut self, msg: ChatMessage, ctx: &mut Self::Context) {
+        ctx.text(msg.0);
+    }
+}
+
+impl StreamHandler<Result<ws::Message, ws::ProtocolError>> for WsChatSession {
+    fn handle(
+        &mut self,
+        msg: Result<ws::Message, ws::ProtocolError>,
+        ctx: &mut Self::Context,
+    ) {
+        let msg = match msg {
+            Err(_) => {
+                ctx.stop();
+                return;
+            }
+            Ok(msg) => msg,
+        };
+
+        debug!("WEBSOCKET MESSAGE: {:?}", msg);
+
+        match msg {
+            ws::Message::Text(text) => {
+                let msg = text.trim();
+
+                if msg.starts_with('/') {
+                    let mut command = msg.splitn(2, ' ');
+
+                    match command.next() {
+                        Some("/list") => self.list_rooms(ctx),
+
+                        Some("/join") => {
+                            if let Some(room_name) = command.next() {
+                                self.join_room(room_name, ctx);
+                            } else {
+                                ctx.text("!!! room name is required");
+                            }
+                        }
+
+                        Some("/name") => {
+                            if let Some(name) = command.next() {
+                                self.name = Some(name.to_owned());
+                                ctx.text(format!("name changed to: {}", name));
+                            } else {
+                                ctx.text("!!! name is required");
+                            }
+                        }
+
+                        _ => ctx.text(format!("!!! unknown command: {:?}", msg)),
+                    }
+
+                    return;
+                }
+                self.send_msg(msg);
+            }
+
+            ws::Message::Close(_) => {
+                ctx.stop();
+            }
+
+            _ => {}
+        }
+    }
+}

websocket-chat-broker/static/index.html

@@ -1,90 +1,204 @@
 <!DOCTYPE html>
-<meta charset="utf-8" />
 <html>
-<head>
+  <head>
-<script src="http://ajax.googleapis.com/ajax/libs/jquery/1.4.2/jquery.min.js">
+    <meta charset="utf-8" />
-</script>
+    <title>Websocket Chat Broker</title>
-  <script language="javascript" type="text/javascript">
+
-    $(function() {
+    <style>
-      var conn = null;
+      :root {
-      function log(msg) {
+        font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto,
-        var control = $('#log');
+          Oxygen, Ubuntu, Cantarell, 'Open Sans', 'Helvetica Neue', sans-serif;
-        control.html(control.html() + msg + '<br/>');
+        font-size: 18px;
-        control.scrollTop(control.scrollTop() + 1000);
       }
+
+      input[type='text'] {
+        font-size: inherit;
+      }
+
+      #log {
+        width: 30em;
+        height: 20em;
+        overflow: auto;
+        margin: 0.5em 0;
+
+        border: 1px solid black;
+      }
+
+      #status {
+        padding: 0 0.2em;
+      }
+
+      #text {
+        width: 17em;
+        padding: 0.5em;
+      }
+
+      .msg {
+        margin: 0;
+        padding: 0.25em 0.5em;
+      }
+
+      .msg--status {
+        /* a light yellow */
+        background-color: #ffffc9;
+      }
+
+      .msg--message {
+        /* a light blue */
+        background-color: #d2f4ff;
+      }
+
+      .msg--error {
+        background-color: pink;
+      }
+    </style>
+  </head>
+  <body>
+    <h1>Chat!</h1>
+
+    <div>
+      <button id="connect">Connect</button>
+      <span>Status:</span>
+      <span id="status">disconnected</span>
+    </div>
+
+    <div id="log"></div>
+
+    <form id="chatform">
+      <input type="text" id="text" />
+      <input type="submit" id="send" />
+    </form>
+
+    <hr />
+
+    <section>
+      <h2>Commands</h2>
+      <table style="border-spacing: 0.5em;">
+        <tr>
+          <td>
+            <code>/list</code>
+          </td>
+          <td>
+            list all available rooms
+          </td>
+        </tr>
+        <tr>
+          <td>
+            <code>/join name</code>
+          </td>
+          <td>
+            join room, if room does not exist, create new one
+          </td>
+        </tr>
+        <tr>
+          <td>
+            <code>/name name</code>
+          </td>
+          <td>
+            set session name
+          </td>
+        </tr>
+        <tr>
+          <td>
+            <code>some message</code>
+          </td>
+          <td>
+            just string, send message to all peers in same room
+          </td>
+        </tr>
+      </table>
+    </section>
+
+    <script>
+      const $status = document.querySelector('#status')
+      const $connectButton = document.querySelector('#connect')
+      const $log = document.querySelector('#log')
+      const $form = document.querySelector('#chatform')
+      const $input = document.querySelector('#text')
+
+      /** @type {WebSocket | null} */
+      var socket = null
+
+      function log(msg, type = 'status') {
+        $log.innerHTML += `<p class="msg msg--${type}">${msg}</p>`
+        $log.scrollTop += 1000
+      }
+
       function connect() {
-        disconnect();
+        disconnect()
-        var wsUri = (window.location.protocol=='https:'&&'wss://'||'ws://')+window.location.host + '/ws/';
+
-        conn = new WebSocket(wsUri);
+        const { location } = window
-        log('Connecting...');
+
-        conn.onopen = function() {
+        const proto = location.protocol.startsWith('https') ? 'wss' : 'ws'
-          log('Connected.');
+        const wsUri = `${proto}://${location.host}/ws/`
-          update_ui();
+
-        };
+        log('Connecting...')
-        conn.onmessage = function(e) {
+        socket = new WebSocket(wsUri)
-          log('Received: ' + e.data);
+
-        };
+        socket.onopen = () => {
-        conn.onclose = function() {
+          log('Connected')
-          log('Disconnected.');
+          updateConnectionStatus()
-          conn = null;
+        }
-          update_ui();
+
-        };
+        socket.onmessage = (ev) => {
+          log('Received: ' + ev.data, 'message')
+        }
+
+        socket.onclose = () => {
+          log('Disconnected')
+          socket = null
+          updateConnectionStatus()
+        }
       }
+
       function disconnect() {
-        if (conn != null) {
+        if (socket) {
-          log('Disconnecting...');
+          log('Disconnecting...')
-          conn.close();
+          socket.close()
-          conn = null;
+          socket = null
-          update_ui();
+
+          updateConnectionStatus()
         }
       }
-      function update_ui() {
+
-        var msg = '';
+      function updateConnectionStatus() {
-        if (conn == null) {
+        if (socket) {
-          $('#status').text('disconnected');
+          $status.style.backgroundColor = 'transparent'
-          $('#connect').html('Connect');
+          $status.style.color = 'green'
+          $status.textContent = `connected`
+          $connectButton.innerHTML = 'Disconnect'
+          $input.focus()
         } else {
-          $('#status').text('connected (' + conn.protocol + ')');
+          $status.style.backgroundColor = 'red'
-          $('#connect').html('Disconnect');
+          $status.style.color = 'white'
+          $status.textContent = 'disconnected'
+          $connectButton.textContent = 'Connect'
         }
       }
-      $('#connect').click(function() {
+
-        if (conn == null) {
+      $connectButton.addEventListener('click', () => {
-          connect();
+        if (socket) {
+          disconnect()
         } else {
-          disconnect();
+          connect()
         }
-        update_ui();
+
-        return false;
+        updateConnectionStatus()
-      });
+      })
-      $('#send').click(function() {
+
-        var text = $('#text').val();
+      $form.addEventListener('submit', (ev) => {
-        log('Sending: ' + text);
+        ev.preventDefault()
-        conn.send(text);
+
-        $('#text').val('').focus();
+        const text = $input.value
-        return false;
+
-      });
+        log('Sending: ' + text)
-      $('#text').keyup(function(e) {
+        socket.send(text)
-        if (e.keyCode === 13) {
+
-          $('#send').click();
+        $input.value = ''
-          return false;
+        $input.focus()
-        }
+      })
-      });
+
-    });
+      updateConnectionStatus()
-</script>
+    </script>
-</head>
+  </body>
-<body>
-<h3>Chat!</h3>
-<div>
-  <button id="connect">Connect</button>&nbsp;|&nbsp;Status:
-  <span id="status">disconnected</span>
-</div>
-<div id="log"
-     style="width:20em;height:15em;overflow:auto;border:1px solid black">
-</div>
-<form id="chatform" onsubmit="return false;">
-  <input id="text" type="text" />
-  <input id="send" type="button" value="Send" />
-</form>
-</body>
 </html>