Restructure folders (#411)

websockets/chat-broker/src/main.rs

@@ -0,0 +1,37 @@
+use log::info;
+
+use actix_files::Files;
+use actix_web::{web, App, Error, HttpRequest, HttpResponse, HttpServer};
+use actix_web_actors::ws;
+
+mod message;
+mod server;
+mod session;
+
+use session::WsChatSession;
+
+async fn chat_route(
+    req: HttpRequest,
+    stream: web::Payload,
+) -> Result<HttpResponse, Error> {
+    ws::start(WsChatSession::default(), &req, stream)
+}
+
+#[actix_web::main]
+async fn main() -> std::io::Result<()> {
+    env_logger::Builder::from_env(env_logger::Env::default().default_filter_or("info"))
+        .init();
+
+    let addr = "127.0.0.1:8080";
+
+    let srv = HttpServer::new(move || {
+        App::new()
+            .service(web::resource("/ws/").to(chat_route))
+            .service(Files::new("/", "./static/").index_file("index.html"))
+    })
+    .bind(&addr)?;
+
+    info!("Starting http server: {}", &addr);
+
+    srv.run().await
+}

websockets/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);

websockets/chat-broker/src/server.rs

@@ -0,0 +1,127 @@
+use actix::prelude::*;
+use actix_broker::BrokerSubscribe;
+
+use std::collections::HashMap;
+use std::mem;
+
+use crate::message::{ChatMessage, JoinRoom, LeaveRoom, ListRooms, SendMessage};
+
+type Client = Recipient<ChatMessage>;
+type Room = HashMap<usize, Client>;
+
+#[derive(Default)]
+pub struct WsChatServer {
+    rooms: HashMap<String, Room>,
+}
+
+impl WsChatServer {
+    fn take_room(&mut self, room_name: &str) -> Option<Room> {
+        let room = self.rooms.get_mut(room_name)?;
+        let room = mem::replace(room, HashMap::new());
+        Some(room)
+    }
+
+    fn add_client_to_room(
+        &mut self,
+        room_name: &str,
+        id: Option<usize>,
+        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) {
+                    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.insert(room_name.to_owned(), room);
+
+        id
+    }
+
+    fn send_chat_message(
+        &mut self,
+        room_name: &str,
+        msg: &str,
+        _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(())
+    }
+}
+
+impl Actor for WsChatServer {
+    type Context = Context<Self>;
+
+    fn started(&mut self, ctx: &mut Self::Context) {
+        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 {
+        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)
+    }
+}
+
+impl Handler<LeaveRoom> for WsChatServer {
+    type Result = ();
+
+    fn handle(&mut self, msg: LeaveRoom, _ctx: &mut Self::Context) {
+        if let Some(room) = self.rooms.get_mut(&msg.0) {
+            room.remove(&msg.1);
+        }
+    }
+}
+
+impl Handler<ListRooms> for WsChatServer {
+    type Result = MessageResult<ListRooms>;
+
+    fn handle(&mut self, _: ListRooms, _ctx: &mut Self::Context) -> Self::Result {
+        MessageResult(self.rooms.keys().cloned().collect())
+    }
+}
+
+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);
+    }
+}
+
+impl SystemService for WsChatServer {}
+impl Supervised for WsChatServer {}

websockets/chat-broker/src/session.rs

@@ -0,0 +1,161 @@
+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(reason) => {
+                ctx.close(reason);
+                ctx.stop();
+            }
+            _ => {}
+        }
+    }
+}