Rust Actix Websockets & Actix Broker

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.

I've tried to modify the official example of Actix-Websockets with the Actix-Broker (Chat-Broker-Example), but I'm having a hard time debugging it (beeing not familiar with VSCode, but thats another story).

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.

Question: 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?

The code is uploaded publicly at github (GitHub Repository).

For your convinience I'll add the ws_client, ws_server and main.rs files below.

The changes I've done compared to the example are:

• Removed #[derive(Default)] from WsChatServer and implemented it myself
• Wrapped WsChatServer rooms in Arc and RwLock to ensure memory safty. (Needs to be overhauled)
• Removed Message ListRooms and corresponding functions

I'd highly appreciate any help, tips or suggestions!

ws_server.rs:

use std::{collections::HashMap, sync::{Arc, RwLock}};

use actix::prelude::*;
use actix_broker::BrokerSubscribe;

use crate::{messages::{ChatMessage, JoinRoom, LeaveRoom, SendMessage}};

type Client = Recipient<ChatMessage>;
type Room = HashMap<usize, Client>;

#[derive(Clone)]
pub struct WsChatServer {
    rooms: Arc<RwLock<HashMap<String, Room>>>,
}

lazy_static! {
    static ref ROOMS: Arc<RwLock<HashMap<String, Room>>> = Arc::new(RwLock::new(Default::default()));
}

impl Default for WsChatServer {
    fn default() -> 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: &str) {
        let mut rooms = match ROOMS.write() {
            Ok(rooms) => rooms,
            Err(err) => {
                log::debug!("Error while requesting write lock. Error was: {}", err);
                return;
            },
        };

        if !rooms.contains_key(room_name) {
            let room: HashMap<usize, Client> = HashMap::new();
            rooms.insert(room_name.to_string(), room);
        }
    }



    fn take_room(&mut self, room_name: &str) -> Option<Room> {
        let mut guard = match self.rooms.write() {
            Ok(guard) => guard,
            Err(err) => {
                log::debug!("Error waiting for write lock. Error was: {}", err);
                return None;
            },
        };
        let room = match guard.get_mut(room_name){
            Some(room) => room,
            None => {
                log::debug!("Failed to get mutable reference of RW Guard");
                return None;
            },
        };
        let room = std::mem::take(room);
        Some(room)
    }

    fn add_client_to_room(&mut self, room_name: &str, id: Option<usize>, client: Client) -> usize {
        log::info!("In add_client_to_room Handler. Adding Client to room: {}", room_name);
        let mut id = id.unwrap_or_else(rand::random::<usize>);

        if let Some(room) = self.rooms.write().unwrap().get_mut(room_name) {
            loop {
                if room.contains_key(&id) {
                    id = rand::random::<usize>();
                } 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(&mut self, room_name: &str, msg: &str, _src: usize) -> Option<()> {
        let mut room = match self.take_room(room_name) {
            Some(room) => room,
            None => {
                    log::debug!("Error, could not take room.");
                    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<Self>;

    fn started(&mut self, ctx: &mut Self::Context) {
        log::info!("WsChatServer has started.");
        self.subscribe_system_async::<LeaveRoom>(ctx);
        self.subscribe_system_async::<SendMessage>(ctx);
    }
}

impl Handler<JoinRoom> for WsChatServer {
    type Result = MessageResult<JoinRoom>;

    fn handle(&mut self, msg: JoinRoom, _ctx: &mut Self::Context) -> Self::Result {
        log::info!("In Join Room Handler.");
        let JoinRoom(room_name, client) = msg;

        let id = self.add_client_to_room(&room_name, None, client);
        
        MessageResult(id)
    }
}

impl Handler<LeaveRoom> for WsChatServer {
    type Result = ();

    fn handle(&mut self, msg: LeaveRoom, _ctx: &mut Self::Context) {
        log::info!("Removing ws client from room.");
        if let Some(room) = self.rooms.write().unwrap().get_mut(&msg.0) {
            room.remove(&msg.1);
        }
    }
}

impl Handler<SendMessage> for WsChatServer {
    type Result = ();

    fn handle(&mut self, msg: SendMessage, _ctx: &mut Self::Context) {
        let SendMessage(room_name, id, msg) = msg;
        self.send_chat_message(&room_name, &msg, id);
    }
}

ws_client.rs:

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: &str) -> WsConn {
        WsConn {
            room: room.to_string(),
            id: rand::random::<usize>(),
        }        
    }

    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);      
        log::info!("Ws client sent leave msg.");  

        // 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!("Ws client sent join msg."); 
    }
}

impl Actor for WsConn {
    type Context = ws::WebsocketContext<Self>;

    fn started(&mut self, ctx: &mut Self::Context) {
        log::info!("ws client started.");
        self.join_room(self.room.to_owned().as_str(), ctx);
    }

    fn stopped(&mut self, _ctx: &mut Self::Context) {
        log::info!(
            "WsConn closed for {} in room {}",
            self.id,
            self.room
        );
    }
}

impl Handler<ChatMessage> for WsConn {
    type Result = ();

    fn handle(&mut self, msg: ChatMessage, ctx: &mut Self::Context) {
        ctx.text(msg.0);
    }
}

impl StreamHandler<Result<ws::Message, ws::ProtocolError>> for WsConn {
    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,
        };

        log::debug!("WEBSOCKET MESSAGE: {:?}", msg);

        match msg {
            ws::Message::Text(_) => (),
            ws::Message::Close(reason) => {
                ctx.close(reason);
                ctx.stop();
            }
            _ => {}
        }
    }
}

main.rs:

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: &[u8], file_name: &str) -> Result<(), std::io::Error> {
    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() -> std::io::Result<()> {
    std::env::set_var("RUST_LOG", "debug");
    env_logger::init();

    data::MAIN_CONFIG.version = "1.0".to_string();
    data::MAIN_CONFIG.mqtt_broker_address = "test".to_string();
    data::MAIN_CONFIG.wildcard = "#".to_string();
    data::MAIN_CONFIG.splitting_character = ".".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("Main");

    let msg = serde_json::to_string_pretty(&data::MAIN_CONFIG).expect("Expected parsable string");
    
    task::spawn(async move {
        loop {
            match ws_chat_server_2.send_chat_message("Main", &msg.clone(), 0) {
                Some(()) => (),//log::debug!("Got a result from sending chat message"),
                None => (),//log::debug!("Got no result from sending chat message"),
            }
            
            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(("127.0.0.1", 3000))?
    .workers(2)
    .run().await


}