chore: rename some struct

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

@@ -1,7 +1,8 @@
-use crate::core::DocumentRevisionCompact;
+use crate::queue::DocumentRevisionCompact;
+use crate::web_socket::{make_document_ws_manager, EditorCommandSender};
 use crate::{
-    core::{make_document_ws_manager, EditorCommand, EditorCommandQueue, EditorCommandSender},
     errors::FlowyError,
+    queue::{EditorCommand, EditorCommandQueue},
     DocumentUser, DocumentWSReceiver,
 };
 use bytes::Bytes;
@@ -34,11 +35,11 @@ impl ClientDocumentEditor {
         doc_id: &str,
         user: Arc<dyn DocumentUser>,
         mut rev_manager: RevisionManager,
-        web_socket: Arc<dyn RevisionWebSocket>,
-        server: Arc<dyn RevisionCloudService>,
+        rev_web_socket: Arc<dyn RevisionWebSocket>,
+        cloud_service: Arc<dyn RevisionCloudService>,
     ) -> FlowyResult<Arc<Self>> {
         let document_info = rev_manager
-            .load::<DocumentInfoBuilder, DocumentRevisionCompact>(server)
+            .load::<DocumentInfoBuilder, DocumentRevisionCompact>(cloud_service)
             .await?;
         let delta = document_info.delta()?;
         let rev_manager = Arc::new(rev_manager);
@@ -51,7 +52,7 @@ impl ClientDocumentEditor {
             user_id.clone(),
             edit_cmd_tx.clone(),
             rev_manager.clone(),
-            web_socket,
+            rev_web_socket,
         )
         .await;
         let editor = Arc::new(Self {

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

@@ -1,10 +1,15 @@
-pub mod controller;
-pub mod core;
-pub use controller::*;
+pub mod editor;
+pub mod manager;
+mod queue;
+mod web_socket;
+
+pub use manager::*;
 pub mod errors {
     pub use flowy_error::{internal_error, ErrorCode, FlowyError};
 }
 
+pub const DOCUMENT_SYNC_INTERVAL_IN_MILLIS: u64 = 1000;
+
 use crate::errors::FlowyError;
 use flowy_collaboration::entities::document_info::{CreateDocParams, DocumentId, DocumentInfo, ResetDocumentParams};
 use lib_infra::future::FutureResult;

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

@@ -1,4 +1,4 @@
-use crate::{core::ClientDocumentEditor, errors::FlowyError, DocumentCloudService};
+use crate::{editor::ClientDocumentEditor, errors::FlowyError, DocumentCloudService};
 use async_trait::async_trait;
 use bytes::Bytes;
 use dashmap::DashMap;
@@ -29,31 +29,31 @@ pub(crate) trait DocumentWSReceiver: Send + Sync {
 type WebSocketDataReceivers = Arc<DashMap<String, Arc<dyn DocumentWSReceiver>>>;
 pub struct FlowyDocumentManager {
     cloud_service: Arc<dyn DocumentCloudService>,
-    ws_receivers: WebSocketDataReceivers,
-    web_socket: Arc<dyn RevisionWebSocket>,
-    open_cache: Arc<OpenDocCache>,
-    user: Arc<dyn DocumentUser>,
+    ws_data_receivers: WebSocketDataReceivers,
+    rev_web_socket: Arc<dyn RevisionWebSocket>,
+    document_handlers: Arc<DocumentEditorHandlers>,
+    document_user: Arc<dyn DocumentUser>,
 }
 
 impl FlowyDocumentManager {
     pub fn new(
         cloud_service: Arc<dyn DocumentCloudService>,
-        user: Arc<dyn DocumentUser>,
-        web_socket: Arc<dyn RevisionWebSocket>,
+        document_user: Arc<dyn DocumentUser>,
+        rev_web_socket: Arc<dyn RevisionWebSocket>,
     ) -> Self {
-        let ws_receivers = Arc::new(DashMap::new());
-        let open_cache = Arc::new(OpenDocCache::new());
+        let ws_data_receivers = Arc::new(DashMap::new());
+        let document_handlers = Arc::new(DocumentEditorHandlers::new());
         Self {
             cloud_service,
-            ws_receivers,
-            web_socket,
-            open_cache,
-            user,
+            ws_data_receivers,
+            rev_web_socket,
+            document_handlers,
+            document_user,
         }
     }
 
     pub fn init(&self) -> FlowyResult<()> {
-        listen_ws_state_changed(self.web_socket.clone(), self.ws_receivers.clone());
+        listen_ws_state_changed(self.rev_web_socket.clone(), self.ws_data_receivers.clone());
 
         Ok(())
     }
@@ -69,17 +69,17 @@ impl FlowyDocumentManager {
     pub fn close_document<T: AsRef<str>>(&self, doc_id: T) -> Result<(), FlowyError> {
         let doc_id = doc_id.as_ref();
         tracing::Span::current().record("doc_id", &doc_id);
-        self.open_cache.remove(doc_id);
-        self.remove_ws_receiver(doc_id);
+        self.document_handlers.remove(doc_id);
+        self.ws_data_receivers.remove(doc_id);
         Ok(())
     }
 
     #[tracing::instrument(level = "debug", skip(self, doc_id), fields(doc_id), err)]
     pub fn delete<T: AsRef<str>>(&self, doc_id: T) -> Result<(), FlowyError> {
         let doc_id = doc_id.as_ref();
         tracing::Span::current().record("doc_id", &doc_id);
-        self.open_cache.remove(doc_id);
-        self.remove_ws_receiver(doc_id);
+        self.document_handlers.remove(doc_id);
+        self.ws_data_receivers.remove(doc_id);
         Ok(())
     }
 
@@ -94,18 +94,18 @@ impl FlowyDocumentManager {
         })
     }
 
-    pub async fn save_document<T: AsRef<str>>(&self, doc_id: T, revisions: RepeatedRevision) -> FlowyResult<()> {
+    pub async fn receive_revisions<T: AsRef<str>>(&self, doc_id: T, revisions: RepeatedRevision) -> FlowyResult<()> {
         let doc_id = doc_id.as_ref().to_owned();
-        let db_pool = self.user.db_pool()?;
+        let db_pool = self.document_user.db_pool()?;
         let rev_manager = self.make_rev_manager(&doc_id, db_pool)?;
         let _ = rev_manager.reset_object(revisions).await?;
         Ok(())
     }
 
-    pub async fn did_receive_ws_data(&self, data: Bytes) {
+    pub async fn receive_ws_data(&self, data: Bytes) {
         let result: Result<ServerRevisionWSData, protobuf::ProtobufError> = data.try_into();
         match result {
-            Ok(data) => match self.ws_receivers.get(&data.object_id) {
+            Ok(data) => match self.ws_data_receivers.get(&data.object_id) {
                 None => tracing::error!("Can't find any source handler for {:?}-{:?}", data.object_id, data.ty),
                 Some(handler) => match handler.receive_ws_data(data).await {
                     Ok(_) => {}
@@ -117,19 +117,13 @@ impl FlowyDocumentManager {
             }
         }
     }
-
-    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 FlowyDocumentManager {
     async fn get_editor(&self, doc_id: &str) -> FlowyResult<Arc<ClientDocumentEditor>> {
-        match self.open_cache.get(doc_id) {
+        match self.document_handlers.get(doc_id) {
             None => {
-                let db_pool = self.user.db_pool()?;
+                let db_pool = self.document_user.db_pool()?;
                 self.make_editor(doc_id, db_pool).await
             }
             Some(editor) => Ok(editor),
@@ -141,35 +135,26 @@ impl FlowyDocumentManager {
         doc_id: &str,
         pool: Arc<ConnectionPool>,
     ) -> Result<Arc<ClientDocumentEditor>, FlowyError> {
-        let user = self.user.clone();
-        let token = self.user.token()?;
+        let user = self.document_user.clone();
+        let token = self.document_user.token()?;
         let rev_manager = self.make_rev_manager(doc_id, pool.clone())?;
-        let server = Arc::new(DocumentRevisionCloudServiceImpl {
+        let cloud_service = Arc::new(DocumentRevisionCloudServiceImpl {
             token,
             server: self.cloud_service.clone(),
         });
-        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);
+        let doc_editor =
+            ClientDocumentEditor::new(doc_id, user, rev_manager, self.rev_web_socket.clone(), cloud_service).await?;
+        self.ws_data_receivers
+            .insert(doc_id.to_string(), doc_editor.ws_handler());
+        self.document_handlers.insert(doc_id, &doc_editor);
         Ok(doc_editor)
     }
 
     fn make_rev_manager(&self, doc_id: &str, pool: Arc<ConnectionPool>) -> Result<RevisionManager, FlowyError> {
-        let user_id = self.user.user_id()?;
+        let user_id = self.document_user.user_id()?;
         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 DocumentRevisionCloudServiceImpl {
@@ -202,11 +187,11 @@ impl RevisionCloudService for DocumentRevisionCloudServiceImpl {
     }
 }
 
-pub struct OpenDocCache {
+pub struct DocumentEditorHandlers {
     inner: DashMap<String, Arc<ClientDocumentEditor>>,
 }
 
-impl OpenDocCache {
+impl DocumentEditorHandlers {
     fn new() -> Self {
         Self { inner: DashMap::new() }
     }

frontend/rust-lib/flowy-document/src/core/queue.rs → frontend/rust-lib/flowy-document/src/queue.rs

@@ -1,4 +1,5 @@
-use crate::{core::web_socket::EditorCommandReceiver, DocumentUser};
+use crate::web_socket::EditorCommandReceiver;
+use crate::DocumentUser;
 use async_stream::stream;
 use flowy_collaboration::util::make_delta_from_revisions;
 use flowy_collaboration::{

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

@@ -1,7 +1,4 @@
-use crate::{
-    core::{EditorCommand, DOCUMENT_SYNC_INTERVAL_IN_MILLIS},
-    DocumentWSReceiver,
-};
+use crate::{queue::EditorCommand, DocumentWSReceiver, DOCUMENT_SYNC_INTERVAL_IN_MILLIS};
 use async_trait::async_trait;
 use bytes::Bytes;
 use flowy_collaboration::{
@@ -31,7 +28,7 @@ pub(crate) async fn make_document_ws_manager(
     user_id: String,
     edit_cmd_tx: EditorCommandSender,
     rev_manager: Arc<RevisionManager>,
-    web_socket: Arc<dyn RevisionWebSocket>,
+    rev_web_socket: Arc<dyn RevisionWebSocket>,
 ) -> Arc<RevisionWebSocketManager> {
     let composite_sink_provider = Arc::new(CompositeWSSinkDataProvider::new(&doc_id, rev_manager.clone()));
     let resolve_target = Arc::new(DocumentRevisionResolveTarget { edit_cmd_tx });
@@ -50,7 +47,7 @@ pub(crate) async fn make_document_ws_manager(
     let ws_manager = Arc::new(RevisionWebSocketManager::new(
         "Document",
         &doc_id,
-        web_socket,
+        rev_web_socket,
         sink_provider,
         ws_stream_consumer,
         ping_duration,

frontend/rust-lib/flowy-document/tests/document/edit_script.rs

@@ -1,5 +1,6 @@
 use flowy_collaboration::entities::revision::RevisionState;
-use flowy_document::core::{ClientDocumentEditor, DOCUMENT_SYNC_INTERVAL_IN_MILLIS};
+use flowy_document::editor::ClientDocumentEditor;
+use flowy_document::DOCUMENT_SYNC_INTERVAL_IN_MILLIS;
 use flowy_test::{helper::ViewTest, FlowySDKTest};
 use lib_ot::{core::Interval, rich_text::RichTextDelta};
 use std::sync::Arc;
@@ -67,8 +68,8 @@ impl EditorTest {
                     return;
                 }
                 let next_revision = next_revision.unwrap();
-                let mut receiver = rev_manager.revision_ack_receiver();
-                let _ = receiver.recv().await;
+                let mut notify = rev_manager.ack_notify();
+                let _ = notify.recv().await;
                 assert_eq!(next_revision.rev_id, rev_id.unwrap());
             }
             EditorScript::AssertJson(expected) => {

frontend/rust-lib/flowy-folder/src/services/view/controller.rs

@@ -73,7 +73,7 @@ impl ViewController {
             Revision::initial_revision(&user_id, &params.view_id, delta_data).into();
         let _ = self
             .document_manager
-            .save_document(&params.view_id, repeated_revision)
+            .receive_revisions(&params.view_id, repeated_revision)
             .await?;
         let view = self.create_view_on_server(params).await?;
         let _ = self.create_view_on_local(view.clone()).await?;
@@ -94,7 +94,10 @@ impl ViewController {
         let delta_data = Bytes::from(view_data);
         let user_id = self.user.user_id()?;
         let repeated_revision: RepeatedRevision = Revision::initial_revision(&user_id, view_id, delta_data).into();
-        let _ = self.document_manager.save_document(view_id, repeated_revision).await?;
+        let _ = self
+            .document_manager
+            .receive_revisions(view_id, repeated_revision)
+            .await?;
         Ok(())
     }
 

frontend/rust-lib/flowy-folder/tests/workspace/script.rs

@@ -203,8 +203,8 @@ impl FolderTest {
                 }
                 let next_revision = next_revision
                     .unwrap_or_else(|| panic!("Expected Next revision is {}, but receive None", rev_id.unwrap()));
-                let mut receiver = rev_manager.revision_ack_receiver();
-                let _ = receiver.recv().await;
+                let mut notify = rev_manager.ack_notify();
+                let _ = notify.recv().await;
                 assert_eq!(next_revision.rev_id, rev_id.unwrap());
             }
         }

frontend/rust-lib/flowy-sdk/src/deps_resolve/document_deps.rs

@@ -98,7 +98,7 @@ impl WSMessageReceiver for DocumentWSMessageReceiverImpl {
     fn receive_message(&self, msg: WebSocketRawMessage) {
         let handler = self.0.clone();
         tokio::spawn(async move {
-            handler.did_receive_ws_data(Bytes::from(msg.data)).await;
+            handler.receive_ws_data(Bytes::from(msg.data)).await;
         });
     }
 }