title: Rust Actix Websockets & Actix Broker
body: <p>I'm quite new to rust and actix, but I tried to build a technology prototype where a server sends protobuf messages to clients via websockets. The protobuf part works and is no problem, but I struggle with the websocket part.</p>
  <p>I've tried to modify the official example of Actix-Websockets with the Actix-Broker (<a href="https://github.com/actix/examples/tree/master/websockets" rel="nofollow noreferrer">Chat-Broker-Example</a>), but I'm having a hard time debugging it (beeing not familiar with VSCode, but thats another story).</p>
  <p>The effect is, that the broker instance is not started and does not receive any messages. If I start the server manually via a supervisor (the example does not need to do that) than it still won't receive any messages.</p>
  <p><strong>Question:</strong>
  Has anybody any idea why the broker wont start atomaticly or doesnt receive the messages?
  Do I have any fundamential missunderstandings?
  Has anybody any idea how to make the programm work?</p>
  <p>The code is uploaded publicly at github (<a href="https://github.com/PBetzler/Actix-Websocket-Test" rel="nofollow noreferrer">GitHub Repository</a>).</p>
  <p>For your convinience I'll add the ws_client, ws_server and main.rs files below.</p>
  <p>The changes I've done compared to the example are:</p>
  <ul>
  <li>Removed #[derive(Default)] from WsChatServer and implemented it myself</li>
  <li>Wrapped WsChatServer rooms in Arc and RwLock to ensure memory safty. (Needs to be overhauled)</li>
  <li>Removed Message ListRooms and corresponding functions</li>
  </ul>
  <p>I'd highly appreciate any help, tips or suggestions!</p>
  <p>ws_server.rs:</p>
  <pre><code>use std::{collections::HashMap, sync::{Arc, RwLock}};
  
  use actix::prelude::*;
  use actix_broker::BrokerSubscribe;
  
  use crate::{messages::{ChatMessage, JoinRoom, LeaveRoom, SendMessage}};
  
  type Client = Recipient&lt;ChatMessage&gt;;
  type Room = HashMap&lt;usize, Client&gt;;
  
  #[derive(Clone)]
  pub struct WsChatServer {
      rooms: Arc&lt;RwLock&lt;HashMap&lt;String, Room&gt;&gt;&gt;,
  }
  
  lazy_static! {
      static ref ROOMS: Arc&lt;RwLock&lt;HashMap&lt;String, Room&gt;&gt;&gt; = Arc::new(RwLock::new(Default::default()));
  }
  
  impl Default for WsChatServer {
      fn default() -&gt; Self {
          let ws = WsChatServer { rooms: ROOMS.clone() };
          return ws;
      }
  }
  
  
  impl SystemService for WsChatServer {}
  impl Supervised for WsChatServer {}
  
  impl WsChatServer {
  
      pub fn create_room(room_name: &amp;str) {
          let mut rooms = match ROOMS.write() {
              Ok(rooms) =&gt; rooms,
              Err(err) =&gt; {
                  log::debug!(&quot;Error while requesting write lock. Error was: {}&quot;, err);
                  return;
              },
          };
  
          if !rooms.contains_key(room_name) {
              let room: HashMap&lt;usize, Client&gt; = HashMap::new();
              rooms.insert(room_name.to_string(), room);
          }
      }
  
  
  
      fn take_room(&amp;mut self, room_name: &amp;str) -&gt; Option&lt;Room&gt; {
          let mut guard = match self.rooms.write() {
              Ok(guard) =&gt; guard,
              Err(err) =&gt; {
                  log::debug!(&quot;Error waiting for write lock. Error was: {}&quot;, err);
                  return None;
              },
          };
          let room = match guard.get_mut(room_name){
              Some(room) =&gt; room,
              None =&gt; {
                  log::debug!(&quot;Failed to get mutable reference of RW Guard&quot;);
                  return None;
              },
          };
          let room = std::mem::take(room);
          Some(room)
      }
  
      fn add_client_to_room(&amp;mut self, room_name: &amp;str, id: Option&lt;usize&gt;, client: Client) -&gt; usize {
          log::info!(&quot;In add_client_to_room Handler. Adding Client to room: {}&quot;, room_name);
          let mut id = id.unwrap_or_else(rand::random::&lt;usize&gt;);
  
          if let Some(room) = self.rooms.write().unwrap().get_mut(room_name) {
              loop {
                  if room.contains_key(&amp;id) {
                      id = rand::random::&lt;usize&gt;();
                  } else {
                      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.write().unwrap().insert(room_name.to_owned(), room);
  
          id
      }
  
      pub fn send_chat_message(&amp;mut self, room_name: &amp;str, msg: &amp;str, _src: usize) -&gt; Option&lt;()&gt; {
          let mut room = match self.take_room(room_name) {
              Some(room) =&gt; room,
              None =&gt; {
                      log::debug!(&quot;Error, could not take room.&quot;);
                      return None;
                  },
          };
  
          for (id, client) in room.drain() {
              if client.try_send(ChatMessage(msg.to_owned())).is_ok() {
                  self.add_client_to_room(room_name, Some(id), client);
              }
          }
  
          Some(())
      }
  }
  
  impl Actor for WsChatServer {
      type Context = Context&lt;Self&gt;;
  
      fn started(&amp;mut self, ctx: &amp;mut Self::Context) {
          log::info!(&quot;WsChatServer has started.&quot;);
          self.subscribe_system_async::&lt;LeaveRoom&gt;(ctx);
          self.subscribe_system_async::&lt;SendMessage&gt;(ctx);
      }
  }
  
  impl Handler&lt;JoinRoom&gt; for WsChatServer {
      type Result = MessageResult&lt;JoinRoom&gt;;
  
      fn handle(&amp;mut self, msg: JoinRoom, _ctx: &amp;mut Self::Context) -&gt; Self::Result {
          log::info!(&quot;In Join Room Handler.&quot;);
          let JoinRoom(room_name, client) = msg;
  
          let id = self.add_client_to_room(&amp;room_name, None, client);
          
          MessageResult(id)
      }
  }
  
  impl Handler&lt;LeaveRoom&gt; for WsChatServer {
      type Result = ();
  
      fn handle(&amp;mut self, msg: LeaveRoom, _ctx: &amp;mut Self::Context) {
          log::info!(&quot;Removing ws client from room.&quot;);
          if let Some(room) = self.rooms.write().unwrap().get_mut(&amp;msg.0) {
              room.remove(&amp;msg.1);
          }
      }
  }
  
  impl Handler&lt;SendMessage&gt; for WsChatServer {
      type Result = ();
  
      fn handle(&amp;mut self, msg: SendMessage, _ctx: &amp;mut Self::Context) {
          let SendMessage(room_name, id, msg) = msg;
          self.send_chat_message(&amp;room_name, &amp;msg, id);
      }
  }
  
  </code></pre>
  <p>ws_client.rs:</p>
  <pre><code>use actix::{Actor, ActorContext, StreamHandler, Handler, SystemService, AsyncContext, WrapFuture, ActorFutureExt, fut, ContextFutureSpawner};
  use actix_web_actors::ws;
  use actix_broker::BrokerIssue;
  
  use crate::messages::{ChatMessage, LeaveRoom, JoinRoom};
  use crate::ws_server::WsChatServer;
  
  
  pub struct WsConn {
      room: String,
      id: usize,
  }
  
  impl WsConn {
      pub fn new(room: &amp;str) -&gt; WsConn {
          WsConn {
              room: room.to_string(),
              id: rand::random::&lt;usize&gt;(),
          }        
      }
  
      pub fn join_room(&amp;mut self, room_name: &amp;str, ctx: &amp;mut ws::WebsocketContext&lt;Self&gt;) {
          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);      
          log::info!(&quot;Ws client sent leave msg.&quot;);  
  
          // Then send a join message for the new room
          let join_msg = JoinRoom(
              room_name.to_owned(),
              ctx.address().recipient(),
          );
          
  
          WsChatServer::from_registry()
              .send(join_msg)
              .into_actor(self)
              .then(move |id, act, _ctx| {
                  if let Ok(id) = id {
                      act.id = id;
                      act.room = room_name.clone().to_string();
                  }
  
                  fut::ready(())
              })
              .wait(ctx);
  
          log::info!(&quot;Ws client sent join msg.&quot;); 
      }
  }
  
  impl Actor for WsConn {
      type Context = ws::WebsocketContext&lt;Self&gt;;
  
      fn started(&amp;mut self, ctx: &amp;mut Self::Context) {
          log::info!(&quot;ws client started.&quot;);
          self.join_room(self.room.to_owned().as_str(), ctx);
      }
  
      fn stopped(&amp;mut self, _ctx: &amp;mut Self::Context) {
          log::info!(
              &quot;WsConn closed for {} in room {}&quot;,
              self.id,
              self.room
          );
      }
  }
  
  impl Handler&lt;ChatMessage&gt; for WsConn {
      type Result = ();
  
      fn handle(&amp;mut self, msg: ChatMessage, ctx: &amp;mut Self::Context) {
          ctx.text(msg.0);
      }
  }
  
  impl StreamHandler&lt;Result&lt;ws::Message, ws::ProtocolError&gt;&gt; for WsConn {
      fn handle(&amp;mut self, msg: Result&lt;ws::Message, ws::ProtocolError&gt;, ctx: &amp;mut Self::Context) {
          let msg = match msg {
              Err(_) =&gt; {
                  ctx.stop();
                  return;
              }
              Ok(msg) =&gt; msg,
          };
  
          log::debug!(&quot;WEBSOCKET MESSAGE: {:?}&quot;, msg);
  
          match msg {
              ws::Message::Text(_) =&gt; (),
              ws::Message::Close(reason) =&gt; {
                  ctx.close(reason);
                  ctx.stop();
              }
              _ =&gt; {}
          }
      }
  }
  
  
  </code></pre>
  <p>main.rs:</p>
  <pre><code>use std::time::Duration;
  use std::{env, thread::sleep};
  use std::fs::File;
  use std::io::Write;
  use actix_web::{App, HttpServer, middleware::Logger};
  use actix_cors::Cors;
  use tokio::task;
  
  #[macro_use]
  extern crate lazy_static;
  
  mod protobuf_messages;
  mod actions;
  mod data;
  mod ws_clients;
  mod messages;
  mod ws_server;
  
  pub async fn write_to_file(buf: &amp;[u8], file_name: &amp;str) -&gt; Result&lt;(), std::io::Error&gt; {
      let dir = env::current_dir().unwrap();
      let file_handler = dir.join(file_name);
      let mut file = File::create(file_handler).unwrap();
      file.write_all(buf)
  }
  
  
  #[actix_web::main]
  async fn main() -&gt; std::io::Result&lt;()&gt; {
      std::env::set_var(&quot;RUST_LOG&quot;, &quot;debug&quot;);
      env_logger::init();
  
      data::MAIN_CONFIG.version = &quot;1.0&quot;.to_string();
      data::MAIN_CONFIG.mqtt_broker_address = &quot;test&quot;.to_string();
      data::MAIN_CONFIG.wildcard = &quot;#&quot;.to_string();
      data::MAIN_CONFIG.splitting_character = &quot;.&quot;.to_string();
      data::MAIN_CONFIG.active_configs = [].to_vec();
  
      let ws_chat_server = ws_server::WsChatServer::default();
      let mut ws_chat_server_2 = ws_chat_server.clone();
  
      actix::Supervisor::start(|_| ws_chat_server);
  
      ws_server::WsChatServer::create_room(&quot;Main&quot;);
  
      let msg = serde_json::to_string_pretty(&amp;data::MAIN_CONFIG).expect(&quot;Expected parsable string&quot;);
      
      task::spawn(async move {
          loop {
              match ws_chat_server_2.send_chat_message(&quot;Main&quot;, &amp;msg.clone(), 0) {
                  Some(()) =&gt; (),//log::debug!(&quot;Got a result from sending chat message&quot;),
                  None =&gt; (),//log::debug!(&quot;Got no result from sending chat message&quot;),
              }
              
              sleep(Duration::from_secs(1));
          }        
      });
  
  
      HttpServer::new(move|| App::new()
          .wrap(Logger::default())
          .wrap(Cors::default().allow_any_origin().allow_any_header().allow_any_method())
          .service(actions::get_json)
          .service(actions::get_protobuf)
          .service(actions::start_ws_connection)
      )
      .bind((&quot;127.0.0.1&quot;, 3000))?
      .workers(2)
      .run().await
  
  
  }
  </code></pre>