WsDocumentMessage supports command/delta

backend/Cargo.toml

@@ -53,6 +53,7 @@ jsonwebtoken = "7.2"
 sql-builder = "3.1.1"
 lazy_static = "1.4"
 tokio = { version = "1", features = ["full"] }
+parking_lot = "0.11"
 
 flowy-user = { path = "../rust-lib/flowy-user" }
 flowy-workspace = { path = "../rust-lib/flowy-workspace" }

backend/src/application.rs

@@ -24,7 +24,6 @@ use crate::{
         ws::WSServer,
     },
 };
-use flowy_ws::WsSource;
 
 pub struct Application {
     port: u16,
@@ -55,7 +54,7 @@ pub fn run(listener: TcpListener, app_ctx: AppContext) -> Result<Server, std::io
     let pg_pool = Data::new(pg_pool);
     let domain = domain();
     let secret: String = secret();
-    let ws_biz_handlers = Data::new(make_ws_biz_handlers());
+    let ws_biz_handlers = Data::new(make_ws_biz_handlers(pg_pool.clone()));
     actix_rt::spawn(period_check(pg_pool.clone()));
 
     let server = HttpServer::new(move || {

backend/src/service/doc/ws_handler.rs

@@ -1,17 +1,54 @@
 use crate::service::{util::parse_from_bytes, ws::WsBizHandler};
+use actix_web::web::Data;
 use bytes::Bytes;
-use flowy_document::protobuf::Revision;
+use dashmap::DashMap;
+use flowy_document::{
+    protobuf::{Revision, WsDataType, WsDocumentData},
+    services::doc::Document,
+};
+use parking_lot::RwLock;
 use protobuf::Message;
+use sqlx::PgPool;
+use std::sync::Arc;
 
-pub struct DocWsBizHandler {}
+#[rustfmt::skip]
+//
+//                 Frontend              │                 Backend
+//
+// ┌──────────┐        ┌──────────┐      │     ┌─────────┐            ┌───────────────┐
+// │  user 1  │───────▶│WsManager │───────────▶│ws_client│───────────▶│DocWsBizHandler│
+// └──────────┘        └──────────┘      │     └─────────┘            └───────────────┘
+//
+//   WsDocumentData────▶WsMessage ────▶ Message  ─────▶WsMessage ─────▶WsDocumentData
+
+pub struct DocWsBizHandler {
+    pg_pool: Data<PgPool>,
+    edit_docs: DashMap<String, Arc<RwLock<EditDoc>>>,
+}
 
 impl DocWsBizHandler {
-    pub fn new() -> Self { Self {} }
+    pub fn new(pg_pool: Data<PgPool>) -> Self {
+        Self {
+            edit_docs: DashMap::new(),
+            pg_pool,
+        }
+    }
 }
 
 impl WsBizHandler for DocWsBizHandler {
     fn receive_data(&self, data: Bytes) {
-        let revision: Revision = parse_from_bytes(&data).unwrap();
-        log::warn!("{:?}", revision);
+        let document_data: WsDocumentData = parse_from_bytes(&data).unwrap();
+        match document_data.ty {
+            WsDataType::Command => {},
+            WsDataType::Delta => {
+                let revision: Revision = parse_from_bytes(&document_data.data).unwrap();
+                log::warn!("{:?}", revision);
+            },
+        }
     }
 }
+
+pub struct EditDoc {
+    doc_id: String,
+    document: Document,
+}

backend/src/service/mod.rs

@@ -1,7 +1,8 @@
 use crate::service::{doc::ws_handler::DocWsBizHandler, ws::WsBizHandlers};
-use flowy_ws::WsSource;
+use actix_web::web::Data;
+use flowy_ws::WsModule;
+use sqlx::PgPool;
 use std::sync::Arc;
-use tokio::sync::RwLock;
 
 pub mod app;
 pub mod doc;
@@ -12,15 +13,15 @@ pub mod view;
 pub mod workspace;
 pub mod ws;
 
-pub fn make_ws_biz_handlers() -> WsBizHandlers {
+pub fn make_ws_biz_handlers(pg_pool: Data<PgPool>) -> WsBizHandlers {
     let mut ws_biz_handlers = WsBizHandlers::new();
 
     // doc
-    let doc_biz_handler = DocWsBizHandler::new();
-    ws_biz_handlers.register(WsSource::Doc, wrap(doc_biz_handler));
+    let doc_biz_handler = DocWsBizHandler::new(pg_pool);
+    ws_biz_handlers.register(WsModule::Doc, wrap(doc_biz_handler));
 
     //
     ws_biz_handlers
 }
 
-fn wrap<T>(val: T) -> Arc<RwLock<T>> { Arc::new(RwLock::new(val)) }
+fn wrap<T>(val: T) -> Arc<T> { Arc::new(val) }

backend/src/service/ws/biz_handler.rs

@@ -1,31 +1,28 @@
 use bytes::Bytes;
-use dashmap::{mapref::one::Ref, DashMap};
-use flowy_ws::WsSource;
-use std::sync::Arc;
-use tokio::sync::RwLock;
+use flowy_ws::WsModule;
+use std::{collections::HashMap, sync::Arc};
 
 pub trait WsBizHandler: Send + Sync {
     fn receive_data(&self, data: Bytes);
 }
 
-pub type BizHandler = Arc<RwLock<dyn WsBizHandler>>;
-
+pub type BizHandler = Arc<dyn WsBizHandler>;
 pub struct WsBizHandlers {
-    inner: DashMap<WsSource, BizHandler>,
+    inner: HashMap<WsModule, BizHandler>,
 }
 
 impl WsBizHandlers {
     pub fn new() -> Self {
         Self {
-            inner: DashMap::new(),
+            inner: HashMap::new(),
         }
     }
 
-    pub fn register(&self, source: WsSource, handler: BizHandler) {
+    pub fn register(&mut self, source: WsModule, handler: BizHandler) {
         self.inner.insert(source, handler);
     }
 
-    pub fn get(&self, source: &WsSource) -> Option<BizHandler> {
+    pub fn get(&self, source: &WsModule) -> Option<BizHandler> {
         match self.inner.get(source) {
             None => None,
             Some(handler) => Some(handler.clone()),

backend/src/service/ws/ws_client.rs

@@ -9,13 +9,12 @@ use crate::{
         WsBizHandlers,
     },
 };
-use actix::{fut::wrap_future, *};
+use actix::*;
 use actix_web::web::Data;
 use actix_web_actors::{ws, ws::Message::Text};
 use bytes::Bytes;
-use flowy_ws::{WsMessage, WsSource};
-use std::{convert::TryFrom, pin::Pin, time::Instant};
-use tokio::sync::RwLock;
+use flowy_ws::WsMessage;
+use std::{convert::TryFrom, time::Instant};
 
 pub struct WSClient {
     session_id: SessionId,
@@ -55,18 +54,16 @@ impl WSClient {
         let msg = ClientMessage::new(self.session_id.clone(), data);
         self.server.do_send(msg);
     }
-}
 
-async fn handle_binary_message(biz_handlers: Data<WsBizHandlers>, bytes: Bytes) {
-    let message: WsMessage = WsMessage::try_from(bytes).unwrap();
-    match biz_handlers.get(&message.source) {
-        None => {
-            log::error!("Can't find the handler for {:?}", message.source);
-        },
-        Some(handler) => handler
-            .write()
-            .await
-            .receive_data(Bytes::from(message.data)),
+    fn handle_binary_message(&self, bytes: Bytes) {
+        // TODO: ok to unwrap?
+        let message: WsMessage = WsMessage::try_from(bytes).unwrap();
+        match self.biz_handlers.get(&message.module) {
+            None => {
+                log::error!("Can't find the handler for {:?}", message.module);
+            },
+            Some(handler) => handler.receive_data(Bytes::from(message.data)),
+        }
     }
 }
 
@@ -83,8 +80,7 @@ impl StreamHandler<Result<ws::Message, ws::ProtocolError>> for WSClient {
             },
             Ok(ws::Message::Binary(bytes)) => {
                 log::debug!(" Receive {} binary", &self.session_id);
-                let biz_handlers = self.biz_handlers.clone();
-                ctx.spawn(wrap_future(handle_binary_message(biz_handlers, bytes)));
+                self.handle_binary_message(bytes);
             },
             Ok(Text(_)) => {
                 log::warn!("Receive unexpected text message");

backend/tests/ws/helper.rs

@@ -1,5 +1,5 @@
 use crate::helper::TestServer;
-use flowy_ws::{WsController, WsSender, WsState};
+use flowy_ws::{WsController, WsModule, WsSender, WsState};
 use parking_lot::RwLock;
 use std::sync::Arc;
 
@@ -26,7 +26,7 @@ impl WsTest {
                     WsScriptRunner {
                         scripts: scripts.clone(),
                         sender: sender.clone(),
-                        source: "editor".to_owned(),
+                        source: WsModule::Doc,
                     }
                     .run();
                 },
@@ -54,7 +54,7 @@ impl WsTest {
 struct WsScriptRunner {
     scripts: Vec<WsScript>,
     sender: Arc<WsSender>,
-    source: String,
+    source: WsModule,
 }
 
 impl WsScriptRunner {