|
@@ -1,5 +1,6 @@
|
|
|
+use crate::entities::{EditParams, EditPayloadPB};
|
|
|
use crate::queue::TextBlockRevisionCompactor;
|
|
|
-use crate::{editor::TextBlockEditor, errors::FlowyError, BlockCloudService};
|
|
|
+use crate::{editor::TextBlockEditor, errors::FlowyError, TextEditorCloudService};
|
|
|
use bytes::Bytes;
|
|
|
use dashmap::DashMap;
|
|
|
use flowy_database::ConnectionPool;
|
|
@@ -16,30 +17,30 @@ use flowy_sync::entities::{
|
|
|
use lib_infra::future::FutureResult;
|
|
|
use std::{convert::TryInto, sync::Arc};
|
|
|
|
|
|
-pub trait TextBlockUser: Send + Sync {
|
|
|
+pub trait TextEditorUser: Send + Sync {
|
|
|
fn user_dir(&self) -> Result<String, FlowyError>;
|
|
|
fn user_id(&self) -> Result<String, FlowyError>;
|
|
|
fn token(&self) -> Result<String, FlowyError>;
|
|
|
fn db_pool(&self) -> Result<Arc<ConnectionPool>, FlowyError>;
|
|
|
}
|
|
|
|
|
|
-pub struct TextBlockManager {
|
|
|
- cloud_service: Arc<dyn BlockCloudService>,
|
|
|
+pub struct TextEditorManager {
|
|
|
+ cloud_service: Arc<dyn TextEditorCloudService>,
|
|
|
rev_web_socket: Arc<dyn RevisionWebSocket>,
|
|
|
- editor_map: Arc<TextBlockEditorMap>,
|
|
|
- user: Arc<dyn TextBlockUser>,
|
|
|
+ editor_map: Arc<TextEditorMap>,
|
|
|
+ user: Arc<dyn TextEditorUser>,
|
|
|
}
|
|
|
|
|
|
-impl TextBlockManager {
|
|
|
+impl TextEditorManager {
|
|
|
pub fn new(
|
|
|
- cloud_service: Arc<dyn BlockCloudService>,
|
|
|
- text_block_user: Arc<dyn TextBlockUser>,
|
|
|
+ cloud_service: Arc<dyn TextEditorCloudService>,
|
|
|
+ text_block_user: Arc<dyn TextEditorUser>,
|
|
|
rev_web_socket: Arc<dyn RevisionWebSocket>,
|
|
|
) -> Self {
|
|
|
Self {
|
|
|
cloud_service,
|
|
|
rev_web_socket,
|
|
|
- editor_map: Arc::new(TextBlockEditorMap::new()),
|
|
|
+ editor_map: Arc::new(TextEditorMap::new()),
|
|
|
user: text_block_user,
|
|
|
}
|
|
|
}
|
|
@@ -50,45 +51,47 @@ impl TextBlockManager {
|
|
|
Ok(())
|
|
|
}
|
|
|
|
|
|
- #[tracing::instrument(level = "trace", skip(self, block_id), fields(block_id), err)]
|
|
|
- pub async fn open_block<T: AsRef<str>>(&self, block_id: T) -> Result<Arc<TextBlockEditor>, FlowyError> {
|
|
|
- let block_id = block_id.as_ref();
|
|
|
- tracing::Span::current().record("block_id", &block_id);
|
|
|
- self.get_block_editor(block_id).await
|
|
|
+ #[tracing::instrument(level = "trace", skip(self, editor_id), fields(editor_id), err)]
|
|
|
+ pub async fn open_text_editor<T: AsRef<str>>(&self, editor_id: T) -> Result<Arc<TextBlockEditor>, FlowyError> {
|
|
|
+ let editor_id = editor_id.as_ref();
|
|
|
+ tracing::Span::current().record("editor_id", &editor_id);
|
|
|
+ self.get_text_editor(editor_id).await
|
|
|
}
|
|
|
|
|
|
- #[tracing::instrument(level = "trace", skip(self, block_id), fields(block_id), err)]
|
|
|
- pub fn close_block<T: AsRef<str>>(&self, block_id: T) -> Result<(), FlowyError> {
|
|
|
- let block_id = block_id.as_ref();
|
|
|
- tracing::Span::current().record("block_id", &block_id);
|
|
|
- self.editor_map.remove(block_id);
|
|
|
+ #[tracing::instrument(level = "trace", skip(self, editor_id), fields(editor_id), err)]
|
|
|
+ pub fn close_text_editor<T: AsRef<str>>(&self, editor_id: T) -> Result<(), FlowyError> {
|
|
|
+ let editor_id = editor_id.as_ref();
|
|
|
+ tracing::Span::current().record("editor_id", &editor_id);
|
|
|
+ self.editor_map.remove(editor_id);
|
|
|
Ok(())
|
|
|
}
|
|
|
|
|
|
- #[tracing::instrument(level = "debug", skip(self, doc_id), fields(doc_id), err)]
|
|
|
- pub fn delete_block<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.editor_map.remove(doc_id);
|
|
|
- Ok(())
|
|
|
- }
|
|
|
-
|
|
|
- #[tracing::instrument(level = "debug", skip(self, delta), fields(doc_id = %delta.block_id), err)]
|
|
|
+ #[tracing::instrument(level = "debug", skip(self, delta), err)]
|
|
|
pub async fn receive_local_delta(&self, delta: TextBlockDeltaPB) -> Result<TextBlockDeltaPB, FlowyError> {
|
|
|
- let editor = self.get_block_editor(&delta.block_id).await?;
|
|
|
+ let editor = self.get_text_editor(&delta.text_block_id).await?;
|
|
|
let _ = editor.compose_local_delta(Bytes::from(delta.delta_str)).await?;
|
|
|
- let document_json = editor.delta_str().await?;
|
|
|
+ let delta_str = editor.delta_str().await?;
|
|
|
Ok(TextBlockDeltaPB {
|
|
|
- block_id: delta.block_id.clone(),
|
|
|
- delta_str: document_json,
|
|
|
+ text_block_id: delta.text_block_id.clone(),
|
|
|
+ delta_str,
|
|
|
})
|
|
|
}
|
|
|
|
|
|
- pub async fn create_block<T: AsRef<str>>(&self, doc_id: T, revisions: RepeatedRevision) -> FlowyResult<()> {
|
|
|
- let doc_id = doc_id.as_ref().to_owned();
|
|
|
+ pub async fn apply_edit(&self, params: EditParams) -> FlowyResult<()> {
|
|
|
+ let editor = self.get_text_editor(¶ms.text_block_id).await?;
|
|
|
+ let _ = editor.compose_local_delta(Bytes::from(params.delta)).await?;
|
|
|
+ Ok(())
|
|
|
+ }
|
|
|
+
|
|
|
+ pub async fn create_text_block<T: AsRef<str>>(
|
|
|
+ &self,
|
|
|
+ text_block_id: T,
|
|
|
+ revisions: RepeatedRevision,
|
|
|
+ ) -> FlowyResult<()> {
|
|
|
+ let doc_id = text_block_id.as_ref().to_owned();
|
|
|
let db_pool = self.user.db_pool()?;
|
|
|
// Maybe we could save the block to disk without creating the RevisionManager
|
|
|
- let rev_manager = self.make_rev_manager(&doc_id, db_pool)?;
|
|
|
+ let rev_manager = self.make_text_block_rev_manager(&doc_id, db_pool)?;
|
|
|
let _ = rev_manager.reset_object(revisions).await?;
|
|
|
Ok(())
|
|
|
}
|
|
@@ -110,26 +113,26 @@ impl TextBlockManager {
|
|
|
}
|
|
|
}
|
|
|
|
|
|
-impl TextBlockManager {
|
|
|
- async fn get_block_editor(&self, block_id: &str) -> FlowyResult<Arc<TextBlockEditor>> {
|
|
|
+impl TextEditorManager {
|
|
|
+ async fn get_text_editor(&self, block_id: &str) -> FlowyResult<Arc<TextBlockEditor>> {
|
|
|
match self.editor_map.get(block_id) {
|
|
|
None => {
|
|
|
let db_pool = self.user.db_pool()?;
|
|
|
- self.make_text_block_editor(block_id, db_pool).await
|
|
|
+ self.make_text_editor(block_id, db_pool).await
|
|
|
}
|
|
|
Some(editor) => Ok(editor),
|
|
|
}
|
|
|
}
|
|
|
|
|
|
#[tracing::instrument(level = "trace", skip(self, pool), err)]
|
|
|
- async fn make_text_block_editor(
|
|
|
+ async fn make_text_editor(
|
|
|
&self,
|
|
|
block_id: &str,
|
|
|
pool: Arc<ConnectionPool>,
|
|
|
) -> Result<Arc<TextBlockEditor>, FlowyError> {
|
|
|
let user = self.user.clone();
|
|
|
let token = self.user.token()?;
|
|
|
- let rev_manager = self.make_rev_manager(block_id, pool.clone())?;
|
|
|
+ let rev_manager = self.make_text_block_rev_manager(block_id, pool.clone())?;
|
|
|
let cloud_service = Arc::new(TextBlockRevisionCloudService {
|
|
|
token,
|
|
|
server: self.cloud_service.clone(),
|
|
@@ -140,7 +143,11 @@ impl TextBlockManager {
|
|
|
Ok(doc_editor)
|
|
|
}
|
|
|
|
|
|
- fn make_rev_manager(&self, doc_id: &str, pool: Arc<ConnectionPool>) -> Result<RevisionManager, FlowyError> {
|
|
|
+ fn make_text_block_rev_manager(
|
|
|
+ &self,
|
|
|
+ doc_id: &str,
|
|
|
+ pool: Arc<ConnectionPool>,
|
|
|
+ ) -> Result<RevisionManager, FlowyError> {
|
|
|
let user_id = self.user.user_id()?;
|
|
|
let disk_cache = SQLiteTextBlockRevisionPersistence::new(&user_id, pool.clone());
|
|
|
let rev_persistence = RevisionPersistence::new(&user_id, doc_id, disk_cache);
|
|
@@ -161,7 +168,7 @@ impl TextBlockManager {
|
|
|
|
|
|
struct TextBlockRevisionCloudService {
|
|
|
token: String,
|
|
|
- server: Arc<dyn BlockCloudService>,
|
|
|
+ server: Arc<dyn TextEditorCloudService>,
|
|
|
}
|
|
|
|
|
|
impl RevisionCloudService for TextBlockRevisionCloudService {
|
|
@@ -173,7 +180,7 @@ impl RevisionCloudService for TextBlockRevisionCloudService {
|
|
|
let user_id = user_id.to_string();
|
|
|
|
|
|
FutureResult::new(async move {
|
|
|
- match server.read_block(&token, params).await? {
|
|
|
+ match server.read_text_block(&token, params).await? {
|
|
|
None => Err(FlowyError::record_not_found().context("Remote doesn't have this document")),
|
|
|
Some(doc) => {
|
|
|
let delta_data = Bytes::from(doc.text.clone());
|
|
@@ -193,36 +200,36 @@ impl RevisionCloudService for TextBlockRevisionCloudService {
|
|
|
}
|
|
|
}
|
|
|
|
|
|
-pub struct TextBlockEditorMap {
|
|
|
+pub struct TextEditorMap {
|
|
|
inner: DashMap<String, Arc<TextBlockEditor>>,
|
|
|
}
|
|
|
|
|
|
-impl TextBlockEditorMap {
|
|
|
+impl TextEditorMap {
|
|
|
fn new() -> Self {
|
|
|
Self { inner: DashMap::new() }
|
|
|
}
|
|
|
|
|
|
- pub(crate) fn insert(&self, block_id: &str, doc: &Arc<TextBlockEditor>) {
|
|
|
- if self.inner.contains_key(block_id) {
|
|
|
- log::warn!("Doc:{} already exists in cache", block_id);
|
|
|
+ pub(crate) fn insert(&self, editor_id: &str, doc: &Arc<TextBlockEditor>) {
|
|
|
+ if self.inner.contains_key(editor_id) {
|
|
|
+ log::warn!("Doc:{} already exists in cache", editor_id);
|
|
|
}
|
|
|
- self.inner.insert(block_id.to_string(), doc.clone());
|
|
|
+ self.inner.insert(editor_id.to_string(), doc.clone());
|
|
|
}
|
|
|
|
|
|
- pub(crate) fn get(&self, block_id: &str) -> Option<Arc<TextBlockEditor>> {
|
|
|
- Some(self.inner.get(block_id)?.clone())
|
|
|
+ pub(crate) fn get(&self, editor_id: &str) -> Option<Arc<TextBlockEditor>> {
|
|
|
+ Some(self.inner.get(editor_id)?.clone())
|
|
|
}
|
|
|
|
|
|
- pub(crate) fn remove(&self, block_id: &str) {
|
|
|
- if let Some(editor) = self.get(block_id) {
|
|
|
+ pub(crate) fn remove(&self, editor_id: &str) {
|
|
|
+ if let Some(editor) = self.get(editor_id) {
|
|
|
editor.stop()
|
|
|
}
|
|
|
- self.inner.remove(block_id);
|
|
|
+ self.inner.remove(editor_id);
|
|
|
}
|
|
|
}
|
|
|
|
|
|
#[tracing::instrument(level = "trace", skip(web_socket, handlers))]
|
|
|
-fn listen_ws_state_changed(web_socket: Arc<dyn RevisionWebSocket>, handlers: Arc<TextBlockEditorMap>) {
|
|
|
+fn listen_ws_state_changed(web_socket: Arc<dyn RevisionWebSocket>, handlers: Arc<TextEditorMap>) {
|
|
|
tokio::spawn(async move {
|
|
|
let mut notify = web_socket.subscribe_state_changed().await;
|
|
|
while let Ok(state) = notify.recv().await {
|