config folder ws

frontend/rust-lib/flowy-document/src/context.rs

@@ -1,11 +1,6 @@
-use crate::{
-    controller::DocumentController,
-    errors::FlowyError,
-    ws_receivers::DocumentWSReceivers,
-    DocumentCloudService,
-};
+use crate::{controller::DocumentController, errors::FlowyError};
 use flowy_database::ConnectionPool;
-use flowy_sync::RevisionWebSocket;
+
 use std::sync::Arc;
 
 pub trait DocumentUser: Send + Sync {
@@ -21,24 +16,6 @@ pub struct DocumentContext {
 }
 
 impl DocumentContext {
-    pub fn new(
-        user: Arc<dyn DocumentUser>,
-        ws_receivers: Arc<DocumentWSReceivers>,
-        ws_sender: Arc<dyn RevisionWebSocket>,
-        cloud_service: Arc<dyn DocumentCloudService>,
-    ) -> DocumentContext {
-        let doc_ctrl = Arc::new(DocumentController::new(
-            cloud_service,
-            user.clone(),
-            ws_receivers,
-            ws_sender,
-        ));
-        Self {
-            controller: doc_ctrl,
-            user,
-        }
-    }
-
     pub fn init(&self) -> Result<(), FlowyError> {
         let _ = self.controller.init()?;
         Ok(())

frontend/rust-lib/flowy-document/src/controller.rs

@@ -1,49 +1,52 @@
-use crate::{
-    context::DocumentUser,
-    core::ClientDocumentEditor,
-    errors::FlowyError,
-    ws_receivers::DocumentWSReceivers,
-    DocumentCloudService,
-};
+use crate::{context::DocumentUser, core::ClientDocumentEditor, errors::FlowyError, DocumentCloudService};
+use async_trait::async_trait;
 use bytes::Bytes;
 use dashmap::DashMap;
 use flowy_collaboration::entities::{
     doc::{DocumentDelta, DocumentId},
     revision::{md5, RepeatedRevision, Revision},
+    ws::ServerRevisionWSData,
 };
 use flowy_database::ConnectionPool;
 use flowy_error::FlowyResult;
 use flowy_sync::{RevisionCache, RevisionCloudService, RevisionManager, RevisionWebSocket, WSStateReceiver};
 use lib_infra::future::FutureResult;
-use std::sync::Arc;
+use lib_ws::WSConnectState;
+use std::{convert::TryInto, sync::Arc};
 
+#[async_trait]
+pub(crate) trait DocumentWSReceiver: Send + Sync {
+    async fn receive_ws_data(&self, data: ServerRevisionWSData) -> Result<(), FlowyError>;
+    fn connect_state_changed(&self, state: WSConnectState);
+}
+type WebSocketDataReceivers = Arc<DashMap<String, Arc<dyn DocumentWSReceiver>>>;
 pub struct DocumentController {
     cloud_service: Arc<dyn DocumentCloudService>,
-    ws_receivers: Arc<DocumentWSReceivers>,
-    ws_sender: Arc<dyn RevisionWebSocket>,
+    ws_receivers: WebSocketDataReceivers,
+    web_socket: Arc<dyn RevisionWebSocket>,
     open_cache: Arc<OpenDocCache>,
     user: Arc<dyn DocumentUser>,
 }
 
 impl DocumentController {
-    pub(crate) fn new(
+    pub fn new(
         cloud_service: Arc<dyn DocumentCloudService>,
         user: Arc<dyn DocumentUser>,
-        ws_receivers: Arc<DocumentWSReceivers>,
-        ws_sender: Arc<dyn RevisionWebSocket>,
+        web_socket: Arc<dyn RevisionWebSocket>,
     ) -> Self {
+        let ws_receivers = Arc::new(DashMap::new());
         let open_cache = Arc::new(OpenDocCache::new());
         Self {
             cloud_service,
             ws_receivers,
-            ws_sender,
+            web_socket,
             open_cache,
             user,
         }
     }
 
     pub(crate) fn init(&self) -> FlowyResult<()> {
-        let notify = self.ws_sender.subscribe_state_changed();
+        let notify = self.web_socket.subscribe_state_changed();
         listen_ws_state_changed(notify, self.ws_receivers.clone());
 
         Ok(())
@@ -61,7 +64,7 @@ impl DocumentController {
         let doc_id = doc_id.as_ref();
         tracing::Span::current().record("doc_id", &doc_id);
         self.open_cache.remove(doc_id);
-        self.ws_receivers.remove(doc_id);
+        self.remove_ws_receiver(doc_id);
         Ok(())
     }
 
@@ -70,7 +73,7 @@ impl DocumentController {
         let doc_id = doc_id.as_ref();
         tracing::Span::current().record("doc_id", &doc_id);
         self.open_cache.remove(doc_id);
-        self.ws_receivers.remove(doc_id);
+        self.remove_ws_receiver(doc_id);
         Ok(())
     }
 
@@ -93,6 +96,25 @@ impl DocumentController {
         Ok(())
     }
 
+    pub async fn did_receive_ws_data(&self, data: Bytes) {
+        let data: ServerRevisionWSData = data.try_into().unwrap();
+        match self.ws_receivers.get(&data.object_id) {
+            None => tracing::error!("Can't find any source handler for {:?}", data.object_id),
+            Some(handler) => match handler.receive_ws_data(data).await {
+                Ok(_) => {},
+                Err(e) => tracing::error!("{}", e),
+            },
+        }
+    }
+
+    pub async fn ws_connect_state_changed(&self, state: &WSConnectState) {
+        for receiver in self.ws_receivers.iter() {
+            receiver.value().connect_state_changed(state.clone());
+        }
+    }
+}
+
+impl DocumentController {
     async fn get_editor(&self, doc_id: &str) -> FlowyResult<Arc<ClientDocumentEditor>> {
         match self.open_cache.get(doc_id) {
             None => {
@@ -102,9 +124,7 @@ impl DocumentController {
             Some(editor) => Ok(editor),
         }
     }
-}
 
-impl DocumentController {
     async fn make_editor(
         &self,
         doc_id: &str,
@@ -117,8 +137,8 @@ impl DocumentController {
             token,
             server: self.cloud_service.clone(),
         });
-        let doc_editor = ClientDocumentEditor::new(doc_id, user, rev_manager, self.ws_sender.clone(), server).await?;
-        self.ws_receivers.add(doc_id, doc_editor.ws_handler());
+        let doc_editor = ClientDocumentEditor::new(doc_id, user, rev_manager, self.web_socket.clone(), server).await?;
+        self.add_ws_receiver(doc_id, doc_editor.ws_handler());
         self.open_cache.insert(&doc_id, &doc_editor);
         Ok(doc_editor)
     }
@@ -128,6 +148,15 @@ impl DocumentController {
         let cache = Arc::new(RevisionCache::new(&user_id, doc_id, pool));
         Ok(RevisionManager::new(&user_id, doc_id, cache))
     }
+
+    fn add_ws_receiver(&self, object_id: &str, receiver: Arc<dyn DocumentWSReceiver>) {
+        if self.ws_receivers.contains_key(object_id) {
+            log::error!("Duplicate handler registered for {:?}", object_id);
+        }
+        self.ws_receivers.insert(object_id.to_string(), receiver);
+    }
+
+    fn remove_ws_receiver(&self, id: &str) { self.ws_receivers.remove(id); }
 }
 
 struct RevisionServerImpl {
@@ -194,10 +223,12 @@ impl OpenDocCache {
 }
 
 #[tracing::instrument(level = "debug", skip(state_receiver, receivers))]
-fn listen_ws_state_changed(mut state_receiver: WSStateReceiver, receivers: Arc<DocumentWSReceivers>) {
+fn listen_ws_state_changed(mut state_receiver: WSStateReceiver, receivers: WebSocketDataReceivers) {
     tokio::spawn(async move {
         while let Ok(state) = state_receiver.recv().await {
-            receivers.ws_connect_state_changed(&state).await;
+            for receiver in receivers.iter() {
+                receiver.value().connect_state_changed(state.clone());
+            }
         }
     });
 }

frontend/rust-lib/flowy-document/src/core/editor.rs

@@ -2,7 +2,7 @@ use crate::{
     context::DocumentUser,
     core::{make_document_ws_manager, EditorCommand, EditorCommandQueue, EditorCommandSender},
     errors::FlowyError,
-    ws_receivers::DocumentWSReceiver,
+    DocumentWSReceiver,
 };
 use bytes::Bytes;
 use flowy_collaboration::{
@@ -38,7 +38,7 @@ impl ClientDocumentEditor {
         doc_id: &str,
         user: Arc<dyn DocumentUser>,
         mut rev_manager: RevisionManager,
-        ws: Arc<dyn RevisionWebSocket>,
+        web_socket: Arc<dyn RevisionWebSocket>,
         server: Arc<dyn RevisionCloudService>,
     ) -> FlowyResult<Arc<Self>> {
         let document_info = rev_manager.load::<DocumentInfoBuilder>(server).await?;
@@ -53,7 +53,7 @@ impl ClientDocumentEditor {
             user_id.clone(),
             edit_cmd_tx.clone(),
             rev_manager.clone(),
-            ws,
+            web_socket,
         )
         .await;
         let editor = Arc::new(Self {

frontend/rust-lib/flowy-document/src/core/web_socket.rs

@@ -1,6 +1,6 @@
 use crate::{
     core::{EditorCommand, TransformDeltas, SYNC_INTERVAL_IN_MILLIS},
-    ws_receivers::DocumentWSReceiver,
+    DocumentWSReceiver,
 };
 use async_trait::async_trait;
 use bytes::Bytes;
@@ -37,7 +37,7 @@ pub(crate) async fn make_document_ws_manager(
     user_id: String,
     edit_cmd_tx: EditorCommandSender,
     rev_manager: Arc<RevisionManager>,
-    ws_conn: Arc<dyn RevisionWebSocket>,
+    web_socket: Arc<dyn RevisionWebSocket>,
 ) -> Arc<RevisionWebSocketManager> {
     let shared_sink = Arc::new(SharedWSSinkDataProvider::new(rev_manager.clone()));
     let ws_stream_consumer = Arc::new(DocumentWebSocketSteamConsumerAdapter {
@@ -50,7 +50,7 @@ pub(crate) async fn make_document_ws_manager(
     let ping_duration = Duration::from_millis(SYNC_INTERVAL_IN_MILLIS);
     let ws_manager = Arc::new(RevisionWebSocketManager::new(
         &doc_id,
-        ws_conn,
+        web_socket,
         data_provider,
         ws_stream_consumer,
         ping_duration,

frontend/rust-lib/flowy-document/src/lib.rs

@@ -1,10 +1,9 @@
 pub mod context;
-pub(crate) mod controller;
+mod controller;
 pub mod core;
 // mod notify;
 pub mod protobuf;
-pub mod ws_receivers;
-
+pub use controller::*;
 pub mod errors {
     pub use flowy_error::{internal_error, ErrorCode, FlowyError};
 }

frontend/rust-lib/flowy-document/src/ws_receivers.rs

@@ -1,56 +0,0 @@
-use crate::errors::FlowyError;
-use async_trait::async_trait;
-use bytes::Bytes;
-use dashmap::DashMap;
-use flowy_collaboration::entities::ws::{ServerRevisionWSData};
-use lib_ws::WSConnectState;
-use std::{convert::TryInto, sync::Arc};
-
-#[async_trait]
-pub(crate) trait DocumentWSReceiver: Send + Sync {
-    async fn receive_ws_data(&self, data: ServerRevisionWSData) -> Result<(), FlowyError>;
-    fn connect_state_changed(&self, state: WSConnectState);
-}
-
-pub struct DocumentWSReceivers {
-    // key: the document id
-    // value: DocumentWSReceiver
-    receivers: Arc<DashMap<String, Arc<dyn DocumentWSReceiver>>>,
-}
-
-impl std::default::Default for DocumentWSReceivers {
-    fn default() -> Self {
-        let receivers: Arc<DashMap<String, Arc<dyn DocumentWSReceiver>>> = Arc::new(DashMap::new());
-        DocumentWSReceivers { receivers }
-    }
-}
-
-impl DocumentWSReceivers {
-    pub fn new() -> Self { DocumentWSReceivers::default() }
-
-    pub(crate) fn add(&self, doc_id: &str, receiver: Arc<dyn DocumentWSReceiver>) {
-        if self.receivers.contains_key(doc_id) {
-            log::error!("Duplicate handler registered for {:?}", doc_id);
-        }
-        self.receivers.insert(doc_id.to_string(), receiver);
-    }
-
-    pub(crate) fn remove(&self, id: &str) { self.receivers.remove(id); }
-
-    pub async fn did_receive_data(&self, data: Bytes) {
-        let data: ServerRevisionWSData = data.try_into().unwrap();
-        match self.receivers.get(&data.object_id) {
-            None => tracing::error!("Can't find any source handler for {:?}", data.object_id),
-            Some(handler) => match handler.receive_ws_data(data).await {
-                Ok(_) => {},
-                Err(e) => tracing::error!("{}", e),
-            },
-        }
-    }
-
-    pub async fn ws_connect_state_changed(&self, state: &WSConnectState) {
-        for receiver in self.receivers.iter() {
-            receiver.value().connect_state_changed(state.clone());
-        }
-    }
-}