add folder sync test

This commit is contained in:
appflowy 2022-01-22 18:48:43 +08:00
parent 1e66aae051
commit ccb7d0181f
66 changed files with 1190 additions and 786 deletions

View File

@ -13,7 +13,7 @@ use crate::services::{
folder::ws_receiver::make_folder_ws_receiver,
};
use flowy_collaboration::server_document::ServerDocumentManager;
use lib_ws::WSModule;
use lib_ws::WSChannel;
use sqlx::PgPool;
use std::sync::Arc;
@ -37,10 +37,10 @@ impl AppContext {
let document_manager = Arc::new(ServerDocumentManager::new(document_persistence));
let document_ws_receiver = make_document_ws_receiver(flowy_persistence.clone(), document_manager.clone());
ws_receivers.set(WSModule::Doc, document_ws_receiver);
ws_receivers.set(WSChannel::Document, document_ws_receiver);
let folder_ws_receiver = make_folder_ws_receiver(flowy_persistence.clone());
ws_receivers.set(WSModule::Folder, folder_ws_receiver);
ws_receivers.set(WSChannel::Folder, folder_ws_receiver);
AppContext {
ws_server,

View File

@ -7,6 +7,7 @@ use actix_rt::task::spawn_blocking;
use async_stream::stream;
use backend_service::errors::{internal_error, Result, ServerError};
use crate::services::web_socket::revision_data_to_ws_message;
use flowy_collaboration::{
protobuf::{
ClientRevisionWSData as ClientRevisionWSDataPB,
@ -17,6 +18,7 @@ use flowy_collaboration::{
synchronizer::{RevisionSyncResponse, RevisionUser},
};
use futures::stream::StreamExt;
use lib_ws::WSChannel;
use std::sync::Arc;
use tokio::sync::{mpsc, oneshot};
@ -135,15 +137,15 @@ impl RevisionUser for DocumentRevisionUser {
fn receive(&self, resp: RevisionSyncResponse) {
let result = match resp {
RevisionSyncResponse::Pull(data) => {
let msg: WebSocketMessage = data.into();
let msg: WebSocketMessage = revision_data_to_ws_message(data, WSChannel::Document);
self.socket.try_send(msg).map_err(internal_error)
},
RevisionSyncResponse::Push(data) => {
let msg: WebSocketMessage = data.into();
let msg: WebSocketMessage = revision_data_to_ws_message(data, WSChannel::Document);
self.socket.try_send(msg).map_err(internal_error)
},
RevisionSyncResponse::Ack(data) => {
let msg: WebSocketMessage = data.into();
let msg: WebSocketMessage = revision_data_to_ws_message(data, WSChannel::Document);
self.socket.try_send(msg).map_err(internal_error)
},
};

View File

@ -1,11 +1,12 @@
use crate::{
context::FlowyPersistence,
services::web_socket::{entities::Socket, WSClientData, WSUser, WebSocketMessage},
services::web_socket::{entities::Socket, revision_data_to_ws_message, WSClientData, WSUser, WebSocketMessage},
util::serde_ext::parse_from_bytes,
};
use actix_rt::task::spawn_blocking;
use async_stream::stream;
use backend_service::errors::{internal_error, Result};
use flowy_collaboration::{
protobuf::{
ClientRevisionWSData as ClientRevisionWSDataPB,
@ -14,6 +15,7 @@ use flowy_collaboration::{
synchronizer::{RevisionSyncResponse, RevisionUser},
};
use futures::stream::StreamExt;
use lib_ws::WSChannel;
use std::sync::Arc;
use tokio::sync::{mpsc, oneshot};
@ -111,15 +113,15 @@ impl RevisionUser for FolderRevisionUser {
fn receive(&self, resp: RevisionSyncResponse) {
let result = match resp {
RevisionSyncResponse::Pull(data) => {
let msg: WebSocketMessage = data.into();
let msg: WebSocketMessage = revision_data_to_ws_message(data, WSChannel::Folder);
self.socket.try_send(msg).map_err(internal_error)
},
RevisionSyncResponse::Push(data) => {
let msg: WebSocketMessage = data.into();
let msg: WebSocketMessage = revision_data_to_ws_message(data, WSChannel::Folder);
self.socket.try_send(msg).map_err(internal_error)
},
RevisionSyncResponse::Ack(data) => {
let msg: WebSocketMessage = data.into();
let msg: WebSocketMessage = revision_data_to_ws_message(data, WSChannel::Folder);
self.socket.try_send(msg).map_err(internal_error)
},
};

View File

@ -1,7 +1,7 @@
use actix::Message;
use bytes::Bytes;
use flowy_collaboration::entities::ws_data::{ClientRevisionWSData, ServerRevisionWSData};
use lib_ws::{WSModule, WebSocketRawMessage};
use flowy_collaboration::entities::ws_data::ServerRevisionWSData;
use lib_ws::{WSChannel, WebSocketRawMessage};
use std::convert::TryInto;
#[derive(Debug, Message, Clone)]
@ -14,27 +14,12 @@ impl std::ops::Deref for WebSocketMessage {
fn deref(&self) -> &Self::Target { &self.0 }
}
impl std::convert::From<ClientRevisionWSData> for WebSocketMessage {
fn from(data: ClientRevisionWSData) -> Self {
let bytes: Bytes = data.try_into().unwrap();
let msg = WebSocketRawMessage {
module: WSModule::Doc,
data: bytes.to_vec(),
};
let bytes: Bytes = msg.try_into().unwrap();
WebSocketMessage(bytes)
}
}
impl std::convert::From<ServerRevisionWSData> for WebSocketMessage {
fn from(data: ServerRevisionWSData) -> Self {
let bytes: Bytes = data.try_into().unwrap();
let msg = WebSocketRawMessage {
module: WSModule::Doc,
data: bytes.to_vec(),
};
let bytes: Bytes = msg.try_into().unwrap();
WebSocketMessage(bytes)
}
pub fn revision_data_to_ws_message(data: ServerRevisionWSData, channel: WSChannel) -> WebSocketMessage {
let bytes: Bytes = data.try_into().unwrap();
let msg = WebSocketRawMessage {
channel,
data: bytes.to_vec(),
};
let bytes: Bytes = msg.try_into().unwrap();
WebSocketMessage(bytes)
}

View File

@ -11,7 +11,7 @@ use actix::*;
use actix_web::web::Data;
use actix_web_actors::{ws, ws::Message::Text};
use bytes::Bytes;
use lib_ws::{WSModule, WebSocketRawMessage};
use lib_ws::{WSChannel, WebSocketRawMessage};
use std::{collections::HashMap, convert::TryFrom, sync::Arc, time::Instant};
pub trait WebSocketReceiver: Send + Sync {
@ -19,7 +19,7 @@ pub trait WebSocketReceiver: Send + Sync {
}
pub struct WebSocketReceivers {
inner: HashMap<WSModule, Arc<dyn WebSocketReceiver>>,
inner: HashMap<WSChannel, Arc<dyn WebSocketReceiver>>,
}
impl std::default::Default for WebSocketReceivers {
@ -29,11 +29,11 @@ impl std::default::Default for WebSocketReceivers {
impl WebSocketReceivers {
pub fn new() -> Self { WebSocketReceivers::default() }
pub fn set(&mut self, source: WSModule, receiver: Arc<dyn WebSocketReceiver>) {
pub fn set(&mut self, source: WSChannel, receiver: Arc<dyn WebSocketReceiver>) {
self.inner.insert(source, receiver);
}
pub fn get(&self, source: &WSModule) -> Option<Arc<dyn WebSocketReceiver>> { self.inner.get(source).cloned() }
pub fn get(&self, source: &WSChannel) -> Option<Arc<dyn WebSocketReceiver>> { self.inner.get(source).cloned() }
}
#[derive(Debug)]
@ -86,9 +86,9 @@ impl WSClient {
fn handle_binary_message(&self, bytes: Bytes, socket: Socket) {
// TODO: ok to unwrap?
let message: WebSocketRawMessage = WebSocketRawMessage::try_from(bytes).unwrap();
match self.ws_receivers.get(&message.module) {
match self.ws_receivers.get(&message.channel) {
None => {
log::error!("Can't find the receiver for {:?}", message.module);
log::error!("Can't find the receiver for {:?}", message.channel);
},
Some(handler) => {
let client_data = WSClientData {

View File

@ -1,6 +1,6 @@
#![allow(clippy::all)]
use crate::util::helper::{ViewTest, *};
use crate::util::helper::{BackendViewTest, *};
use flowy_collaboration::{
client_document::{ClientDocument, PlainDoc},
entities::{
@ -17,13 +17,13 @@ use flowy_core_data_model::entities::{
#[actix_rt::test]
async fn workspace_create() {
let test = WorkspaceTest::new().await;
let test = BackendWorkspaceTest::new().await;
tracing::info!("{:?}", test.workspace);
}
#[actix_rt::test]
async fn workspace_read() {
let test = WorkspaceTest::new().await;
let test = BackendWorkspaceTest::new().await;
let read_params = WorkspaceId::new(Some(test.workspace.id.clone()));
let repeated_workspace = test.server.read_workspaces(read_params).await;
tracing::info!("{:?}", repeated_workspace);
@ -31,7 +31,7 @@ async fn workspace_read() {
#[actix_rt::test]
async fn workspace_read_with_belongs() {
let test = WorkspaceTest::new().await;
let test = BackendWorkspaceTest::new().await;
let _ = test.create_app().await;
let _ = test.create_app().await;
@ -45,7 +45,7 @@ async fn workspace_read_with_belongs() {
#[actix_rt::test]
async fn workspace_update() {
let test = WorkspaceTest::new().await;
let test = BackendWorkspaceTest::new().await;
let new_name = "rename workspace name";
let new_desc = "rename workspace description";
@ -65,7 +65,7 @@ async fn workspace_update() {
#[actix_rt::test]
async fn workspace_delete() {
let test = WorkspaceTest::new().await;
let test = BackendWorkspaceTest::new().await;
let delete_params = WorkspaceId {
workspace_id: Some(test.workspace.id.clone()),
};
@ -78,20 +78,20 @@ async fn workspace_delete() {
#[actix_rt::test]
async fn app_create() {
let test = AppTest::new().await;
let test = BackendAppTest::new().await;
tracing::info!("{:?}", test.app);
}
#[actix_rt::test]
async fn app_read() {
let test = AppTest::new().await;
let test = BackendAppTest::new().await;
let read_params = AppId::new(&test.app.id);
assert_eq!(test.server.read_app(read_params).await.is_some(), true);
}
#[actix_rt::test]
async fn app_read_with_belongs() {
let test = AppTest::new().await;
let test = BackendAppTest::new().await;
let _ = create_test_view(&test.server, &test.app.id).await;
let _ = create_test_view(&test.server, &test.app.id).await;
@ -103,7 +103,7 @@ async fn app_read_with_belongs() {
#[actix_rt::test]
async fn app_read_with_belongs_in_trash() {
let test = AppTest::new().await;
let test = BackendAppTest::new().await;
let _ = create_test_view(&test.server, &test.app.id).await;
let view = create_test_view(&test.server, &test.app.id).await;
@ -117,7 +117,7 @@ async fn app_read_with_belongs_in_trash() {
#[actix_rt::test]
async fn app_update() {
let test = AppTest::new().await;
let test = BackendAppTest::new().await;
let new_name = "flowy";
@ -131,7 +131,7 @@ async fn app_update() {
#[actix_rt::test]
async fn app_delete() {
let test = AppTest::new().await;
let test = BackendAppTest::new().await;
let delete_params = AppId {
app_id: test.app.id.clone(),
@ -143,13 +143,13 @@ async fn app_delete() {
#[actix_rt::test]
async fn view_create() {
let test = ViewTest::new().await;
let test = BackendViewTest::new().await;
tracing::info!("{:?}", test.view);
}
#[actix_rt::test]
async fn view_update() {
let test = ViewTest::new().await;
let test = BackendViewTest::new().await;
let new_name = "name view name";
// update
@ -164,7 +164,7 @@ async fn view_update() {
#[actix_rt::test]
async fn view_delete() {
let test = ViewTest::new().await;
let test = BackendViewTest::new().await;
test.server.create_view_trash(&test.view.id).await;
let trash_ids = test
@ -185,7 +185,7 @@ async fn view_delete() {
#[actix_rt::test]
async fn trash_delete() {
let test = ViewTest::new().await;
let test = BackendViewTest::new().await;
test.server.create_view_trash(&test.view.id).await;
let identifier = TrashId {
@ -199,7 +199,7 @@ async fn trash_delete() {
#[actix_rt::test]
async fn trash_delete_all() {
let test = ViewTest::new().await;
let test = BackendViewTest::new().await;
test.server.create_view_trash(&test.view.id).await;
test.server.delete_view_trash(RepeatedTrashId::all()).await;
@ -226,7 +226,7 @@ async fn workspace_list_read() {
#[actix_rt::test]
async fn doc_read() {
let test = ViewTest::new().await;
let test = BackendViewTest::new().await;
let params = DocumentId {
doc_id: test.view.id.clone(),
};
@ -268,7 +268,7 @@ async fn doc_create() {
#[actix_rt::test]
async fn doc_delete() {
let test = ViewTest::new().await;
let test = BackendViewTest::new().await;
let delete_params = RepeatedViewId {
items: vec![test.view.id.clone()],
};

View File

@ -78,7 +78,7 @@ impl ScriptContext {
async fn open_doc(&mut self) {
let doc_id = self.doc_id.clone();
let edit_context = self.client_sdk.document_ctx.controller.open_document(doc_id).await.unwrap();
let edit_context = self.client_sdk.document_manager.open_document(doc_id).await.unwrap();
self.client_editor = Some(edit_context);
}

View File

@ -11,7 +11,7 @@ use flowy_collaboration::{
client_document::default::initial_delta_string,
entities::document_info::{CreateDocParams, DocumentId, DocumentInfo},
};
use flowy_core_data_model::entities::prelude::*;
use flowy_core_data_model::entities::{app::*, trash::*, view::*, workspace::*};
use flowy_net::http_server::{
core::*,
document::{create_document_request, read_document_request},
@ -327,12 +327,12 @@ pub async fn create_test_view(application: &TestUserServer, app_id: &str) -> Vie
app
}
pub struct WorkspaceTest {
pub struct BackendWorkspaceTest {
pub server: TestUserServer,
pub workspace: Workspace,
}
impl WorkspaceTest {
impl BackendWorkspaceTest {
pub async fn new() -> Self {
let server = TestUserServer::new().await;
let workspace = create_test_workspace(&server).await;
@ -342,13 +342,13 @@ impl WorkspaceTest {
pub async fn create_app(&self) -> App { create_test_app(&self.server, &self.workspace.id).await }
}
pub struct AppTest {
pub struct BackendAppTest {
pub server: TestUserServer,
pub workspace: Workspace,
pub app: App,
}
impl AppTest {
impl BackendAppTest {
pub async fn new() -> Self {
let server = TestUserServer::new().await;
let workspace = create_test_workspace(&server).await;
@ -357,14 +357,14 @@ impl AppTest {
}
}
pub struct ViewTest {
pub struct BackendViewTest {
pub server: TestUserServer,
pub workspace: Workspace,
pub app: App,
pub view: View,
}
impl ViewTest {
impl BackendViewTest {
pub async fn new() -> Self {
let server = TestUserServer::new().await;
let workspace = create_test_workspace(&server).await;

View File

@ -1,6 +1,6 @@
import 'dart:async';
import 'package:dartz/dartz.dart';
import 'package:flowy_sdk/protobuf/flowy-collaboration/doc.pb.dart';
import 'package:flowy_sdk/protobuf/flowy-collaboration/document_info.pb.dart';
import 'package:flowy_sdk/protobuf/flowy-error/errors.pb.dart';
abstract class IDoc {

View File

@ -4,7 +4,7 @@ import 'dart:typed_data';
import 'package:dartz/dartz.dart';
import 'package:app_flowy/workspace/domain/i_doc.dart';
import 'package:app_flowy/workspace/infrastructure/repos/doc_repo.dart';
import 'package:flowy_sdk/protobuf/flowy-collaboration/doc.pb.dart';
import 'package:flowy_sdk/protobuf/flowy-collaboration/document_info.pb.dart';
import 'package:flowy_sdk/protobuf/flowy-error/errors.pb.dart';
class IDocImpl extends IDoc {

View File

@ -1,6 +1,6 @@
import 'package:dartz/dartz.dart';
import 'package:flowy_sdk/dispatch/dispatch.dart';
import 'package:flowy_sdk/protobuf/flowy-collaboration/doc.pb.dart';
import 'package:flowy_sdk/protobuf/flowy-collaboration/document_info.pb.dart';
import 'package:flowy_sdk/protobuf/flowy-core-data-model/view.pb.dart';
import 'package:flowy_sdk/protobuf/flowy-error/errors.pb.dart';
@ -12,7 +12,7 @@ class DocRepository {
Future<Either<DocumentDelta, FlowyError>> readDoc() {
final request = QueryViewRequest(viewIds: [docId]);
return WorkspaceEventOpenView(request).send();
return WorkspaceEventOpenDocument(request).send();
}
Future<Either<DocumentDelta, FlowyError>> composeDelta({required String data}) {

View File

@ -268,13 +268,13 @@ class WorkspaceEventCopyLink {
}
}
class WorkspaceEventOpenView {
class WorkspaceEventOpenDocument {
QueryViewRequest request;
WorkspaceEventOpenView(this.request);
WorkspaceEventOpenDocument(this.request);
Future<Either<DocumentDelta, FlowyError>> send() {
final request = FFIRequest.create()
..event = WorkspaceEvent.OpenView.toString()
..event = WorkspaceEvent.OpenDocument.toString()
..payload = requestToBytes(this.request);
return Dispatch.asyncRequest(request)
@ -350,12 +350,12 @@ class WorkspaceEventDeleteTrash {
}
}
class WorkspaceEventRestoreAll {
WorkspaceEventRestoreAll();
class WorkspaceEventRestoreAllTrash {
WorkspaceEventRestoreAllTrash();
Future<Either<Unit, FlowyError>> send() {
final request = FFIRequest.create()
..event = WorkspaceEvent.RestoreAll.toString();
..event = WorkspaceEvent.RestoreAllTrash.toString();
return Dispatch.asyncRequest(request).then((bytesResult) => bytesResult.fold(
(bytes) => left(unit),
@ -364,12 +364,12 @@ class WorkspaceEventRestoreAll {
}
}
class WorkspaceEventDeleteAll {
WorkspaceEventDeleteAll();
class WorkspaceEventDeleteAllTrash {
WorkspaceEventDeleteAllTrash();
Future<Either<Unit, FlowyError>> send() {
final request = FFIRequest.create()
..event = WorkspaceEvent.DeleteAll.toString();
..event = WorkspaceEvent.DeleteAllTrash.toString();
return Dispatch.asyncRequest(request).then((bytesResult) => bytesResult.fold(
(bytes) => left(unit),

View File

@ -26,13 +26,13 @@ class WorkspaceEvent extends $pb.ProtobufEnum {
static const WorkspaceEvent DeleteView = WorkspaceEvent._(204, const $core.bool.fromEnvironment('protobuf.omit_enum_names') ? '' : 'DeleteView');
static const WorkspaceEvent DuplicateView = WorkspaceEvent._(205, const $core.bool.fromEnvironment('protobuf.omit_enum_names') ? '' : 'DuplicateView');
static const WorkspaceEvent CopyLink = WorkspaceEvent._(206, const $core.bool.fromEnvironment('protobuf.omit_enum_names') ? '' : 'CopyLink');
static const WorkspaceEvent OpenView = WorkspaceEvent._(207, const $core.bool.fromEnvironment('protobuf.omit_enum_names') ? '' : 'OpenView');
static const WorkspaceEvent OpenDocument = WorkspaceEvent._(207, const $core.bool.fromEnvironment('protobuf.omit_enum_names') ? '' : 'OpenDocument');
static const WorkspaceEvent CloseView = WorkspaceEvent._(208, const $core.bool.fromEnvironment('protobuf.omit_enum_names') ? '' : 'CloseView');
static const WorkspaceEvent ReadTrash = WorkspaceEvent._(300, const $core.bool.fromEnvironment('protobuf.omit_enum_names') ? '' : 'ReadTrash');
static const WorkspaceEvent PutbackTrash = WorkspaceEvent._(301, const $core.bool.fromEnvironment('protobuf.omit_enum_names') ? '' : 'PutbackTrash');
static const WorkspaceEvent DeleteTrash = WorkspaceEvent._(302, const $core.bool.fromEnvironment('protobuf.omit_enum_names') ? '' : 'DeleteTrash');
static const WorkspaceEvent RestoreAll = WorkspaceEvent._(303, const $core.bool.fromEnvironment('protobuf.omit_enum_names') ? '' : 'RestoreAll');
static const WorkspaceEvent DeleteAll = WorkspaceEvent._(304, const $core.bool.fromEnvironment('protobuf.omit_enum_names') ? '' : 'DeleteAll');
static const WorkspaceEvent RestoreAllTrash = WorkspaceEvent._(303, const $core.bool.fromEnvironment('protobuf.omit_enum_names') ? '' : 'RestoreAllTrash');
static const WorkspaceEvent DeleteAllTrash = WorkspaceEvent._(304, const $core.bool.fromEnvironment('protobuf.omit_enum_names') ? '' : 'DeleteAllTrash');
static const WorkspaceEvent ApplyDocDelta = WorkspaceEvent._(400, const $core.bool.fromEnvironment('protobuf.omit_enum_names') ? '' : 'ApplyDocDelta');
static const WorkspaceEvent ExportDocument = WorkspaceEvent._(500, const $core.bool.fromEnvironment('protobuf.omit_enum_names') ? '' : 'ExportDocument');
@ -53,13 +53,13 @@ class WorkspaceEvent extends $pb.ProtobufEnum {
DeleteView,
DuplicateView,
CopyLink,
OpenView,
OpenDocument,
CloseView,
ReadTrash,
PutbackTrash,
DeleteTrash,
RestoreAll,
DeleteAll,
RestoreAllTrash,
DeleteAllTrash,
ApplyDocDelta,
ExportDocument,
];

View File

@ -28,17 +28,17 @@ const WorkspaceEvent$json = const {
const {'1': 'DeleteView', '2': 204},
const {'1': 'DuplicateView', '2': 205},
const {'1': 'CopyLink', '2': 206},
const {'1': 'OpenView', '2': 207},
const {'1': 'OpenDocument', '2': 207},
const {'1': 'CloseView', '2': 208},
const {'1': 'ReadTrash', '2': 300},
const {'1': 'PutbackTrash', '2': 301},
const {'1': 'DeleteTrash', '2': 302},
const {'1': 'RestoreAll', '2': 303},
const {'1': 'DeleteAll', '2': 304},
const {'1': 'RestoreAllTrash', '2': 303},
const {'1': 'DeleteAllTrash', '2': 304},
const {'1': 'ApplyDocDelta', '2': 400},
const {'1': 'ExportDocument', '2': 500},
],
};
/// Descriptor for `WorkspaceEvent`. Decode as a `google.protobuf.EnumDescriptorProto`.
final $typed_data.Uint8List workspaceEventDescriptor = $convert.base64Decode('Cg5Xb3Jrc3BhY2VFdmVudBITCg9DcmVhdGVXb3Jrc3BhY2UQABIUChBSZWFkQ3VyV29ya3NwYWNlEAESEgoOUmVhZFdvcmtzcGFjZXMQAhITCg9EZWxldGVXb3Jrc3BhY2UQAxIRCg1PcGVuV29ya3NwYWNlEAQSFQoRUmVhZFdvcmtzcGFjZUFwcHMQBRINCglDcmVhdGVBcHAQZRINCglEZWxldGVBcHAQZhILCgdSZWFkQXBwEGcSDQoJVXBkYXRlQXBwEGgSDwoKQ3JlYXRlVmlldxDJARINCghSZWFkVmlldxDKARIPCgpVcGRhdGVWaWV3EMsBEg8KCkRlbGV0ZVZpZXcQzAESEgoNRHVwbGljYXRlVmlldxDNARINCghDb3B5TGluaxDOARINCghPcGVuVmlldxDPARIOCglDbG9zZVZpZXcQ0AESDgoJUmVhZFRyYXNoEKwCEhEKDFB1dGJhY2tUcmFzaBCtAhIQCgtEZWxldGVUcmFzaBCuAhIPCgpSZXN0b3JlQWxsEK8CEg4KCURlbGV0ZUFsbBCwAhISCg1BcHBseURvY0RlbHRhEJADEhMKDkV4cG9ydERvY3VtZW50EPQD');
final $typed_data.Uint8List workspaceEventDescriptor = $convert.base64Decode('Cg5Xb3Jrc3BhY2VFdmVudBITCg9DcmVhdGVXb3Jrc3BhY2UQABIUChBSZWFkQ3VyV29ya3NwYWNlEAESEgoOUmVhZFdvcmtzcGFjZXMQAhITCg9EZWxldGVXb3Jrc3BhY2UQAxIRCg1PcGVuV29ya3NwYWNlEAQSFQoRUmVhZFdvcmtzcGFjZUFwcHMQBRINCglDcmVhdGVBcHAQZRINCglEZWxldGVBcHAQZhILCgdSZWFkQXBwEGcSDQoJVXBkYXRlQXBwEGgSDwoKQ3JlYXRlVmlldxDJARINCghSZWFkVmlldxDKARIPCgpVcGRhdGVWaWV3EMsBEg8KCkRlbGV0ZVZpZXcQzAESEgoNRHVwbGljYXRlVmlldxDNARINCghDb3B5TGluaxDOARIRCgxPcGVuRG9jdW1lbnQQzwESDgoJQ2xvc2VWaWV3ENABEg4KCVJlYWRUcmFzaBCsAhIRCgxQdXRiYWNrVHJhc2gQrQISEAoLRGVsZXRlVHJhc2gQrgISFAoPUmVzdG9yZUFsbFRyYXNoEK8CEhMKDkRlbGV0ZUFsbFRyYXNoELACEhIKDUFwcGx5RG9jRGVsdGEQkAMSEwoORXhwb3J0RG9jdW1lbnQQ9AM=');

View File

@ -15,19 +15,19 @@ export 'msg.pbenum.dart';
class WebSocketRawMessage extends $pb.GeneratedMessage {
static final $pb.BuilderInfo _i = $pb.BuilderInfo(const $core.bool.fromEnvironment('protobuf.omit_message_names') ? '' : 'WebSocketRawMessage', createEmptyInstance: create)
..e<WSModule>(1, const $core.bool.fromEnvironment('protobuf.omit_field_names') ? '' : 'module', $pb.PbFieldType.OE, defaultOrMaker: WSModule.Doc, valueOf: WSModule.valueOf, enumValues: WSModule.values)
..e<WSChannel>(1, const $core.bool.fromEnvironment('protobuf.omit_field_names') ? '' : 'channel', $pb.PbFieldType.OE, defaultOrMaker: WSChannel.Document, valueOf: WSChannel.valueOf, enumValues: WSChannel.values)
..a<$core.List<$core.int>>(2, const $core.bool.fromEnvironment('protobuf.omit_field_names') ? '' : 'data', $pb.PbFieldType.OY)
..hasRequiredFields = false
;
WebSocketRawMessage._() : super();
factory WebSocketRawMessage({
WSModule? module,
WSChannel? channel,
$core.List<$core.int>? data,
}) {
final _result = create();
if (module != null) {
_result.module = module;
if (channel != null) {
_result.channel = channel;
}
if (data != null) {
_result.data = data;
@ -56,13 +56,13 @@ class WebSocketRawMessage extends $pb.GeneratedMessage {
static WebSocketRawMessage? _defaultInstance;
@$pb.TagNumber(1)
WSModule get module => $_getN(0);
WSChannel get channel => $_getN(0);
@$pb.TagNumber(1)
set module(WSModule v) { setField(1, v); }
set channel(WSChannel v) { setField(1, v); }
@$pb.TagNumber(1)
$core.bool hasModule() => $_has(0);
$core.bool hasChannel() => $_has(0);
@$pb.TagNumber(1)
void clearModule() => clearField(1);
void clearChannel() => clearField(1);
@$pb.TagNumber(2)
$core.List<$core.int> get data => $_getN(1);

View File

@ -9,18 +9,18 @@
import 'dart:core' as $core;
import 'package:protobuf/protobuf.dart' as $pb;
class WSModule extends $pb.ProtobufEnum {
static const WSModule Doc = WSModule._(0, const $core.bool.fromEnvironment('protobuf.omit_enum_names') ? '' : 'Doc');
static const WSModule Folder = WSModule._(1, const $core.bool.fromEnvironment('protobuf.omit_enum_names') ? '' : 'Folder');
class WSChannel extends $pb.ProtobufEnum {
static const WSChannel Document = WSChannel._(0, const $core.bool.fromEnvironment('protobuf.omit_enum_names') ? '' : 'Document');
static const WSChannel Folder = WSChannel._(1, const $core.bool.fromEnvironment('protobuf.omit_enum_names') ? '' : 'Folder');
static const $core.List<WSModule> values = <WSModule> [
Doc,
static const $core.List<WSChannel> values = <WSChannel> [
Document,
Folder,
];
static final $core.Map<$core.int, WSModule> _byValue = $pb.ProtobufEnum.initByValue(values);
static WSModule? valueOf($core.int value) => _byValue[value];
static final $core.Map<$core.int, WSChannel> _byValue = $pb.ProtobufEnum.initByValue(values);
static WSChannel? valueOf($core.int value) => _byValue[value];
const WSModule._($core.int v, $core.String n) : super(v, n);
const WSChannel._($core.int v, $core.String n) : super(v, n);
}

View File

@ -8,25 +8,25 @@
import 'dart:core' as $core;
import 'dart:convert' as $convert;
import 'dart:typed_data' as $typed_data;
@$core.Deprecated('Use wSModuleDescriptor instead')
const WSModule$json = const {
'1': 'WSModule',
@$core.Deprecated('Use wSChannelDescriptor instead')
const WSChannel$json = const {
'1': 'WSChannel',
'2': const [
const {'1': 'Doc', '2': 0},
const {'1': 'Document', '2': 0},
const {'1': 'Folder', '2': 1},
],
};
/// Descriptor for `WSModule`. Decode as a `google.protobuf.EnumDescriptorProto`.
final $typed_data.Uint8List wSModuleDescriptor = $convert.base64Decode('CghXU01vZHVsZRIHCgNEb2MQABIKCgZGb2xkZXIQAQ==');
/// Descriptor for `WSChannel`. Decode as a `google.protobuf.EnumDescriptorProto`.
final $typed_data.Uint8List wSChannelDescriptor = $convert.base64Decode('CglXU0NoYW5uZWwSDAoIRG9jdW1lbnQQABIKCgZGb2xkZXIQAQ==');
@$core.Deprecated('Use webSocketRawMessageDescriptor instead')
const WebSocketRawMessage$json = const {
'1': 'WebSocketRawMessage',
'2': const [
const {'1': 'module', '3': 1, '4': 1, '5': 14, '6': '.WSModule', '10': 'module'},
const {'1': 'channel', '3': 1, '4': 1, '5': 14, '6': '.WSChannel', '10': 'channel'},
const {'1': 'data', '3': 2, '4': 1, '5': 12, '10': 'data'},
],
};
/// Descriptor for `WebSocketRawMessage`. Decode as a `google.protobuf.DescriptorProto`.
final $typed_data.Uint8List webSocketRawMessageDescriptor = $convert.base64Decode('ChNXZWJTb2NrZXRSYXdNZXNzYWdlEiEKBm1vZHVsZRgBIAEoDjIJLldTTW9kdWxlUgZtb2R1bGUSEgoEZGF0YRgCIAEoDFIEZGF0YQ==');
final $typed_data.Uint8List webSocketRawMessageDescriptor = $convert.base64Decode('ChNXZWJTb2NrZXRSYXdNZXNzYWdlEiQKB2NoYW5uZWwYASABKA4yCi5XU0NoYW5uZWxSB2NoYW5uZWwSEgoEZGF0YRgCIAEoDFIEZGF0YQ==');

View File

@ -45,8 +45,11 @@ chrono = "0.4"
[dev-dependencies]
serial_test = "0.5.1"
serde_json = "1.0"
flowy-core = { path = "../flowy-core", features = ["flowy_unit_test"]}
flowy-test = { path = "../flowy-test" }
[features]
default = []
http_server = []
http_server = []
flowy_unit_test = ["lib-ot/flowy_unit_test", "flowy-sync/flowy_unit_test"]

View File

@ -2,13 +2,13 @@ use bytes::Bytes;
use chrono::Utc;
use flowy_collaboration::client_document::default::{initial_delta, initial_read_me};
use flowy_core_data_model::user_default;
use flowy_document::context::DocumentContext;
use flowy_sync::RevisionWebSocket;
use lazy_static::lazy_static;
use flowy_collaboration::{entities::ws_data::ServerRevisionWSData, folder::FolderPad};
use flowy_document::FlowyDocumentManager;
use parking_lot::RwLock;
use std::{collections::HashMap, convert::TryInto, sync::Arc};
use std::{collections::HashMap, convert::TryInto, fmt::Formatter, sync::Arc};
use tokio::sync::RwLock as TokioRwLock;
use crate::{
@ -31,6 +31,26 @@ lazy_static! {
static ref INIT_FOLDER_FLAG: RwLock<HashMap<String, bool>> = RwLock::new(HashMap::new());
}
const FOLDER_ID: &str = "user_folder";
const FOLDER_ID_SPLIT: &str = ":";
#[derive(Clone)]
pub struct FolderId(String);
impl FolderId {
pub fn new(user_id: &str) -> Self { Self(format!("{}{}{}", user_id, FOLDER_ID_SPLIT, FOLDER_ID)) }
}
impl std::fmt::Display for FolderId {
fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { f.write_str(FOLDER_ID) }
}
impl std::fmt::Debug for FolderId {
fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { f.write_str(FOLDER_ID) }
}
impl AsRef<str> for FolderId {
fn as_ref(&self) -> &str { &self.0 }
}
pub struct FolderManager {
pub user: Arc<dyn WorkspaceUser>,
pub(crate) cloud_service: Arc<dyn FolderCouldServiceV1>,
@ -48,7 +68,7 @@ impl FolderManager {
user: Arc<dyn WorkspaceUser>,
cloud_service: Arc<dyn FolderCouldServiceV1>,
database: Arc<dyn WorkspaceDatabase>,
flowy_document: Arc<DocumentContext>,
document_manager: Arc<FlowyDocumentManager>,
web_socket: Arc<dyn RevisionWebSocket>,
) -> Self {
if let Ok(token) = user.token() {
@ -69,7 +89,7 @@ impl FolderManager {
persistence.clone(),
cloud_service.clone(),
trash_controller.clone(),
flowy_document,
document_manager,
));
let app_controller = Arc::new(AppController::new(
@ -130,11 +150,12 @@ impl FolderManager {
return Ok(());
}
}
let _ = self.persistence.initialize(user_id).await?;
let folder_id = FolderId::new(user_id);
let _ = self.persistence.initialize(user_id, &folder_id).await?;
let token = self.user.token()?;
let pool = self.persistence.db_pool()?;
let folder_editor = FolderEditor::new(user_id, &token, pool, self.web_socket.clone()).await?;
let folder_editor = FolderEditor::new(user_id, &folder_id, &token, pool, self.web_socket.clone()).await?;
*self.folder_editor.write().await = Some(Arc::new(folder_editor));
let _ = self.app_controller.initialize()?;
@ -177,7 +198,8 @@ impl DefaultFolderBuilder {
}
}
let folder = FolderPad::new(vec![workspace.clone()], vec![])?;
let _ = persistence.save_folder(user_id, folder).await?;
let folder_id = FolderId::new(user_id);
let _ = persistence.save_folder(user_id, &folder_id, folder).await?;
let repeated_workspace = RepeatedWorkspace { items: vec![workspace] };
send_dart_notification(token, WorkspaceNotification::UserCreateWorkspace)
.payload(repeated_workspace)
@ -185,3 +207,8 @@ impl DefaultFolderBuilder {
Ok(())
}
}
#[cfg(feature = "flowy_unit_test")]
impl FolderManager {
pub async fn folder_editor(&self) -> Arc<FolderEditor> { self.folder_editor.read().await.clone().unwrap() }
}

View File

@ -53,7 +53,7 @@ pub enum WorkspaceEvent {
CopyLink = 206,
#[event(input = "QueryViewRequest", output = "DocumentDelta")]
OpenView = 207,
OpenDocument = 207,
#[event(input = "QueryViewRequest")]
CloseView = 208,
@ -68,10 +68,10 @@ pub enum WorkspaceEvent {
DeleteTrash = 302,
#[event()]
RestoreAll = 303,
RestoreAllTrash = 303,
#[event()]
DeleteAll = 304,
DeleteAllTrash = 304,
#[event(input = "DocumentDelta", output = "DocumentDelta")]
ApplyDocDelta = 400,

View File

@ -2,7 +2,7 @@ pub use flowy_core_data_model::entities;
pub mod event;
pub mod module;
mod services;
pub mod services;
#[macro_use]
mod macros;

View File

@ -62,7 +62,7 @@ pub fn create(folder: Arc<FolderManager>) -> Module {
.event(WorkspaceEvent::UpdateView, update_view_handler)
.event(WorkspaceEvent::DeleteView, delete_view_handler)
.event(WorkspaceEvent::DuplicateView, duplicate_view_handler)
.event(WorkspaceEvent::OpenView, open_view_handler)
.event(WorkspaceEvent::OpenDocument, open_document_handler)
.event(WorkspaceEvent::CloseView, close_view_handler)
.event(WorkspaceEvent::ApplyDocDelta, document_delta_handler);
@ -70,8 +70,8 @@ pub fn create(folder: Arc<FolderManager>) -> Module {
.event(WorkspaceEvent::ReadTrash, read_trash_handler)
.event(WorkspaceEvent::PutbackTrash, putback_trash_handler)
.event(WorkspaceEvent::DeleteTrash, delete_trash_handler)
.event(WorkspaceEvent::RestoreAll, restore_all_handler)
.event(WorkspaceEvent::DeleteAll, delete_all_handler);
.event(WorkspaceEvent::RestoreAllTrash, restore_all_trash_handler)
.event(WorkspaceEvent::DeleteAllTrash, delete_all_trash_handler);
module = module.event(WorkspaceEvent::ExportDocument, export_handler);

View File

@ -41,13 +41,13 @@ pub enum WorkspaceEvent {
DeleteView = 204,
DuplicateView = 205,
CopyLink = 206,
OpenView = 207,
OpenDocument = 207,
CloseView = 208,
ReadTrash = 300,
PutbackTrash = 301,
DeleteTrash = 302,
RestoreAll = 303,
DeleteAll = 304,
RestoreAllTrash = 303,
DeleteAllTrash = 304,
ApplyDocDelta = 400,
ExportDocument = 500,
}
@ -75,13 +75,13 @@ impl ::protobuf::ProtobufEnum for WorkspaceEvent {
204 => ::std::option::Option::Some(WorkspaceEvent::DeleteView),
205 => ::std::option::Option::Some(WorkspaceEvent::DuplicateView),
206 => ::std::option::Option::Some(WorkspaceEvent::CopyLink),
207 => ::std::option::Option::Some(WorkspaceEvent::OpenView),
207 => ::std::option::Option::Some(WorkspaceEvent::OpenDocument),
208 => ::std::option::Option::Some(WorkspaceEvent::CloseView),
300 => ::std::option::Option::Some(WorkspaceEvent::ReadTrash),
301 => ::std::option::Option::Some(WorkspaceEvent::PutbackTrash),
302 => ::std::option::Option::Some(WorkspaceEvent::DeleteTrash),
303 => ::std::option::Option::Some(WorkspaceEvent::RestoreAll),
304 => ::std::option::Option::Some(WorkspaceEvent::DeleteAll),
303 => ::std::option::Option::Some(WorkspaceEvent::RestoreAllTrash),
304 => ::std::option::Option::Some(WorkspaceEvent::DeleteAllTrash),
400 => ::std::option::Option::Some(WorkspaceEvent::ApplyDocDelta),
500 => ::std::option::Option::Some(WorkspaceEvent::ExportDocument),
_ => ::std::option::Option::None
@ -106,13 +106,13 @@ impl ::protobuf::ProtobufEnum for WorkspaceEvent {
WorkspaceEvent::DeleteView,
WorkspaceEvent::DuplicateView,
WorkspaceEvent::CopyLink,
WorkspaceEvent::OpenView,
WorkspaceEvent::OpenDocument,
WorkspaceEvent::CloseView,
WorkspaceEvent::ReadTrash,
WorkspaceEvent::PutbackTrash,
WorkspaceEvent::DeleteTrash,
WorkspaceEvent::RestoreAll,
WorkspaceEvent::DeleteAll,
WorkspaceEvent::RestoreAllTrash,
WorkspaceEvent::DeleteAllTrash,
WorkspaceEvent::ApplyDocDelta,
WorkspaceEvent::ExportDocument,
];
@ -143,72 +143,72 @@ impl ::protobuf::reflect::ProtobufValue for WorkspaceEvent {
}
static file_descriptor_proto_data: &'static [u8] = b"\
\n\x0bevent.proto*\xcb\x03\n\x0eWorkspaceEvent\x12\x13\n\x0fCreateWorksp\
\n\x0bevent.proto*\xd9\x03\n\x0eWorkspaceEvent\x12\x13\n\x0fCreateWorksp\
ace\x10\0\x12\x14\n\x10ReadCurWorkspace\x10\x01\x12\x12\n\x0eReadWorkspa\
ces\x10\x02\x12\x13\n\x0fDeleteWorkspace\x10\x03\x12\x11\n\rOpenWorkspac\
e\x10\x04\x12\x15\n\x11ReadWorkspaceApps\x10\x05\x12\r\n\tCreateApp\x10e\
\x12\r\n\tDeleteApp\x10f\x12\x0b\n\x07ReadApp\x10g\x12\r\n\tUpdateApp\
\x10h\x12\x0f\n\nCreateView\x10\xc9\x01\x12\r\n\x08ReadView\x10\xca\x01\
\x12\x0f\n\nUpdateView\x10\xcb\x01\x12\x0f\n\nDeleteView\x10\xcc\x01\x12\
\x12\n\rDuplicateView\x10\xcd\x01\x12\r\n\x08CopyLink\x10\xce\x01\x12\r\
\n\x08OpenView\x10\xcf\x01\x12\x0e\n\tCloseView\x10\xd0\x01\x12\x0e\n\tR\
eadTrash\x10\xac\x02\x12\x11\n\x0cPutbackTrash\x10\xad\x02\x12\x10\n\x0b\
DeleteTrash\x10\xae\x02\x12\x0f\n\nRestoreAll\x10\xaf\x02\x12\x0e\n\tDel\
eteAll\x10\xb0\x02\x12\x12\n\rApplyDocDelta\x10\x90\x03\x12\x13\n\x0eExp\
ortDocument\x10\xf4\x03J\xab\x08\n\x06\x12\x04\0\0\x1c\x01\n\x08\n\x01\
\x0c\x12\x03\0\0\x12\n\n\n\x02\x05\0\x12\x04\x02\0\x1c\x01\n\n\n\x03\x05\
\0\x01\x12\x03\x02\x05\x13\n\x0b\n\x04\x05\0\x02\0\x12\x03\x03\x04\x18\n\
\x0c\n\x05\x05\0\x02\0\x01\x12\x03\x03\x04\x13\n\x0c\n\x05\x05\0\x02\0\
\x02\x12\x03\x03\x16\x17\n\x0b\n\x04\x05\0\x02\x01\x12\x03\x04\x04\x19\n\
\x0c\n\x05\x05\0\x02\x01\x01\x12\x03\x04\x04\x14\n\x0c\n\x05\x05\0\x02\
\x01\x02\x12\x03\x04\x17\x18\n\x0b\n\x04\x05\0\x02\x02\x12\x03\x05\x04\
\x17\n\x0c\n\x05\x05\0\x02\x02\x01\x12\x03\x05\x04\x12\n\x0c\n\x05\x05\0\
\x02\x02\x02\x12\x03\x05\x15\x16\n\x0b\n\x04\x05\0\x02\x03\x12\x03\x06\
\x04\x18\n\x0c\n\x05\x05\0\x02\x03\x01\x12\x03\x06\x04\x13\n\x0c\n\x05\
\x05\0\x02\x03\x02\x12\x03\x06\x16\x17\n\x0b\n\x04\x05\0\x02\x04\x12\x03\
\x07\x04\x16\n\x0c\n\x05\x05\0\x02\x04\x01\x12\x03\x07\x04\x11\n\x0c\n\
\x05\x05\0\x02\x04\x02\x12\x03\x07\x14\x15\n\x0b\n\x04\x05\0\x02\x05\x12\
\x03\x08\x04\x1a\n\x0c\n\x05\x05\0\x02\x05\x01\x12\x03\x08\x04\x15\n\x0c\
\n\x05\x05\0\x02\x05\x02\x12\x03\x08\x18\x19\n\x0b\n\x04\x05\0\x02\x06\
\x12\x03\t\x04\x14\n\x0c\n\x05\x05\0\x02\x06\x01\x12\x03\t\x04\r\n\x0c\n\
\x05\x05\0\x02\x06\x02\x12\x03\t\x10\x13\n\x0b\n\x04\x05\0\x02\x07\x12\
\x03\n\x04\x14\n\x0c\n\x05\x05\0\x02\x07\x01\x12\x03\n\x04\r\n\x0c\n\x05\
\x05\0\x02\x07\x02\x12\x03\n\x10\x13\n\x0b\n\x04\x05\0\x02\x08\x12\x03\
\x0b\x04\x12\n\x0c\n\x05\x05\0\x02\x08\x01\x12\x03\x0b\x04\x0b\n\x0c\n\
\x05\x05\0\x02\x08\x02\x12\x03\x0b\x0e\x11\n\x0b\n\x04\x05\0\x02\t\x12\
\x03\x0c\x04\x14\n\x0c\n\x05\x05\0\x02\t\x01\x12\x03\x0c\x04\r\n\x0c\n\
\x05\x05\0\x02\t\x02\x12\x03\x0c\x10\x13\n\x0b\n\x04\x05\0\x02\n\x12\x03\
\r\x04\x15\n\x0c\n\x05\x05\0\x02\n\x01\x12\x03\r\x04\x0e\n\x0c\n\x05\x05\
\0\x02\n\x02\x12\x03\r\x11\x14\n\x0b\n\x04\x05\0\x02\x0b\x12\x03\x0e\x04\
\x13\n\x0c\n\x05\x05\0\x02\x0b\x01\x12\x03\x0e\x04\x0c\n\x0c\n\x05\x05\0\
\x02\x0b\x02\x12\x03\x0e\x0f\x12\n\x0b\n\x04\x05\0\x02\x0c\x12\x03\x0f\
\x04\x15\n\x0c\n\x05\x05\0\x02\x0c\x01\x12\x03\x0f\x04\x0e\n\x0c\n\x05\
\x05\0\x02\x0c\x02\x12\x03\x0f\x11\x14\n\x0b\n\x04\x05\0\x02\r\x12\x03\
\x10\x04\x15\n\x0c\n\x05\x05\0\x02\r\x01\x12\x03\x10\x04\x0e\n\x0c\n\x05\
\x05\0\x02\r\x02\x12\x03\x10\x11\x14\n\x0b\n\x04\x05\0\x02\x0e\x12\x03\
\x11\x04\x18\n\x0c\n\x05\x05\0\x02\x0e\x01\x12\x03\x11\x04\x11\n\x0c\n\
\x05\x05\0\x02\x0e\x02\x12\x03\x11\x14\x17\n\x0b\n\x04\x05\0\x02\x0f\x12\
\x03\x12\x04\x13\n\x0c\n\x05\x05\0\x02\x0f\x01\x12\x03\x12\x04\x0c\n\x0c\
\n\x05\x05\0\x02\x0f\x02\x12\x03\x12\x0f\x12\n\x0b\n\x04\x05\0\x02\x10\
\x12\x03\x13\x04\x13\n\x0c\n\x05\x05\0\x02\x10\x01\x12\x03\x13\x04\x0c\n\
\x0c\n\x05\x05\0\x02\x10\x02\x12\x03\x13\x0f\x12\n\x0b\n\x04\x05\0\x02\
\x11\x12\x03\x14\x04\x14\n\x0c\n\x05\x05\0\x02\x11\x01\x12\x03\x14\x04\r\
\n\x0c\n\x05\x05\0\x02\x11\x02\x12\x03\x14\x10\x13\n\x0b\n\x04\x05\0\x02\
\x12\x12\x03\x15\x04\x14\n\x0c\n\x05\x05\0\x02\x12\x01\x12\x03\x15\x04\r\
\n\x0c\n\x05\x05\0\x02\x12\x02\x12\x03\x15\x10\x13\n\x0b\n\x04\x05\0\x02\
\x13\x12\x03\x16\x04\x17\n\x0c\n\x05\x05\0\x02\x13\x01\x12\x03\x16\x04\
\x10\n\x0c\n\x05\x05\0\x02\x13\x02\x12\x03\x16\x13\x16\n\x0b\n\x04\x05\0\
\x02\x14\x12\x03\x17\x04\x16\n\x0c\n\x05\x05\0\x02\x14\x01\x12\x03\x17\
\x04\x0f\n\x0c\n\x05\x05\0\x02\x14\x02\x12\x03\x17\x12\x15\n\x0b\n\x04\
\x05\0\x02\x15\x12\x03\x18\x04\x15\n\x0c\n\x05\x05\0\x02\x15\x01\x12\x03\
\x18\x04\x0e\n\x0c\n\x05\x05\0\x02\x15\x02\x12\x03\x18\x11\x14\n\x0b\n\
\x04\x05\0\x02\x16\x12\x03\x19\x04\x14\n\x0c\n\x05\x05\0\x02\x16\x01\x12\
\x03\x19\x04\r\n\x0c\n\x05\x05\0\x02\x16\x02\x12\x03\x19\x10\x13\n\x0b\n\
\x04\x05\0\x02\x17\x12\x03\x1a\x04\x18\n\x0c\n\x05\x05\0\x02\x17\x01\x12\
\x03\x1a\x04\x11\n\x0c\n\x05\x05\0\x02\x17\x02\x12\x03\x1a\x14\x17\n\x0b\
\n\x04\x05\0\x02\x18\x12\x03\x1b\x04\x19\n\x0c\n\x05\x05\0\x02\x18\x01\
\x12\x03\x1b\x04\x12\n\x0c\n\x05\x05\0\x02\x18\x02\x12\x03\x1b\x15\x18b\
\x06proto3\
\x12\n\rDuplicateView\x10\xcd\x01\x12\r\n\x08CopyLink\x10\xce\x01\x12\
\x11\n\x0cOpenDocument\x10\xcf\x01\x12\x0e\n\tCloseView\x10\xd0\x01\x12\
\x0e\n\tReadTrash\x10\xac\x02\x12\x11\n\x0cPutbackTrash\x10\xad\x02\x12\
\x10\n\x0bDeleteTrash\x10\xae\x02\x12\x14\n\x0fRestoreAllTrash\x10\xaf\
\x02\x12\x13\n\x0eDeleteAllTrash\x10\xb0\x02\x12\x12\n\rApplyDocDelta\
\x10\x90\x03\x12\x13\n\x0eExportDocument\x10\xf4\x03J\xab\x08\n\x06\x12\
\x04\0\0\x1c\x01\n\x08\n\x01\x0c\x12\x03\0\0\x12\n\n\n\x02\x05\0\x12\x04\
\x02\0\x1c\x01\n\n\n\x03\x05\0\x01\x12\x03\x02\x05\x13\n\x0b\n\x04\x05\0\
\x02\0\x12\x03\x03\x04\x18\n\x0c\n\x05\x05\0\x02\0\x01\x12\x03\x03\x04\
\x13\n\x0c\n\x05\x05\0\x02\0\x02\x12\x03\x03\x16\x17\n\x0b\n\x04\x05\0\
\x02\x01\x12\x03\x04\x04\x19\n\x0c\n\x05\x05\0\x02\x01\x01\x12\x03\x04\
\x04\x14\n\x0c\n\x05\x05\0\x02\x01\x02\x12\x03\x04\x17\x18\n\x0b\n\x04\
\x05\0\x02\x02\x12\x03\x05\x04\x17\n\x0c\n\x05\x05\0\x02\x02\x01\x12\x03\
\x05\x04\x12\n\x0c\n\x05\x05\0\x02\x02\x02\x12\x03\x05\x15\x16\n\x0b\n\
\x04\x05\0\x02\x03\x12\x03\x06\x04\x18\n\x0c\n\x05\x05\0\x02\x03\x01\x12\
\x03\x06\x04\x13\n\x0c\n\x05\x05\0\x02\x03\x02\x12\x03\x06\x16\x17\n\x0b\
\n\x04\x05\0\x02\x04\x12\x03\x07\x04\x16\n\x0c\n\x05\x05\0\x02\x04\x01\
\x12\x03\x07\x04\x11\n\x0c\n\x05\x05\0\x02\x04\x02\x12\x03\x07\x14\x15\n\
\x0b\n\x04\x05\0\x02\x05\x12\x03\x08\x04\x1a\n\x0c\n\x05\x05\0\x02\x05\
\x01\x12\x03\x08\x04\x15\n\x0c\n\x05\x05\0\x02\x05\x02\x12\x03\x08\x18\
\x19\n\x0b\n\x04\x05\0\x02\x06\x12\x03\t\x04\x14\n\x0c\n\x05\x05\0\x02\
\x06\x01\x12\x03\t\x04\r\n\x0c\n\x05\x05\0\x02\x06\x02\x12\x03\t\x10\x13\
\n\x0b\n\x04\x05\0\x02\x07\x12\x03\n\x04\x14\n\x0c\n\x05\x05\0\x02\x07\
\x01\x12\x03\n\x04\r\n\x0c\n\x05\x05\0\x02\x07\x02\x12\x03\n\x10\x13\n\
\x0b\n\x04\x05\0\x02\x08\x12\x03\x0b\x04\x12\n\x0c\n\x05\x05\0\x02\x08\
\x01\x12\x03\x0b\x04\x0b\n\x0c\n\x05\x05\0\x02\x08\x02\x12\x03\x0b\x0e\
\x11\n\x0b\n\x04\x05\0\x02\t\x12\x03\x0c\x04\x14\n\x0c\n\x05\x05\0\x02\t\
\x01\x12\x03\x0c\x04\r\n\x0c\n\x05\x05\0\x02\t\x02\x12\x03\x0c\x10\x13\n\
\x0b\n\x04\x05\0\x02\n\x12\x03\r\x04\x15\n\x0c\n\x05\x05\0\x02\n\x01\x12\
\x03\r\x04\x0e\n\x0c\n\x05\x05\0\x02\n\x02\x12\x03\r\x11\x14\n\x0b\n\x04\
\x05\0\x02\x0b\x12\x03\x0e\x04\x13\n\x0c\n\x05\x05\0\x02\x0b\x01\x12\x03\
\x0e\x04\x0c\n\x0c\n\x05\x05\0\x02\x0b\x02\x12\x03\x0e\x0f\x12\n\x0b\n\
\x04\x05\0\x02\x0c\x12\x03\x0f\x04\x15\n\x0c\n\x05\x05\0\x02\x0c\x01\x12\
\x03\x0f\x04\x0e\n\x0c\n\x05\x05\0\x02\x0c\x02\x12\x03\x0f\x11\x14\n\x0b\
\n\x04\x05\0\x02\r\x12\x03\x10\x04\x15\n\x0c\n\x05\x05\0\x02\r\x01\x12\
\x03\x10\x04\x0e\n\x0c\n\x05\x05\0\x02\r\x02\x12\x03\x10\x11\x14\n\x0b\n\
\x04\x05\0\x02\x0e\x12\x03\x11\x04\x18\n\x0c\n\x05\x05\0\x02\x0e\x01\x12\
\x03\x11\x04\x11\n\x0c\n\x05\x05\0\x02\x0e\x02\x12\x03\x11\x14\x17\n\x0b\
\n\x04\x05\0\x02\x0f\x12\x03\x12\x04\x13\n\x0c\n\x05\x05\0\x02\x0f\x01\
\x12\x03\x12\x04\x0c\n\x0c\n\x05\x05\0\x02\x0f\x02\x12\x03\x12\x0f\x12\n\
\x0b\n\x04\x05\0\x02\x10\x12\x03\x13\x04\x17\n\x0c\n\x05\x05\0\x02\x10\
\x01\x12\x03\x13\x04\x10\n\x0c\n\x05\x05\0\x02\x10\x02\x12\x03\x13\x13\
\x16\n\x0b\n\x04\x05\0\x02\x11\x12\x03\x14\x04\x14\n\x0c\n\x05\x05\0\x02\
\x11\x01\x12\x03\x14\x04\r\n\x0c\n\x05\x05\0\x02\x11\x02\x12\x03\x14\x10\
\x13\n\x0b\n\x04\x05\0\x02\x12\x12\x03\x15\x04\x14\n\x0c\n\x05\x05\0\x02\
\x12\x01\x12\x03\x15\x04\r\n\x0c\n\x05\x05\0\x02\x12\x02\x12\x03\x15\x10\
\x13\n\x0b\n\x04\x05\0\x02\x13\x12\x03\x16\x04\x17\n\x0c\n\x05\x05\0\x02\
\x13\x01\x12\x03\x16\x04\x10\n\x0c\n\x05\x05\0\x02\x13\x02\x12\x03\x16\
\x13\x16\n\x0b\n\x04\x05\0\x02\x14\x12\x03\x17\x04\x16\n\x0c\n\x05\x05\0\
\x02\x14\x01\x12\x03\x17\x04\x0f\n\x0c\n\x05\x05\0\x02\x14\x02\x12\x03\
\x17\x12\x15\n\x0b\n\x04\x05\0\x02\x15\x12\x03\x18\x04\x1a\n\x0c\n\x05\
\x05\0\x02\x15\x01\x12\x03\x18\x04\x13\n\x0c\n\x05\x05\0\x02\x15\x02\x12\
\x03\x18\x16\x19\n\x0b\n\x04\x05\0\x02\x16\x12\x03\x19\x04\x19\n\x0c\n\
\x05\x05\0\x02\x16\x01\x12\x03\x19\x04\x12\n\x0c\n\x05\x05\0\x02\x16\x02\
\x12\x03\x19\x15\x18\n\x0b\n\x04\x05\0\x02\x17\x12\x03\x1a\x04\x18\n\x0c\
\n\x05\x05\0\x02\x17\x01\x12\x03\x1a\x04\x11\n\x0c\n\x05\x05\0\x02\x17\
\x02\x12\x03\x1a\x14\x17\n\x0b\n\x04\x05\0\x02\x18\x12\x03\x1b\x04\x19\n\
\x0c\n\x05\x05\0\x02\x18\x01\x12\x03\x1b\x04\x12\n\x0c\n\x05\x05\0\x02\
\x18\x02\x12\x03\x1b\x15\x18b\x06proto3\
";
static file_descriptor_proto_lazy: ::protobuf::rt::LazyV2<::protobuf::descriptor::FileDescriptorProto> = ::protobuf::rt::LazyV2::INIT;

View File

@ -17,13 +17,13 @@ enum WorkspaceEvent {
DeleteView = 204;
DuplicateView = 205;
CopyLink = 206;
OpenView = 207;
OpenDocument = 207;
CloseView = 208;
ReadTrash = 300;
PutbackTrash = 301;
DeleteTrash = 302;
RestoreAll = 303;
DeleteAll = 304;
RestoreAllTrash = 303;
DeleteAllTrash = 304;
ApplyDocDelta = 400;
ExportDocument = 500;
}

View File

@ -1,9 +1,10 @@
use crate::services::{persistence::FOLDER_ID, web_socket::make_folder_ws_manager};
use crate::services::web_socket::make_folder_ws_manager;
use flowy_collaboration::{
entities::{revision::Revision, ws_data::ServerRevisionWSData},
folder::{FolderChange, FolderPad},
};
use crate::controller::FolderId;
use flowy_error::{FlowyError, FlowyResult};
use flowy_sync::{
RevisionCache,
@ -20,6 +21,7 @@ use std::sync::Arc;
pub struct FolderEditor {
user_id: String,
pub(crate) folder_id: FolderId,
pub(crate) folder: Arc<RwLock<FolderPad>>,
rev_manager: Arc<RevisionManager>,
ws_manager: Arc<RevisionWebSocketManager>,
@ -28,23 +30,33 @@ pub struct FolderEditor {
impl FolderEditor {
pub async fn new(
user_id: &str,
folder_id: &FolderId,
token: &str,
pool: Arc<ConnectionPool>,
web_socket: Arc<dyn RevisionWebSocket>,
) -> FlowyResult<Self> {
let cache = Arc::new(RevisionCache::new(user_id, FOLDER_ID, pool));
let mut rev_manager = RevisionManager::new(user_id, FOLDER_ID, cache);
let cache = Arc::new(RevisionCache::new(user_id, folder_id.as_ref(), pool));
let mut rev_manager = RevisionManager::new(user_id, folder_id.as_ref(), cache);
let cloud = Arc::new(FolderRevisionCloudServiceImpl {
token: token.to_string(),
});
let folder_pad = Arc::new(RwLock::new(rev_manager.load::<FolderPadBuilder>(cloud).await?));
let folder = Arc::new(RwLock::new(rev_manager.load::<FolderPadBuilder>(cloud).await?));
let rev_manager = Arc::new(rev_manager);
let ws_manager = make_folder_ws_manager(user_id, rev_manager.clone(), web_socket, folder_pad.clone()).await;
let ws_manager = make_folder_ws_manager(
user_id,
folder_id.as_ref(),
rev_manager.clone(),
web_socket,
folder.clone(),
)
.await;
let user_id = user_id.to_owned();
let folder_id = folder_id.to_owned();
Ok(Self {
user_id,
folder: folder_pad,
folder_id,
folder,
rev_manager,
ws_manager,
})
@ -52,7 +64,7 @@ impl FolderEditor {
pub async fn receive_ws_data(&self, data: ServerRevisionWSData) -> FlowyResult<()> {
let _ = self.ws_manager.ws_passthrough_tx.send(data).await.map_err(|e| {
let err_msg = format!("{} passthrough error: {}", FOLDER_ID, e);
let err_msg = format!("{} passthrough error: {}", self.folder_id, e);
FlowyError::internal().context(err_msg)
})?;
Ok(())
@ -97,3 +109,8 @@ impl RevisionCloudService for FolderRevisionCloudServiceImpl {
FutureResult::new(async move { Ok(vec![]) })
}
}
#[cfg(feature = "flowy_unit_test")]
impl FolderEditor {
pub fn rev_manager(&self) -> Arc<RevisionManager> { self.rev_manager.clone() }
}

View File

@ -4,9 +4,11 @@ pub(crate) use view::controller::*;
pub(crate) use workspace::controller::*;
pub(crate) mod app;
pub(crate) mod folder_editor;
pub mod folder_editor;
pub(crate) mod persistence;
pub(crate) mod trash;
pub(crate) mod view;
mod web_socket;
pub(crate) mod workspace;
pub const FOLDER_SYNC_INTERVAL_IN_MILLIS: u64 = 1000;

View File

@ -11,13 +11,13 @@ use tokio::sync::RwLock;
pub use version_1::{app_sql::*, trash_sql::*, v1_impl::V1Transaction, view_sql::*, workspace_sql::*};
use crate::{
controller::FolderId,
module::WorkspaceDatabase,
services::{folder_editor::FolderEditor, persistence::migration::FolderMigration},
};
use flowy_core_data_model::entities::{
app::App,
prelude::RepeatedTrash,
trash::Trash,
trash::{RepeatedTrash, Trash},
view::View,
workspace::Workspace,
};
@ -25,8 +25,6 @@ use flowy_error::{FlowyError, FlowyResult};
use flowy_sync::{mk_revision_disk_cache, RevisionRecord};
use lib_sqlite::ConnectionPool;
pub const FOLDER_ID: &str = "flowy_folder";
pub trait FolderPersistenceTransaction {
fn create_workspace(&self, user_id: &str, workspace: Workspace) -> FlowyResult<()>;
fn read_workspaces(&self, user_id: &str, workspace_id: Option<String>) -> FlowyResult<Vec<Workspace>>;
@ -99,21 +97,21 @@ impl FolderPersistence {
pub fn db_pool(&self) -> FlowyResult<Arc<ConnectionPool>> { self.database.db_pool() }
pub async fn initialize(&self, user_id: &str) -> FlowyResult<()> {
pub async fn initialize(&self, user_id: &str, folder_id: &FolderId) -> FlowyResult<()> {
let migrations = FolderMigration::new(user_id, self.database.clone());
if let Some(migrated_folder) = migrations.run_v1_migration()? {
tracing::trace!("Save migration folder");
self.save_folder(user_id, migrated_folder).await?;
self.save_folder(user_id, folder_id, migrated_folder).await?;
}
Ok(())
}
pub async fn save_folder(&self, user_id: &str, folder: FolderPad) -> FlowyResult<()> {
pub async fn save_folder(&self, user_id: &str, folder_id: &FolderId, folder: FolderPad) -> FlowyResult<()> {
let pool = self.database.db_pool()?;
let delta_data = folder.delta().to_bytes();
let md5 = folder.md5();
let revision = Revision::new(FOLDER_ID, 0, 0, delta_data, user_id, md5);
let revision = Revision::new(folder_id.as_ref(), 0, 0, delta_data, user_id, md5);
let record = RevisionRecord {
revision,
state: RevisionState::Sync,

View File

@ -9,7 +9,9 @@ use crate::services::persistence::{
};
use flowy_core_data_model::entities::{
app::App,
prelude::{RepeatedTrash, Trash, View, Workspace},
trash::{RepeatedTrash, Trash},
view::View,
workspace::Workspace,
};
use flowy_error::FlowyResult;
use lib_sqlite::DBConnection;

View File

@ -2,13 +2,13 @@ use crate::services::{
folder_editor::FolderEditor,
persistence::{AppChangeset, FolderPersistenceTransaction, ViewChangeset, WorkspaceChangeset},
};
use flowy_core_data_model::entities::{
app::App,
prelude::{RepeatedTrash, Trash, View, Workspace},
trash::{RepeatedTrash, Trash},
view::View,
workspace::Workspace,
};
use flowy_error::{FlowyError, FlowyResult};
use std::sync::Arc;
impl FolderPersistenceTransaction for FolderEditor {

View File

@ -65,7 +65,7 @@ impl TrashController {
}
#[tracing::instrument(level = "debug", skip(self) err)]
pub async fn restore_all(&self) -> FlowyResult<()> {
pub async fn restore_all_trash(&self) -> FlowyResult<()> {
let repeated_trash = self
.persistence
.begin_transaction(|transaction| {
@ -86,7 +86,7 @@ impl TrashController {
}
#[tracing::instrument(level = "debug", skip(self), err)]
pub async fn delete_all(&self) -> FlowyResult<()> {
pub async fn delete_all_trash(&self) -> FlowyResult<()> {
let repeated_trash = self
.persistence
.begin_transaction(|transaction| transaction.read_trash(None))

View File

@ -33,13 +33,13 @@ pub(crate) async fn delete_trash_handler(
}
#[tracing::instrument(skip(controller), err)]
pub(crate) async fn restore_all_handler(controller: Unit<Arc<TrashController>>) -> Result<(), FlowyError> {
let _ = controller.restore_all().await?;
pub(crate) async fn restore_all_trash_handler(controller: Unit<Arc<TrashController>>) -> Result<(), FlowyError> {
let _ = controller.restore_all_trash().await?;
Ok(())
}
#[tracing::instrument(skip(controller), err)]
pub(crate) async fn delete_all_handler(controller: Unit<Arc<TrashController>>) -> Result<(), FlowyError> {
let _ = controller.delete_all().await?;
pub(crate) async fn delete_all_trash_handler(controller: Unit<Arc<TrashController>>) -> Result<(), FlowyError> {
let _ = controller.delete_all_trash().await?;
Ok(())
}

View File

@ -24,7 +24,7 @@ use crate::{
};
use flowy_core_data_model::entities::share::{ExportData, ExportParams};
use flowy_database::kv::KV;
use flowy_document::context::DocumentContext;
use flowy_document::FlowyDocumentManager;
use lib_infra::uuid_string;
const LATEST_VIEW_ID: &str = "latest_view_id";
@ -34,7 +34,7 @@ pub(crate) struct ViewController {
cloud_service: Arc<dyn FolderCouldServiceV1>,
persistence: Arc<FolderPersistence>,
trash_controller: Arc<TrashController>,
document_ctx: Arc<DocumentContext>,
document_manager: Arc<FlowyDocumentManager>,
}
impl ViewController {
@ -43,19 +43,19 @@ impl ViewController {
persistence: Arc<FolderPersistence>,
cloud_service: Arc<dyn FolderCouldServiceV1>,
trash_can: Arc<TrashController>,
document_ctx: Arc<DocumentContext>,
document_manager: Arc<FlowyDocumentManager>,
) -> Self {
Self {
user,
cloud_service,
persistence,
trash_controller: trash_can,
document_ctx,
document_manager,
}
}
pub(crate) fn initialize(&self) -> Result<(), FlowyError> {
let _ = self.document_ctx.init()?;
let _ = self.document_manager.init()?;
self.listen_trash_can_event();
Ok(())
}
@ -73,8 +73,7 @@ impl ViewController {
let repeated_revision: RepeatedRevision =
Revision::initial_revision(&user_id, &params.view_id, delta_data).into();
let _ = self
.document_ctx
.controller
.document_manager
.save_document(&params.view_id, repeated_revision)
.await?;
let view = self.create_view_on_server(params).await?;
@ -96,11 +95,7 @@ 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_ctx
.controller
.save_document(view_id, repeated_revision)
.await?;
let _ = self.document_manager.save_document(view_id, repeated_revision).await?;
Ok(())
}
@ -146,8 +141,8 @@ impl ViewController {
}
#[tracing::instrument(level = "debug", skip(self), err)]
pub(crate) async fn open_view(&self, doc_id: &str) -> Result<DocumentDelta, FlowyError> {
let editor = self.document_ctx.controller.open_document(doc_id).await?;
pub(crate) async fn open_document(&self, doc_id: &str) -> Result<DocumentDelta, FlowyError> {
let editor = self.document_manager.open_document(doc_id).await?;
KV::set_str(LATEST_VIEW_ID, doc_id.to_owned());
let document_json = editor.document_json().await?;
Ok(DocumentDelta {
@ -158,7 +153,7 @@ impl ViewController {
#[tracing::instrument(level = "debug", skip(self), err)]
pub(crate) async fn close_view(&self, doc_id: &str) -> Result<(), FlowyError> {
let _ = self.document_ctx.controller.close_document(doc_id)?;
let _ = self.document_manager.close_document(doc_id)?;
Ok(())
}
@ -169,7 +164,7 @@ impl ViewController {
let _ = KV::remove(LATEST_VIEW_ID);
}
}
let _ = self.document_ctx.controller.close_document(&params.doc_id)?;
let _ = self.document_manager.close_document(&params.doc_id)?;
Ok(())
}
@ -180,7 +175,7 @@ impl ViewController {
.begin_transaction(|transaction| transaction.read_view(doc_id))
.await?;
let editor = self.document_ctx.controller.open_document(doc_id).await?;
let editor = self.document_manager.open_document(doc_id).await?;
let document_json = editor.document_json().await?;
let duplicate_params = CreateViewParams {
belong_to_id: view.belong_to_id.clone(),
@ -198,7 +193,7 @@ impl ViewController {
#[tracing::instrument(level = "debug", skip(self, params), err)]
pub(crate) async fn export_doc(&self, params: ExportParams) -> Result<ExportData, FlowyError> {
let editor = self.document_ctx.controller.open_document(&params.doc_id).await?;
let editor = self.document_manager.open_document(&params.doc_id).await?;
let delta_json = editor.document_json().await?;
Ok(ExportData {
data: delta_json,
@ -238,7 +233,7 @@ impl ViewController {
}
pub(crate) async fn receive_document_delta(&self, params: DocumentDelta) -> Result<DocumentDelta, FlowyError> {
let doc = self.document_ctx.controller.receive_local_delta(params).await?;
let doc = self.document_manager.receive_local_delta(params).await?;
Ok(doc)
}
@ -313,7 +308,7 @@ impl ViewController {
fn listen_trash_can_event(&self) {
let mut rx = self.trash_controller.subscribe();
let persistence = self.persistence.clone();
let document = self.document_ctx.clone();
let document_manager = self.document_manager.clone();
let trash_controller = self.trash_controller.clone();
let _ = tokio::spawn(async move {
loop {
@ -325,17 +320,23 @@ impl ViewController {
}));
if let Some(event) = stream.next().await {
handle_trash_event(persistence.clone(), document.clone(), trash_controller.clone(), event).await
handle_trash_event(
persistence.clone(),
document_manager.clone(),
trash_controller.clone(),
event,
)
.await
}
}
});
}
}
#[tracing::instrument(level = "trace", skip(persistence, context, trash_can))]
#[tracing::instrument(level = "trace", skip(persistence, document_manager, trash_can))]
async fn handle_trash_event(
persistence: Arc<FolderPersistence>,
context: Arc<DocumentContext>,
document_manager: Arc<FlowyDocumentManager>,
trash_can: Arc<TrashController>,
event: TrashEvent,
) {
@ -373,7 +374,7 @@ async fn handle_trash_event(
for identifier in identifiers.items {
let view = transaction.read_view(&identifier.id)?;
let _ = transaction.delete_view(&identifier.id)?;
let _ = context.controller.delete(&identifier.id)?;
let _ = document_manager.delete(&identifier.id)?;
notify_ids.insert(view.belong_to_id);
}

View File

@ -82,12 +82,12 @@ pub(crate) async fn delete_view_handler(
Ok(())
}
pub(crate) async fn open_view_handler(
pub(crate) async fn open_document_handler(
data: Data<QueryViewRequest>,
controller: Unit<Arc<ViewController>>,
) -> DataResult<DocumentDelta, FlowyError> {
let params: ViewId = data.into_inner().try_into()?;
let doc = controller.open_view(&params.view_id).await?;
let doc = controller.open_document(&params.view_id).await?;
data_result(doc)
}

View File

@ -1,4 +1,4 @@
use crate::services::persistence::FOLDER_ID;
use crate::services::FOLDER_SYNC_INTERVAL_IN_MILLIS;
use bytes::Bytes;
use flowy_collaboration::{
entities::{
@ -16,12 +16,12 @@ use std::{sync::Arc, time::Duration};
pub(crate) async fn make_folder_ws_manager(
user_id: &str,
folder_id: &str,
rev_manager: Arc<RevisionManager>,
web_socket: Arc<dyn RevisionWebSocket>,
folder_pad: Arc<RwLock<FolderPad>>,
) -> Arc<RevisionWebSocketManager> {
let object_id = FOLDER_ID;
let composite_sink_provider = Arc::new(CompositeWSSinkDataProvider::new(object_id, rev_manager.clone()));
let composite_sink_provider = Arc::new(CompositeWSSinkDataProvider::new(folder_id, rev_manager.clone()));
let resolve_target = Arc::new(FolderRevisionResolveTarget { folder_pad });
let resolver = RevisionConflictResolver::<PlainTextAttributes>::new(
user_id,
@ -35,9 +35,9 @@ pub(crate) async fn make_folder_ws_manager(
});
let sink_provider = Arc::new(FolderWSSinkDataProviderAdapter(composite_sink_provider));
let ping_duration = Duration::from_millis(2000);
let ping_duration = Duration::from_millis(FOLDER_SYNC_INTERVAL_IN_MILLIS);
Arc::new(RevisionWebSocketManager::new(
object_id,
folder_id,
web_socket,
sink_provider,
ws_stream_consumer,

View File

@ -1,42 +1,77 @@
use flowy_core::{
entities::workspace::{CreateWorkspaceRequest, QueryWorkspaceRequest},
event::WorkspaceEvent::*,
prelude::*,
};
use flowy_test::{event_builder::*, helper::*, FlowySDKTest};
use crate::script::{invalid_workspace_name_test_case, FolderScript::*, FolderTest};
use flowy_collaboration::{client_document::default::initial_delta_string, entities::revision::RevisionState};
use flowy_core::entities::workspace::CreateWorkspaceRequest;
use flowy_test::{event_builder::*, FlowySDKTest};
#[tokio::test]
async fn workspace_read_all() {
let test = WorkspaceTest::new().await;
let workspace = read_workspace(&test.sdk, QueryWorkspaceRequest::new(None)).await;
assert_eq!(workspace.len(), 2);
let mut test = FolderTest::new().await;
test.run_scripts(vec![ReadAllWorkspaces]).await;
// The first workspace will be the default workspace
// The second workspace will be created by FolderTest
assert_eq!(test.all_workspace.len(), 2);
let new_name = "My new workspace".to_owned();
test.run_scripts(vec![
CreateWorkspace {
name: new_name.clone(),
desc: "Daily routines".to_owned(),
},
ReadAllWorkspaces,
])
.await;
assert_eq!(test.all_workspace.len(), 3);
assert_eq!(test.all_workspace[2].name, new_name);
}
#[tokio::test]
async fn workspace_create() {
let mut test = FolderTest::new().await;
let name = "My new workspace".to_owned();
let desc = "Daily routines".to_owned();
test.run_scripts(vec![CreateWorkspace {
name: name.clone(),
desc: desc.clone(),
}])
.await;
let workspace = test.workspace.clone();
assert_eq!(workspace.name, name);
assert_eq!(workspace.desc, desc);
test.run_scripts(vec![
ReadWorkspace(Some(workspace.id.clone())),
AssertWorkspace(workspace),
])
.await;
}
#[tokio::test]
async fn workspace_read() {
let test = WorkspaceTest::new().await;
let request = QueryWorkspaceRequest::new(Some(test.workspace.id.clone()));
let workspace_from_db = read_workspace(&test.sdk, request)
.await
.drain(..1)
.collect::<Vec<Workspace>>()
.pop()
.unwrap();
assert_eq!(test.workspace, workspace_from_db);
let mut test = FolderTest::new().await;
let workspace = test.workspace.clone();
let json = serde_json::to_string(&workspace).unwrap();
test.run_scripts(vec![
ReadWorkspace(Some(workspace.id.clone())),
AssertWorkspaceJson(json),
AssertWorkspace(workspace),
])
.await;
}
#[tokio::test]
async fn workspace_create_with_apps() {
let test = WorkspaceTest::new().await;
let app = create_app(&test.sdk, "App A", "AppFlowy GitHub Project", &test.workspace.id).await;
let request = QueryWorkspaceRequest::new(Some(test.workspace.id.clone()));
let workspace_from_db = read_workspace(&test.sdk, request)
.await
.drain(..1)
.collect::<Vec<Workspace>>()
.pop()
.unwrap();
assert_eq!(&app, workspace_from_db.apps.first_or_crash());
let mut test = FolderTest::new().await;
test.run_scripts(vec![CreateApp {
name: "App",
desc: "App description",
}])
.await;
let app = test.app.clone();
let json = serde_json::to_string(&app).unwrap();
test.run_scripts(vec![ReadApp(app.id), AssertAppJson(json)]).await;
}
#[tokio::test]
@ -48,29 +83,8 @@ async fn workspace_create_with_invalid_name() {
desc: "".to_owned(),
};
assert_eq!(
CoreModuleEventBuilder::new(sdk)
.event(CreateWorkspace)
.request(request)
.async_send()
.await
.error()
.code,
code.value()
)
}
}
#[tokio::test]
async fn workspace_update_with_invalid_name() {
let sdk = FlowySDKTest::default();
for (name, code) in invalid_workspace_name_test_case() {
let request = CreateWorkspaceRequest {
name,
desc: "".to_owned(),
};
assert_eq!(
CoreModuleEventBuilder::new(sdk.clone())
.event(CreateWorkspace)
FolderEventBuilder::new(sdk)
.event(flowy_core::event::WorkspaceEvent::CreateWorkspace)
.request(request)
.async_send()
.await
@ -84,161 +98,262 @@ async fn workspace_update_with_invalid_name() {
#[tokio::test]
#[should_panic]
async fn app_delete() {
let test = AppTest::new().await;
delete_app(&test.sdk, &test.app.id).await;
let query = QueryAppRequest {
app_ids: vec![test.app.id.clone()],
};
let _ = read_app(&test.sdk, query).await;
let mut test = FolderTest::new().await;
let app = test.app.clone();
test.run_scripts(vec![DeleteApp, ReadApp(app.id)]).await;
}
#[tokio::test]
async fn app_delete_then_putback() {
let test = AppTest::new().await;
delete_app(&test.sdk, &test.app.id).await;
putback_trash(
&test.sdk,
TrashId {
id: test.app.id.clone(),
ty: TrashType::App,
},
)
async fn app_delete_then_restore() {
let mut test = FolderTest::new().await;
let app = test.app.clone();
test.run_scripts(vec![
DeleteApp,
RestoreAppFromTrash,
ReadApp(app.id.clone()),
AssertApp(app),
])
.await;
let query = QueryAppRequest {
app_ids: vec![test.app.id.clone()],
};
let app = read_app(&test.sdk, query).await;
assert_eq!(&app, &test.app);
}
#[tokio::test]
async fn app_read() {
let test = AppTest::new().await;
let query = QueryAppRequest {
app_ids: vec![test.app.id.clone()],
};
let app_from_db = read_app(&test.sdk, query).await;
assert_eq!(app_from_db, test.app);
let mut test = FolderTest::new().await;
let app = test.app.clone();
test.run_scripts(vec![ReadApp(app.id.clone()), AssertApp(app)]).await;
}
#[tokio::test]
async fn app_update() {
let mut test = FolderTest::new().await;
let app = test.app.clone();
let new_name = "😁 hell world".to_owned();
assert_ne!(app.name, new_name);
test.run_scripts(vec![
UpdateApp {
name: Some(new_name.clone()),
desc: None,
},
ReadApp(app.id),
])
.await;
assert_eq!(test.app.name, new_name);
}
#[tokio::test]
async fn app_create_with_view() {
let test = AppTest::new().await;
let request_a = CreateViewRequest {
belong_to_id: test.app.id.clone(),
name: "View A".to_string(),
desc: "".to_string(),
thumbnail: Some("http://1.png".to_string()),
view_type: ViewType::Doc,
};
let mut test = FolderTest::new().await;
let mut app = test.app.clone();
test.run_scripts(vec![
CreateView {
name: "View A",
desc: "View A description",
},
CreateView {
name: "View B",
desc: "View B description",
},
ReadApp(app.id),
])
.await;
let request_b = CreateViewRequest {
belong_to_id: test.app.id.clone(),
name: "View B".to_string(),
desc: "".to_string(),
thumbnail: Some("http://1.png".to_string()),
view_type: ViewType::Doc,
};
app = test.app.clone();
assert_eq!(app.belongings.len(), 3);
assert_eq!(app.belongings[1].name, "View A");
assert_eq!(app.belongings[2].name, "View B")
}
let view_a = create_view_with_request(&test.sdk, request_a).await;
let view_b = create_view_with_request(&test.sdk, request_b).await;
#[tokio::test]
async fn view_update() {
let mut test = FolderTest::new().await;
let view = test.view.clone();
let new_name = "😁 123".to_owned();
assert_ne!(view.name, new_name);
let query = QueryAppRequest {
app_ids: vec![test.app.id.clone()],
};
let view_from_db = read_app(&test.sdk, query).await;
test.run_scripts(vec![
UpdateView {
name: Some(new_name.clone()),
desc: None,
},
ReadView(view.id),
])
.await;
assert_eq!(test.view.name, new_name);
}
assert_eq!(view_from_db.belongings[0], view_a);
assert_eq!(view_from_db.belongings[1], view_b);
#[tokio::test]
async fn open_document_view() {
let mut test = FolderTest::new().await;
assert_eq!(test.document_info, None);
test.run_scripts(vec![OpenDocument]).await;
let document_info = test.document_info.unwrap();
assert_eq!(document_info.text, initial_delta_string());
}
#[tokio::test]
#[should_panic]
async fn view_delete() {
let test = FlowySDKTest::default();
let _ = test.init_user().await;
let test = ViewTest::new(&test).await;
test.delete_views(vec![test.view.id.clone()]).await;
let query = QueryViewRequest {
view_ids: vec![test.view.id.clone()],
};
let _ = read_view(&test.sdk, query).await;
let mut test = FolderTest::new().await;
let view = test.view.clone();
test.run_scripts(vec![DeleteView, ReadView(view.id)]).await;
}
#[tokio::test]
async fn view_delete_then_putback() {
let test = FlowySDKTest::default();
let _ = test.init_user().await;
let test = ViewTest::new(&test).await;
test.delete_views(vec![test.view.id.clone()]).await;
putback_trash(
&test.sdk,
TrashId {
id: test.view.id.clone(),
ty: TrashType::View,
},
)
async fn view_delete_then_restore() {
let mut test = FolderTest::new().await;
let view = test.view.clone();
test.run_scripts(vec![
DeleteView,
RestoreViewFromTrash,
ReadView(view.id.clone()),
AssertView(view),
])
.await;
let query = QueryViewRequest {
view_ids: vec![test.view.id.clone()],
};
let view = read_view(&test.sdk, query).await;
assert_eq!(&view, &test.view);
}
#[tokio::test]
async fn view_delete_all() {
let test = FlowySDKTest::default();
let _ = test.init_user().await;
let mut test = FolderTest::new().await;
let app = test.app.clone();
test.run_scripts(vec![
CreateView {
name: "View A",
desc: "View A description",
},
CreateView {
name: "View B",
desc: "View B description",
},
ReadApp(app.id.clone()),
])
.await;
let test = ViewTest::new(&test).await;
let view1 = test.view.clone();
let view2 = create_view(&test.sdk, &test.app.id).await;
let view3 = create_view(&test.sdk, &test.app.id).await;
let view_ids = vec![view1.id.clone(), view2.id.clone(), view3.id.clone()];
assert_eq!(test.app.belongings.len(), 3);
let view_ids = test
.app
.belongings
.iter()
.map(|view| view.id.clone())
.collect::<Vec<String>>();
test.run_scripts(vec![DeleteViews(view_ids), ReadApp(app.id), ReadTrash])
.await;
let query = QueryAppRequest {
app_ids: vec![test.app.id.clone()],
};
let app = read_app(&test.sdk, query.clone()).await;
assert_eq!(app.belongings.len(), view_ids.len());
test.delete_views(view_ids.clone()).await;
assert_eq!(read_app(&test.sdk, query).await.belongings.len(), 0);
assert_eq!(read_trash(&test.sdk).await.len(), view_ids.len());
assert_eq!(test.app.belongings.len(), 0);
assert_eq!(test.trash.len(), 3);
}
#[tokio::test]
async fn view_delete_all_permanent() {
let test = FlowySDKTest::default();
let _ = test.init_user().await;
let mut test = FolderTest::new().await;
let app = test.app.clone();
test.run_scripts(vec![
CreateView {
name: "View A",
desc: "View A description",
},
ReadApp(app.id.clone()),
])
.await;
let test = ViewTest::new(&test).await;
let view1 = test.view.clone();
let view2 = create_view(&test.sdk, &test.app.id).await;
let view_ids = test
.app
.belongings
.iter()
.map(|view| view.id.clone())
.collect::<Vec<String>>();
test.run_scripts(vec![DeleteViews(view_ids), ReadApp(app.id), DeleteAllTrash, ReadTrash])
.await;
let view_ids = vec![view1.id.clone(), view2.id.clone()];
test.delete_views_permanent(view_ids).await;
let query = QueryAppRequest {
app_ids: vec![test.app.id.clone()],
};
assert_eq!(read_app(&test.sdk, query).await.belongings.len(), 0);
assert_eq!(read_trash(&test.sdk).await.len(), 0);
assert_eq!(test.app.belongings.len(), 0);
assert_eq!(test.trash.len(), 0);
}
#[tokio::test]
async fn view_open_doc() {
let test = FlowySDKTest::default();
let _ = test.init_user().await;
let test = ViewTest::new(&test).await;
let request = QueryViewRequest {
view_ids: vec![test.view.id.clone()],
};
let _ = open_view(&test.sdk, request).await;
async fn folder_sync_revision_state() {
let mut test = FolderTest::new().await;
test.run_scripts(vec![
AssertRevisionState {
rev_id: 1,
state: RevisionState::Sync,
},
AssertNextSyncRevId(Some(1)),
AssertRevisionState {
rev_id: 1,
state: RevisionState::Ack,
},
])
.await;
}
#[tokio::test]
async fn folder_sync_revision_seq() {
let mut test = FolderTest::new().await;
test.run_scripts(vec![
AssertRevisionState {
rev_id: 1,
state: RevisionState::Sync,
},
AssertRevisionState {
rev_id: 2,
state: RevisionState::Sync,
},
AssertRevisionState {
rev_id: 3,
state: RevisionState::Sync,
},
AssertNextSyncRevId(Some(1)),
AssertNextSyncRevId(Some(2)),
AssertNextSyncRevId(Some(3)),
AssertRevisionState {
rev_id: 1,
state: RevisionState::Ack,
},
AssertRevisionState {
rev_id: 2,
state: RevisionState::Ack,
},
AssertRevisionState {
rev_id: 3,
state: RevisionState::Ack,
},
])
.await;
}
#[tokio::test]
async fn folder_sync_revision_with_new_app() {
let mut test = FolderTest::new().await;
test.run_scripts(vec![
AssertNextSyncRevId(Some(1)),
AssertNextSyncRevId(Some(2)),
AssertNextSyncRevId(Some(3)),
CreateApp {
name: "New App",
desc: "",
},
AssertCurrentRevId(4),
AssertNextSyncRevId(Some(4)),
AssertNextSyncRevId(None),
])
.await;
}
#[tokio::test]
async fn folder_sync_revision_with_new_view() {
let mut test = FolderTest::new().await;
test.run_scripts(vec![
AssertNextSyncRevId(Some(1)),
AssertNextSyncRevId(Some(2)),
AssertNextSyncRevId(Some(3)),
CreateView {
name: "New App",
desc: "",
},
AssertCurrentRevId(4),
AssertNextSyncRevId(Some(4)),
AssertNextSyncRevId(None),
])
.await;
}

View File

@ -0,0 +1,209 @@
use flowy_collaboration::entities::document_info::DocumentInfo;
use flowy_core::event::WorkspaceEvent::*;
use flowy_core_data_model::entities::{
app::{App, AppId, CreateAppRequest, QueryAppRequest, UpdateAppRequest},
trash::{RepeatedTrash, TrashId, TrashType},
view::{CreateViewRequest, QueryViewRequest, UpdateViewRequest, View, ViewType},
workspace::{CreateWorkspaceRequest, QueryWorkspaceRequest, RepeatedWorkspace, Workspace},
};
use flowy_test::{event_builder::*, FlowySDKTest};
pub async fn create_workspace(sdk: &FlowySDKTest, name: &str, desc: &str) -> Workspace {
let request = CreateWorkspaceRequest {
name: name.to_owned(),
desc: desc.to_owned(),
};
let workspace = FolderEventBuilder::new(sdk.clone())
.event(CreateWorkspace)
.request(request)
.async_send()
.await
.parse::<Workspace>();
workspace
}
pub async fn read_workspace(sdk: &FlowySDKTest, workspace_id: Option<String>) -> Vec<Workspace> {
let request = QueryWorkspaceRequest { workspace_id };
let repeated_workspace = FolderEventBuilder::new(sdk.clone())
.event(ReadWorkspaces)
.request(request.clone())
.async_send()
.await
.parse::<RepeatedWorkspace>();
let workspaces;
if let Some(workspace_id) = &request.workspace_id {
workspaces = repeated_workspace
.into_inner()
.into_iter()
.filter(|workspace| &workspace.id == workspace_id)
.collect::<Vec<Workspace>>();
debug_assert_eq!(workspaces.len(), 1);
} else {
workspaces = repeated_workspace.items;
}
workspaces
}
pub async fn create_app(sdk: &FlowySDKTest, workspace_id: &str, name: &str, desc: &str) -> App {
let create_app_request = CreateAppRequest {
workspace_id: workspace_id.to_owned(),
name: name.to_string(),
desc: desc.to_string(),
color_style: Default::default(),
};
let app = FolderEventBuilder::new(sdk.clone())
.event(CreateApp)
.request(create_app_request)
.async_send()
.await
.parse::<App>();
app
}
pub async fn read_app(sdk: &FlowySDKTest, app_id: &str) -> App {
let request = QueryAppRequest {
app_ids: vec![app_id.to_owned()],
};
let app = FolderEventBuilder::new(sdk.clone())
.event(ReadApp)
.request(request)
.async_send()
.await
.parse::<App>();
app
}
pub async fn update_app(sdk: &FlowySDKTest, app_id: &str, name: Option<String>, desc: Option<String>) {
let request = UpdateAppRequest {
app_id: app_id.to_string(),
name,
desc,
color_style: None,
is_trash: None,
};
FolderEventBuilder::new(sdk.clone())
.event(UpdateApp)
.request(request)
.async_send()
.await;
}
pub async fn delete_app(sdk: &FlowySDKTest, app_id: &str) {
let request = AppId {
app_id: app_id.to_string(),
};
FolderEventBuilder::new(sdk.clone())
.event(DeleteApp)
.request(request)
.async_send()
.await;
}
pub async fn create_view(sdk: &FlowySDKTest, app_id: &str, name: &str, desc: &str, view_type: ViewType) -> View {
let request = CreateViewRequest {
belong_to_id: app_id.to_string(),
name: name.to_string(),
desc: desc.to_string(),
thumbnail: None,
view_type,
};
let view = FolderEventBuilder::new(sdk.clone())
.event(CreateView)
.request(request)
.async_send()
.await
.parse::<View>();
view
}
pub async fn read_view(sdk: &FlowySDKTest, view_ids: Vec<String>) -> View {
let request = QueryViewRequest { view_ids };
FolderEventBuilder::new(sdk.clone())
.event(ReadView)
.request(request)
.async_send()
.await
.parse::<View>()
}
pub async fn update_view(sdk: &FlowySDKTest, view_id: &str, name: Option<String>, desc: Option<String>) {
let request = UpdateViewRequest {
view_id: view_id.to_string(),
name,
desc,
thumbnail: None,
};
FolderEventBuilder::new(sdk.clone())
.event(UpdateView)
.request(request)
.async_send()
.await;
}
pub async fn delete_view(sdk: &FlowySDKTest, view_ids: Vec<String>) {
let request = QueryViewRequest { view_ids };
FolderEventBuilder::new(sdk.clone())
.event(DeleteView)
.request(request)
.async_send()
.await;
}
pub async fn open_document(sdk: &FlowySDKTest, view_id: &str) -> DocumentInfo {
let request = QueryViewRequest {
view_ids: vec![view_id.to_owned()],
};
FolderEventBuilder::new(sdk.clone())
.event(OpenDocument)
.request(request)
.async_send()
.await
.parse::<DocumentInfo>()
}
pub async fn read_trash(sdk: &FlowySDKTest) -> RepeatedTrash {
FolderEventBuilder::new(sdk.clone())
.event(ReadTrash)
.async_send()
.await
.parse::<RepeatedTrash>()
}
pub async fn restore_app_from_trash(sdk: &FlowySDKTest, app_id: &str) {
let id = TrashId {
id: app_id.to_owned(),
ty: TrashType::App,
};
FolderEventBuilder::new(sdk.clone())
.event(PutbackTrash)
.request(id)
.async_send()
.await;
}
pub async fn restore_view_from_trash(sdk: &FlowySDKTest, view_id: &str) {
let id = TrashId {
id: view_id.to_owned(),
ty: TrashType::View,
};
FolderEventBuilder::new(sdk.clone())
.event(PutbackTrash)
.request(id)
.async_send()
.await;
}
pub async fn delete_all_trash(sdk: &FlowySDKTest) {
FolderEventBuilder::new(sdk.clone())
.event(DeleteAllTrash)
.async_send()
.await;
}

View File

@ -1 +1,3 @@
mod folder_test;
mod helper;
mod script;

View File

@ -0,0 +1,218 @@
use crate::helper::*;
use flowy_collaboration::entities::{document_info::DocumentInfo, revision::RevisionState};
use flowy_core::{errors::ErrorCode, services::folder_editor::FolderEditor};
use flowy_core_data_model::entities::{
app::{App, RepeatedApp},
trash::Trash,
view::{RepeatedView, View, ViewType},
workspace::Workspace,
};
use flowy_sync::REVISION_WRITE_INTERVAL_IN_MILLIS;
use flowy_test::FlowySDKTest;
use std::{sync::Arc, time::Duration};
use tokio::time::sleep;
pub enum FolderScript {
// Workspace
ReadAllWorkspaces,
CreateWorkspace { name: String, desc: String },
AssertWorkspaceJson(String),
AssertWorkspace(Workspace),
ReadWorkspace(Option<String>),
// App
CreateApp { name: &'static str, desc: &'static str },
AssertAppJson(String),
AssertApp(App),
ReadApp(String),
UpdateApp { name: Option<String>, desc: Option<String> },
DeleteApp,
// View
CreateView { name: &'static str, desc: &'static str },
AssertView(View),
ReadView(String),
UpdateView { name: Option<String>, desc: Option<String> },
DeleteView,
DeleteViews(Vec<String>),
// Trash
RestoreAppFromTrash,
RestoreViewFromTrash,
ReadTrash,
DeleteAllTrash,
// Document
OpenDocument,
// Sync
AssertCurrentRevId(i64),
AssertNextSyncRevId(Option<i64>),
AssertRevisionState { rev_id: i64, state: RevisionState },
}
pub struct FolderTest {
pub sdk: FlowySDKTest,
pub all_workspace: Vec<Workspace>,
pub workspace: Workspace,
pub app: App,
pub view: View,
pub trash: Vec<Trash>,
pub document_info: Option<DocumentInfo>,
// pub folder_editor:
}
impl FolderTest {
pub async fn new() -> Self {
let sdk = FlowySDKTest::default();
let _ = sdk.init_user().await;
let mut workspace = create_workspace(&sdk, "FolderWorkspace", "Folder test workspace").await;
let mut app = create_app(&sdk, &workspace.id, "Folder App", "Folder test app").await;
let view = create_view(&sdk, &app.id, "Folder View", "Folder test view", ViewType::Doc).await;
app.belongings = RepeatedView {
items: vec![view.clone()],
};
workspace.apps = RepeatedApp {
items: vec![app.clone()],
};
Self {
sdk,
all_workspace: vec![],
workspace,
app,
view,
trash: vec![],
document_info: None,
}
}
pub async fn run_scripts(&mut self, scripts: Vec<FolderScript>) {
for script in scripts {
self.run_script(script).await;
}
}
pub async fn run_script(&mut self, script: FolderScript) {
let sdk = &self.sdk;
let folder_editor: Arc<FolderEditor> = sdk.folder_manager.folder_editor().await;
let rev_manager = folder_editor.rev_manager();
let cache = rev_manager.revision_cache();
match script {
FolderScript::ReadAllWorkspaces => {
let all_workspace = read_workspace(sdk, None).await;
self.all_workspace = all_workspace;
},
FolderScript::CreateWorkspace { name, desc } => {
let workspace = create_workspace(sdk, &name, &desc).await;
self.workspace = workspace;
},
FolderScript::AssertWorkspaceJson(expected_json) => {
let workspace = read_workspace(sdk, Some(self.workspace.id.clone()))
.await
.pop()
.unwrap();
let json = serde_json::to_string(&workspace).unwrap();
assert_eq!(json, expected_json);
},
FolderScript::AssertWorkspace(workspace) => {
assert_eq!(self.workspace, workspace);
},
FolderScript::ReadWorkspace(workspace_id) => {
let workspace = read_workspace(sdk, workspace_id).await.pop().unwrap();
self.workspace = workspace;
},
FolderScript::CreateApp { name, desc } => {
let app = create_app(&sdk, &self.workspace.id, name, desc).await;
self.app = app;
},
FolderScript::AssertAppJson(expected_json) => {
let json = serde_json::to_string(&self.app).unwrap();
assert_eq!(json, expected_json);
},
FolderScript::AssertApp(app) => {
assert_eq!(self.app, app);
},
FolderScript::ReadApp(app_id) => {
let app = read_app(&sdk, &app_id).await;
self.app = app;
},
FolderScript::UpdateApp { name, desc } => {
update_app(&sdk, &self.app.id, name, desc).await;
},
FolderScript::DeleteApp => {
delete_app(&sdk, &self.app.id).await;
},
FolderScript::CreateView { name, desc } => {
let view = create_view(&sdk, &self.app.id, name, desc, ViewType::Doc).await;
self.view = view;
},
FolderScript::AssertView(view) => {
assert_eq!(self.view, view);
},
FolderScript::ReadView(view_id) => {
let view = read_view(&sdk, vec![view_id]).await;
self.view = view;
},
FolderScript::UpdateView { name, desc } => {
update_view(&sdk, &self.view.id, name, desc).await;
},
FolderScript::DeleteView => {
delete_view(&sdk, vec![self.view.id.clone()]).await;
},
FolderScript::DeleteViews(view_ids) => {
delete_view(&sdk, view_ids).await;
},
FolderScript::RestoreAppFromTrash => {
restore_app_from_trash(&sdk, &self.app.id).await;
},
FolderScript::RestoreViewFromTrash => {
restore_view_from_trash(&sdk, &self.view.id).await;
},
FolderScript::ReadTrash => {
let trash = read_trash(&sdk).await;
self.trash = trash.into_inner();
},
FolderScript::DeleteAllTrash => {
delete_all_trash(&sdk).await;
self.trash = vec![];
},
FolderScript::OpenDocument => {
let document_info = open_document(&sdk, &self.view.id).await;
self.document_info = Some(document_info);
},
FolderScript::AssertRevisionState { rev_id, state } => {
let record = cache.get(rev_id).await.unwrap();
assert_eq!(record.state, state);
if let RevisionState::Ack = state {
// There is a defer action that writes the revisions to disk, so we wait here.
// Make sure everything is written.
sleep(Duration::from_millis(2 * REVISION_WRITE_INTERVAL_IN_MILLIS)).await;
}
},
FolderScript::AssertCurrentRevId(rev_id) => {
assert_eq!(rev_manager.rev_id(), rev_id);
},
FolderScript::AssertNextSyncRevId(rev_id) => {
let next_revision = rev_manager.next_sync_revision().await.unwrap();
if rev_id.is_none() {
assert_eq!(next_revision.is_none(), true, "Next revision should be None");
return;
}
let next_revision = next_revision.unwrap();
let mut receiver = rev_manager.revision_ack_receiver();
let _ = receiver.recv().await;
assert_eq!(next_revision.rev_id, rev_id.unwrap());
},
}
}
}
pub fn invalid_workspace_name_test_case() -> Vec<(String, ErrorCode)> {
vec![
("".to_owned(), ErrorCode::WorkspaceNameInvalid),
("1234".repeat(100), ErrorCode::WorkspaceNameTooLong),
]
}

View File

@ -47,6 +47,7 @@ pin-project = "1.0.0"
[dev-dependencies]
flowy-test = { path = "../flowy-test" }
flowy-document = { path = "../flowy-document", features = ["flowy_unit_test"]}
color-eyre = { version = "0.5", default-features = false }
criterion = "0.3"
rand = "0.7.3"

View File

@ -1,23 +0,0 @@
use crate::{controller::DocumentController, errors::FlowyError};
use flowy_database::ConnectionPool;
use std::sync::Arc;
pub trait DocumentUser: 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 DocumentContext {
pub controller: Arc<DocumentController>,
pub user: Arc<dyn DocumentUser>,
}
impl DocumentContext {
pub fn init(&self) -> Result<(), FlowyError> {
let _ = self.controller.init()?;
Ok(())
}
}

View File

@ -1,4 +1,4 @@
use crate::{context::DocumentUser, core::ClientDocumentEditor, errors::FlowyError, DocumentCloudService};
use crate::{core::ClientDocumentEditor, errors::FlowyError, DocumentCloudService};
use async_trait::async_trait;
use bytes::Bytes;
use dashmap::DashMap;
@ -14,13 +14,21 @@ use lib_infra::future::FutureResult;
use lib_ws::WSConnectState;
use std::{convert::TryInto, sync::Arc};
pub trait DocumentUser: 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>;
}
#[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 {
pub struct FlowyDocumentManager {
cloud_service: Arc<dyn DocumentCloudService>,
ws_receivers: WebSocketDataReceivers,
web_socket: Arc<dyn RevisionWebSocket>,
@ -28,7 +36,7 @@ pub struct DocumentController {
user: Arc<dyn DocumentUser>,
}
impl DocumentController {
impl FlowyDocumentManager {
pub fn new(
cloud_service: Arc<dyn DocumentCloudService>,
user: Arc<dyn DocumentUser>,
@ -45,7 +53,7 @@ impl DocumentController {
}
}
pub(crate) fn init(&self) -> FlowyResult<()> {
pub fn init(&self) -> FlowyResult<()> {
let notify = self.web_socket.subscribe_state_changed();
listen_ws_state_changed(notify, self.ws_receivers.clone());
@ -119,7 +127,7 @@ impl DocumentController {
}
}
impl DocumentController {
impl FlowyDocumentManager {
async fn get_editor(&self, doc_id: &str) -> FlowyResult<Arc<ClientDocumentEditor>> {
match self.open_cache.get(doc_id) {
None => {

View File

@ -1,7 +1,7 @@
use crate::{
context::DocumentUser,
core::{make_document_ws_manager, EditorCommand, EditorCommandQueue, EditorCommandSender},
errors::FlowyError,
DocumentUser,
DocumentWSReceiver,
};
use bytes::Bytes;

View File

@ -6,4 +6,4 @@ pub use editor::*;
pub(crate) use queue::*;
pub(crate) use web_socket::*;
pub const SYNC_INTERVAL_IN_MILLIS: u64 = 1000;
pub const DOCUMENT_SYNC_INTERVAL_IN_MILLIS: u64 = 1000;

View File

@ -1,4 +1,4 @@
use crate::{context::DocumentUser, core::web_socket::EditorCommandReceiver};
use crate::{core::web_socket::EditorCommandReceiver, DocumentUser};
use async_stream::stream;
use flowy_collaboration::{
client_document::{history::UndoResult, ClientDocument},

View File

@ -1,5 +1,5 @@
use crate::{
core::{EditorCommand, SYNC_INTERVAL_IN_MILLIS},
core::{EditorCommand, DOCUMENT_SYNC_INTERVAL_IN_MILLIS},
DocumentWSReceiver,
};
use async_trait::async_trait;
@ -46,7 +46,7 @@ pub(crate) async fn make_document_ws_manager(
});
let sink_provider = Arc::new(DocumentWSSinkDataProviderAdapter(composite_sink_provider));
let ping_duration = Duration::from_millis(SYNC_INTERVAL_IN_MILLIS);
let ping_duration = Duration::from_millis(DOCUMENT_SYNC_INTERVAL_IN_MILLIS);
let ws_manager = Arc::new(RevisionWebSocketManager::new(
&doc_id,
web_socket,

View File

@ -1,5 +1,4 @@
pub mod context;
mod controller;
pub mod controller;
pub mod core;
// mod notify;
pub mod protobuf;

View File

@ -1,5 +1,5 @@
use flowy_collaboration::entities::revision::RevisionState;
use flowy_document::core::{ClientDocumentEditor, SYNC_INTERVAL_IN_MILLIS};
use flowy_document::core::{ClientDocumentEditor, DOCUMENT_SYNC_INTERVAL_IN_MILLIS};
use flowy_test::{helper::ViewTest, FlowySDKTest};
use lib_ot::{core::Interval, rich_text::RichTextDelta};
use std::sync::Arc;
@ -26,7 +26,7 @@ impl EditorTest {
let sdk = FlowySDKTest::default();
let _ = sdk.init_user().await;
let test = ViewTest::new(&sdk).await;
let editor = sdk.document_ctx.controller.open_document(&test.view.id).await.unwrap();
let editor = sdk.document_manager.open_document(&test.view.id).await.unwrap();
Self { sdk, editor }
}
@ -79,6 +79,6 @@ impl EditorTest {
assert_eq!(expected_delta, delta);
},
}
sleep(Duration::from_millis(SYNC_INTERVAL_IN_MILLIS)).await;
sleep(Duration::from_millis(DOCUMENT_SYNC_INTERVAL_IN_MILLIS)).await;
}
}

View File

@ -16,7 +16,7 @@ use flowy_collaboration::{
use flowy_core::module::{FolderCouldServiceV1, FolderCouldServiceV2};
use flowy_error::{internal_error, FlowyError};
use futures_util::stream::StreamExt;
use lib_ws::{WSModule, WebSocketRawMessage};
use lib_ws::{WSChannel, WebSocketRawMessage};
use parking_lot::RwLock;
use std::{
convert::{TryFrom, TryInto},
@ -113,12 +113,12 @@ impl LocalWebSocketRunner {
async fn handle_message(&self, message: WebSocketRawMessage) -> Result<(), FlowyError> {
let bytes = Bytes::from(message.data);
let client_data = ClientRevisionWSData::try_from(bytes).map_err(internal_error)?;
match message.module {
WSModule::Doc => {
match message.channel {
WSChannel::Document => {
let _ = self.handle_document_client_data(client_data, "".to_owned()).await?;
Ok(())
},
WSModule::Folder => {
WSChannel::Folder => {
let _ = self.handle_folder_client_data(client_data, "".to_owned()).await?;
Ok(())
},
@ -140,6 +140,7 @@ impl LocalWebSocketRunner {
let user = Arc::new(LocalRevisionUser {
user_id,
client_ws_sender,
channel: WSChannel::Folder,
});
let ty = client_data.ty.clone();
let document_client_data: ClientRevisionWSDataPB = client_data.try_into().unwrap();
@ -175,6 +176,7 @@ impl LocalWebSocketRunner {
let user = Arc::new(LocalRevisionUser {
user_id,
client_ws_sender,
channel: WSChannel::Document,
});
let ty = client_data.ty.clone();
let document_client_data: ClientRevisionWSDataPB = client_data.try_into().unwrap();
@ -197,6 +199,7 @@ impl LocalWebSocketRunner {
struct LocalRevisionUser {
user_id: String,
client_ws_sender: mpsc::UnboundedSender<WebSocketRawMessage>,
channel: WSChannel,
}
impl RevisionUser for LocalRevisionUser {
@ -210,13 +213,14 @@ impl RevisionUser for LocalRevisionUser {
tracing::error!("LocalDocumentUser send message failed: {}", e);
},
};
let channel = self.channel.clone();
tokio::spawn(async move {
match resp {
RevisionSyncResponse::Pull(data) => {
let bytes: Bytes = data.try_into().unwrap();
let msg = WebSocketRawMessage {
module: WSModule::Doc,
channel,
data: bytes.to_vec(),
};
send_fn(sender, msg);
@ -224,7 +228,7 @@ impl RevisionUser for LocalRevisionUser {
RevisionSyncResponse::Push(data) => {
let bytes: Bytes = data.try_into().unwrap();
let msg = WebSocketRawMessage {
module: WSModule::Doc,
channel,
data: bytes.to_vec(),
};
send_fn(sender, msg);
@ -232,7 +236,7 @@ impl RevisionUser for LocalRevisionUser {
RevisionSyncResponse::Ack(data) => {
let bytes: Bytes = data.try_into().unwrap();
let msg = WebSocketRawMessage {
module: WSModule::Doc,
channel,
data: bytes.to_vec(),
};
send_fn(sender, msg);

View File

@ -2,14 +2,14 @@ use crate::ws::connection::{FlowyRawWebSocket, FlowyWebSocket};
use dashmap::DashMap;
use flowy_error::FlowyError;
use lib_infra::future::FutureResult;
use lib_ws::{WSConnectState, WSMessageReceiver, WSModule, WebSocketRawMessage};
use lib_ws::{WSChannel, WSConnectState, WSMessageReceiver, WebSocketRawMessage};
use parking_lot::RwLock;
use std::sync::Arc;
use tokio::sync::{broadcast, broadcast::Receiver, mpsc::UnboundedReceiver};
pub struct LocalWebSocket {
user_id: Arc<RwLock<Option<String>>>,
receivers: Arc<DashMap<WSModule, Arc<dyn WSMessageReceiver>>>,
receivers: Arc<DashMap<WSChannel, Arc<dyn WSMessageReceiver>>>,
state_sender: broadcast::Sender<WSConnectState>,
server_ws_receiver: RwLock<Option<UnboundedReceiver<WebSocketRawMessage>>>,
server_ws_sender: broadcast::Sender<WebSocketRawMessage>,
@ -40,7 +40,7 @@ impl FlowyRawWebSocket for LocalWebSocket {
let receivers = self.receivers.clone();
tokio::spawn(async move {
while let Some(message) = server_ws_receiver.recv().await {
match receivers.get(&message.module) {
match receivers.get(&message.channel) {
None => tracing::error!("Can't find any handler for message: {:?}", message),
Some(receiver) => receiver.receive_message(message.clone()),
}
@ -61,6 +61,7 @@ impl FlowyRawWebSocket for LocalWebSocket {
fn reconnect(&self, _count: usize) -> FutureResult<(), FlowyError> { FutureResult::new(async { Ok(()) }) }
fn add_receiver(&self, receiver: Arc<dyn WSMessageReceiver>) -> Result<(), FlowyError> {
tracing::trace!("Local web socket add ws receiver: {:?}", receiver.source());
self.receivers.insert(receiver.source(), receiver);
Ok(())
}

View File

@ -3,10 +3,10 @@ use bytes::Bytes;
use flowy_collaboration::entities::ws_data::ClientRevisionWSData;
use flowy_database::ConnectionPool;
use flowy_document::{
context::{DocumentContext, DocumentUser},
errors::{internal_error, FlowyError},
DocumentCloudService,
DocumentController,
DocumentUser,
FlowyDocumentManager,
};
use flowy_net::{
http_server::document::DocumentHttpCloudService,
@ -15,7 +15,7 @@ use flowy_net::{
};
use flowy_sync::{RevisionWebSocket, WSStateReceiver};
use flowy_user::services::UserSession;
use lib_ws::{WSMessageReceiver, WSModule, WebSocketRawMessage};
use lib_ws::{WSChannel, WSMessageReceiver, WebSocketRawMessage};
use std::{convert::TryInto, path::Path, sync::Arc};
pub struct DocumentDepsResolver();
@ -25,7 +25,7 @@ impl DocumentDepsResolver {
ws_conn: Arc<FlowyWebSocketConnect>,
user_session: Arc<UserSession>,
server_config: &ClientServerConfiguration,
) -> DocumentContext {
) -> Arc<FlowyDocumentManager> {
let user = Arc::new(DocumentUserImpl(user_session));
let ws_sender = Arc::new(DocumentWebSocketImpl(ws_conn.clone()));
let cloud_service: Arc<dyn DocumentCloudService> = match local_server {
@ -33,14 +33,11 @@ impl DocumentDepsResolver {
Some(local_server) => local_server,
};
let document_controller = Arc::new(DocumentController::new(cloud_service, user.clone(), ws_sender));
let receiver = Arc::new(DocumentWSMessageReceiverImpl(document_controller.clone()));
let manager = Arc::new(FlowyDocumentManager::new(cloud_service, user, ws_sender));
let receiver = Arc::new(DocumentWSMessageReceiverImpl(manager.clone()));
ws_conn.add_ws_message_receiver(receiver).unwrap();
DocumentContext {
controller: document_controller,
user,
}
manager
}
}
@ -68,7 +65,7 @@ impl RevisionWebSocket for DocumentWebSocketImpl {
fn send(&self, data: ClientRevisionWSData) -> Result<(), FlowyError> {
let bytes: Bytes = data.try_into().unwrap();
let msg = WebSocketRawMessage {
module: WSModule::Doc,
channel: WSChannel::Document,
data: bytes.to_vec(),
};
let sender = self.0.web_socket()?;
@ -79,9 +76,9 @@ impl RevisionWebSocket for DocumentWebSocketImpl {
fn subscribe_state_changed(&self) -> WSStateReceiver { self.0.subscribe_websocket_state() }
}
struct DocumentWSMessageReceiverImpl(Arc<DocumentController>);
struct DocumentWSMessageReceiverImpl(Arc<FlowyDocumentManager>);
impl WSMessageReceiver for DocumentWSMessageReceiverImpl {
fn source(&self) -> WSModule { WSModule::Doc }
fn source(&self) -> WSChannel { WSChannel::Document }
fn receive_message(&self, msg: WebSocketRawMessage) {
let handler = self.0.clone();
tokio::spawn(async move {

View File

@ -7,7 +7,7 @@ use flowy_core::{
module::{FolderCouldServiceV1, WorkspaceDatabase, WorkspaceUser},
};
use flowy_database::ConnectionPool;
use flowy_document::context::DocumentContext;
use flowy_document::FlowyDocumentManager;
use flowy_net::{
http_server::core::CoreHttpCloudService,
local_server::LocalServer,
@ -15,7 +15,7 @@ use flowy_net::{
};
use flowy_sync::{RevisionWebSocket, WSStateReceiver};
use flowy_user::services::UserSession;
use lib_ws::{WSMessageReceiver, WSModule, WebSocketRawMessage};
use lib_ws::{WSChannel, WSMessageReceiver, WebSocketRawMessage};
use std::{convert::TryInto, sync::Arc};
pub struct FolderDepsResolver();
@ -24,7 +24,7 @@ impl FolderDepsResolver {
local_server: Option<Arc<LocalServer>>,
user_session: Arc<UserSession>,
server_config: &ClientServerConfiguration,
flowy_document: &Arc<DocumentContext>,
document_manager: &Arc<FlowyDocumentManager>,
ws_conn: Arc<FlowyWebSocketConnect>,
) -> Arc<FolderManager> {
let user: Arc<dyn WorkspaceUser> = Arc::new(WorkspaceUserImpl(user_session.clone()));
@ -39,7 +39,7 @@ impl FolderDepsResolver {
user,
cloud_service,
database,
flowy_document.clone(),
document_manager.clone(),
web_socket,
));
@ -69,7 +69,7 @@ impl RevisionWebSocket for FolderWebSocketImpl {
fn send(&self, data: ClientRevisionWSData) -> Result<(), FlowyError> {
let bytes: Bytes = data.try_into().unwrap();
let msg = WebSocketRawMessage {
module: WSModule::Folder,
channel: WSChannel::Folder,
data: bytes.to_vec(),
};
let sender = self.0.web_socket()?;
@ -82,7 +82,7 @@ impl RevisionWebSocket for FolderWebSocketImpl {
struct FolderWSMessageReceiverImpl(Arc<FolderManager>);
impl WSMessageReceiver for FolderWSMessageReceiverImpl {
fn source(&self) -> WSModule { WSModule::Folder }
fn source(&self) -> WSChannel { WSChannel::Folder }
fn receive_message(&self, msg: WebSocketRawMessage) {
let handler = self.0.clone();
tokio::spawn(async move {

View File

@ -3,7 +3,6 @@ pub mod module;
use crate::deps_resolve::*;
use backend_service::configuration::ClientServerConfiguration;
use flowy_core::{controller::FolderManager, errors::FlowyError};
use flowy_document::context::DocumentContext;
use flowy_net::{
entities::NetworkType,
local_server::LocalServer,
@ -12,6 +11,7 @@ use flowy_net::{
use flowy_user::services::{notifier::UserStatus, UserSession, UserSessionConfig};
use lib_dispatch::prelude::*;
use flowy_document::FlowyDocumentManager;
use module::mk_modules;
pub use module::*;
use std::{
@ -83,8 +83,8 @@ pub struct FlowySDK {
#[allow(dead_code)]
config: FlowySDKConfig,
pub user_session: Arc<UserSession>,
pub document_ctx: Arc<DocumentContext>,
pub core: Arc<FolderManager>,
pub document_manager: Arc<FlowyDocumentManager>,
pub folder_manager: Arc<FolderManager>,
pub dispatcher: Arc<EventDispatcher>,
pub ws_conn: Arc<FlowyWebSocketConnect>,
pub local_server: Option<Arc<LocalServer>>,
@ -108,25 +108,25 @@ impl FlowySDK {
};
let user_session = mk_user_session(&config, &local_server, &config.server_config);
let flowy_document = mk_document(&local_server, &ws_conn, &user_session, &config.server_config);
let core_ctx = mk_core_context(
let document_manager = mk_document(&local_server, &ws_conn, &user_session, &config.server_config);
let folder_manager = mk_folder_manager(
&local_server,
&user_session,
&flowy_document,
&document_manager,
&config.server_config,
&ws_conn,
);
//
let modules = mk_modules(&ws_conn, &core_ctx, &user_session);
let modules = mk_modules(&ws_conn, &folder_manager, &user_session);
let dispatcher = Arc::new(EventDispatcher::construct(|| modules));
_init(&local_server, &dispatcher, &ws_conn, &user_session, &core_ctx);
_init(&local_server, &dispatcher, &ws_conn, &user_session, &folder_manager);
Self {
config,
user_session,
document_ctx: flowy_document,
core: core_ctx,
document_manager,
folder_manager,
dispatcher,
ws_conn,
local_server,
@ -243,10 +243,10 @@ fn mk_user_session(
Arc::new(UserSession::new(user_config, cloud_service))
}
fn mk_core_context(
fn mk_folder_manager(
local_server: &Option<Arc<LocalServer>>,
user_session: &Arc<UserSession>,
flowy_document: &Arc<DocumentContext>,
document_manager: &Arc<FlowyDocumentManager>,
server_config: &ClientServerConfiguration,
ws_conn: &Arc<FlowyWebSocketConnect>,
) -> Arc<FolderManager> {
@ -254,7 +254,7 @@ fn mk_core_context(
local_server.clone(),
user_session.clone(),
server_config,
flowy_document,
document_manager,
ws_conn.clone(),
)
}
@ -264,11 +264,11 @@ pub fn mk_document(
ws_conn: &Arc<FlowyWebSocketConnect>,
user_session: &Arc<UserSession>,
server_config: &ClientServerConfiguration,
) -> Arc<DocumentContext> {
Arc::new(DocumentDepsResolver::resolve(
) -> Arc<FlowyDocumentManager> {
DocumentDepsResolver::resolve(
local_server.clone(),
ws_conn.clone(),
user_session.clone(),
server_config,
))
)
}

View File

@ -1,4 +1,4 @@
use crate::RevisionRecord;
use crate::{RevisionRecord, REVISION_WRITE_INTERVAL_IN_MILLIS};
use dashmap::DashMap;
use flowy_collaboration::entities::revision::RevisionRange;
use flowy_error::{FlowyError, FlowyResult};
@ -113,7 +113,7 @@ impl RevisionMemoryCache {
let delegate = self.delegate.clone();
*self.defer_save.write().await = Some(tokio::spawn(async move {
tokio::time::sleep(Duration::from_millis(600)).await;
tokio::time::sleep(Duration::from_millis(REVISION_WRITE_INTERVAL_IN_MILLIS)).await;
let mut revs_write_guard = pending_write_revs.write().await;
// It may cause performance issues because we hold the write lock of the
// rev_order and the lock will be released after the checkpoint has been written

View File

@ -16,6 +16,7 @@ use std::{
},
};
use tokio::task::spawn_blocking;
pub const REVISION_WRITE_INTERVAL_IN_MILLIS: u64 = 600;
pub struct RevisionCache {
object_id: String,

View File

@ -8,7 +8,8 @@ use flowy_error::{FlowyError, FlowyResult};
use futures_util::{future, stream, stream::StreamExt};
use lib_infra::future::FutureResult;
use std::{collections::VecDeque, sync::Arc};
use tokio::sync::RwLock;
use tokio::sync::{broadcast, RwLock};
pub trait RevisionCloudService: Send + Sync {
fn fetch_object(&self, user_id: &str, object_id: &str) -> FutureResult<Vec<Revision>, FlowyError>;
@ -25,18 +26,27 @@ pub struct RevisionManager {
rev_id_counter: RevIdCounter,
revision_cache: Arc<RevisionCache>,
revision_sync_seq: Arc<RevisionSyncSequence>,
#[cfg(feature = "flowy_unit_test")]
revision_ack_notifier: broadcast::Sender<i64>,
}
impl RevisionManager {
pub fn new(user_id: &str, object_id: &str, revision_cache: Arc<RevisionCache>) -> Self {
let rev_id_counter = RevIdCounter::new(0);
let revision_sync_seq = Arc::new(RevisionSyncSequence::new());
#[cfg(feature = "flowy_unit_test")]
let (revision_ack_notifier, _) = broadcast::channel(1);
Self {
object_id: object_id.to_string(),
user_id: user_id.to_owned(),
rev_id_counter,
revision_cache,
revision_sync_seq,
#[cfg(feature = "flowy_unit_test")]
revision_ack_notifier,
}
}
@ -98,6 +108,9 @@ impl RevisionManager {
pub async fn ack_revision(&self, rev_id: i64) -> Result<(), FlowyError> {
if self.revision_sync_seq.ack(&rev_id).await.is_ok() {
self.revision_cache.ack(rev_id).await;
#[cfg(feature = "flowy_unit_test")]
let _ = self.revision_ack_notifier.send(rev_id);
}
Ok(())
}
@ -256,4 +269,5 @@ impl RevisionSyncSequence {
#[cfg(feature = "flowy_unit_test")]
impl RevisionManager {
pub fn revision_cache(&self) -> Arc<RevisionCache> { self.revision_cache.clone() }
pub fn revision_ack_receiver(&self) -> broadcast::Receiver<i64> { self.revision_ack_notifier.subscribe() }
}

View File

@ -9,13 +9,13 @@ use std::{
sync::Arc,
};
pub type CoreModuleEventBuilder = EventBuilder<FlowyError>;
impl CoreModuleEventBuilder {
pub type FolderEventBuilder = EventBuilder<FlowyError>;
impl FolderEventBuilder {
pub fn new(sdk: FlowySDKTest) -> Self { EventBuilder::test(TestContext::new(sdk)) }
pub fn user_profile(&self) -> &Option<UserProfile> { &self.user_profile }
}
pub type UserModuleEventBuilder = CoreModuleEventBuilder;
pub type UserModuleEventBuilder = FolderEventBuilder;
#[derive(Clone)]
pub struct EventBuilder<E> {

View File

@ -1,13 +1,11 @@
use crate::prelude::*;
use flowy_collaboration::entities::document_info::DocumentInfo;
use flowy_core::{
entities::{
app::*,
trash::{RepeatedTrash, TrashId},
view::*,
workspace::{CreateWorkspaceRequest, QueryWorkspaceRequest, Workspace, *},
workspace::{CreateWorkspaceRequest, QueryWorkspaceRequest, Workspace},
},
errors::ErrorCode,
event::WorkspaceEvent::{CreateWorkspace, OpenWorkspace, *},
};
use flowy_user::{
@ -15,55 +13,10 @@ use flowy_user::{
errors::FlowyError,
event::UserEvent::{InitUser, SignIn, SignOut, SignUp},
};
use lib_dispatch::prelude::{EventDispatcher, ModuleRequest, ToBytes};
use lib_infra::uuid_string;
use std::{fs, path::PathBuf, sync::Arc};
pub struct WorkspaceTest {
pub sdk: FlowySDKTest,
pub workspace: Workspace,
}
impl WorkspaceTest {
pub async fn new() -> Self {
let sdk = FlowySDKTest::default();
let _ = sdk.init_user().await;
let workspace = create_workspace(&sdk, "Workspace", "").await;
open_workspace(&sdk, &workspace.id).await;
Self { sdk, workspace }
}
}
pub struct AppTest {
pub sdk: FlowySDKTest,
pub workspace: Workspace,
pub app: App,
}
impl AppTest {
pub async fn new() -> Self {
let sdk = FlowySDKTest::default();
let _ = sdk.init_user().await;
let workspace = create_workspace(&sdk, "Workspace", "").await;
open_workspace(&sdk, &workspace.id).await;
let app = create_app(&sdk, "App", "AppFlowy GitHub Project", &workspace.id).await;
Self { sdk, workspace, app }
}
pub async fn move_app_to_trash(&self) {
let request = UpdateAppRequest {
app_id: self.app.id.clone(),
name: None,
desc: None,
color_style: None,
is_trash: Some(true),
};
update_app(&self.sdk, request).await;
}
}
pub struct ViewTest {
pub sdk: FlowySDKTest,
pub workspace: Workspace,
@ -84,37 +37,15 @@ impl ViewTest {
view,
}
}
pub async fn delete_views(&self, view_ids: Vec<String>) {
let request = QueryViewRequest { view_ids };
delete_view(&self.sdk, request).await;
}
pub async fn delete_views_permanent(&self, view_ids: Vec<String>) {
let request = QueryViewRequest { view_ids };
delete_view(&self.sdk, request).await;
CoreModuleEventBuilder::new(self.sdk.clone())
.event(DeleteAll)
.async_send()
.await;
}
}
pub fn invalid_workspace_name_test_case() -> Vec<(String, ErrorCode)> {
vec![
("".to_owned(), ErrorCode::WorkspaceNameInvalid),
("1234".repeat(100), ErrorCode::WorkspaceNameTooLong),
]
}
pub async fn create_workspace(sdk: &FlowySDKTest, name: &str, desc: &str) -> Workspace {
async fn create_workspace(sdk: &FlowySDKTest, name: &str, desc: &str) -> Workspace {
let request = CreateWorkspaceRequest {
name: name.to_owned(),
desc: desc.to_owned(),
};
let workspace = CoreModuleEventBuilder::new(sdk.clone())
let workspace = FolderEventBuilder::new(sdk.clone())
.event(CreateWorkspace)
.request(request)
.async_send()
@ -127,37 +58,14 @@ async fn open_workspace(sdk: &FlowySDKTest, workspace_id: &str) {
let request = QueryWorkspaceRequest {
workspace_id: Some(workspace_id.to_owned()),
};
let _ = CoreModuleEventBuilder::new(sdk.clone())
let _ = FolderEventBuilder::new(sdk.clone())
.event(OpenWorkspace)
.request(request)
.async_send()
.await;
}
pub async fn read_workspace(sdk: &FlowySDKTest, request: QueryWorkspaceRequest) -> Vec<Workspace> {
let repeated_workspace = CoreModuleEventBuilder::new(sdk.clone())
.event(ReadWorkspaces)
.request(request.clone())
.async_send()
.await
.parse::<RepeatedWorkspace>();
let workspaces;
if let Some(workspace_id) = &request.workspace_id {
workspaces = repeated_workspace
.into_inner()
.into_iter()
.filter(|workspace| &workspace.id == workspace_id)
.collect::<Vec<Workspace>>();
debug_assert_eq!(workspaces.len(), 1);
} else {
workspaces = repeated_workspace.items;
}
workspaces
}
pub async fn create_app(sdk: &FlowySDKTest, name: &str, desc: &str, workspace_id: &str) -> App {
async fn create_app(sdk: &FlowySDKTest, name: &str, desc: &str, workspace_id: &str) -> App {
let create_app_request = CreateAppRequest {
workspace_id: workspace_id.to_owned(),
name: name.to_string(),
@ -165,7 +73,7 @@ pub async fn create_app(sdk: &FlowySDKTest, name: &str, desc: &str, workspace_id
color_style: Default::default(),
};
let app = CoreModuleEventBuilder::new(sdk.clone())
let app = FolderEventBuilder::new(sdk.clone())
.event(CreateApp)
.request(create_app_request)
.async_send()
@ -174,48 +82,7 @@ pub async fn create_app(sdk: &FlowySDKTest, name: &str, desc: &str, workspace_id
app
}
pub async fn delete_app(sdk: &FlowySDKTest, app_id: &str) {
let delete_app_request = AppId {
app_id: app_id.to_string(),
};
CoreModuleEventBuilder::new(sdk.clone())
.event(DeleteApp)
.request(delete_app_request)
.async_send()
.await;
}
pub async fn update_app(sdk: &FlowySDKTest, request: UpdateAppRequest) {
CoreModuleEventBuilder::new(sdk.clone())
.event(UpdateApp)
.request(request)
.async_send()
.await;
}
pub async fn read_app(sdk: &FlowySDKTest, request: QueryAppRequest) -> App {
let app = CoreModuleEventBuilder::new(sdk.clone())
.event(ReadApp)
.request(request)
.async_send()
.await
.parse::<App>();
app
}
pub async fn create_view_with_request(sdk: &FlowySDKTest, request: CreateViewRequest) -> View {
let view = CoreModuleEventBuilder::new(sdk.clone())
.event(CreateView)
.request(request)
.async_send()
.await
.parse::<View>();
view
}
pub async fn create_view(sdk: &FlowySDKTest, app_id: &str) -> View {
async fn create_view(sdk: &FlowySDKTest, app_id: &str) -> View {
let request = CreateViewRequest {
belong_to_id: app_id.to_string(),
name: "View A".to_string(),
@ -224,57 +91,13 @@ pub async fn create_view(sdk: &FlowySDKTest, app_id: &str) -> View {
view_type: ViewType::Doc,
};
create_view_with_request(sdk, request).await
}
pub async fn update_view(sdk: &FlowySDKTest, request: UpdateViewRequest) {
CoreModuleEventBuilder::new(sdk.clone())
.event(UpdateView)
.request(request)
.async_send()
.await;
}
pub async fn read_view(sdk: &FlowySDKTest, request: QueryViewRequest) -> View {
CoreModuleEventBuilder::new(sdk.clone())
.event(ReadView)
let view = FolderEventBuilder::new(sdk.clone())
.event(CreateView)
.request(request)
.async_send()
.await
.parse::<View>()
}
pub async fn delete_view(sdk: &FlowySDKTest, request: QueryViewRequest) {
CoreModuleEventBuilder::new(sdk.clone())
.event(DeleteView)
.request(request)
.async_send()
.await;
}
pub async fn read_trash(sdk: &FlowySDKTest) -> RepeatedTrash {
CoreModuleEventBuilder::new(sdk.clone())
.event(ReadTrash)
.async_send()
.await
.parse::<RepeatedTrash>()
}
pub async fn putback_trash(sdk: &FlowySDKTest, id: TrashId) {
CoreModuleEventBuilder::new(sdk.clone())
.event(PutbackTrash)
.request(id)
.async_send()
.await;
}
pub async fn open_view(sdk: &FlowySDKTest, request: QueryViewRequest) -> DocumentInfo {
CoreModuleEventBuilder::new(sdk.clone())
.event(OpenView)
.request(request)
.async_send()
.await
.parse::<DocumentInfo>()
.parse::<View>();
view
}
pub fn root_dir() -> String {

View File

@ -2,7 +2,7 @@ use crate::{
client_document::{
default::initial_delta,
history::{History, UndoResult},
view::{View, RECORD_THRESHOLD},
view::{ViewExtensions, RECORD_THRESHOLD},
},
errors::CollaborateError,
};
@ -29,7 +29,7 @@ impl InitialDocumentText for NewlineDoc {
pub struct ClientDocument {
delta: RichTextDelta,
history: History,
view: View,
view: ViewExtensions,
last_edit_time: usize,
notify: Option<mpsc::UnboundedSender<()>>,
}
@ -41,7 +41,7 @@ impl ClientDocument {
ClientDocument {
delta,
history: History::new(),
view: View::new(),
view: ViewExtensions::new(),
last_edit_time: 0,
notify: None,
}

View File

@ -7,13 +7,13 @@ use lib_ot::{
pub const RECORD_THRESHOLD: usize = 400; // in milliseconds
pub struct View {
pub struct ViewExtensions {
insert_exts: Vec<InsertExtension>,
format_exts: Vec<FormatExtension>,
delete_exts: Vec<DeleteExtension>,
}
impl View {
impl ViewExtensions {
pub(crate) fn new() -> Self {
Self {
insert_exts: construct_insert_exts(),

View File

@ -3,7 +3,3 @@ pub mod share;
pub mod trash;
pub mod view;
pub mod workspace;
pub mod prelude {
pub use crate::entities::{app::*, share::*, trash::*, view::*, workspace::*};
}

View File

@ -116,13 +116,13 @@ pub struct CurrentWorkspaceSetting {
#[derive(ProtoBuf, Default)]
pub struct UpdateWorkspaceRequest {
#[pb(index = 1)]
id: String,
pub id: String,
#[pb(index = 2, one_of)]
name: Option<String>,
pub name: Option<String>,
#[pb(index = 3, one_of)]
desc: Option<String>,
pub desc: Option<String>,
}
#[derive(Clone, ProtoBuf, Default, Debug)]

View File

@ -96,7 +96,7 @@ pub fn category_from_str(type_str: &str) -> TypeCategory {
| "TrashType"
| "ViewType"
| "ErrorCode"
| "WSModule"
| "WSChannel"
=> TypeCategory::Enum,
"Option" => TypeCategory::Opt,

View File

@ -6,27 +6,27 @@ use tokio_tungstenite::tungstenite::Message as TokioMessage;
#[derive(ProtoBuf, Debug, Clone, Default)]
pub struct WebSocketRawMessage {
#[pb(index = 1)]
pub module: WSModule,
pub channel: WSChannel,
#[pb(index = 2)]
pub data: Vec<u8>,
}
#[derive(ProtoBuf_Enum, Debug, Clone, Eq, PartialEq, Hash)]
pub enum WSModule {
Doc = 0,
Folder = 1,
pub enum WSChannel {
Document = 0,
Folder = 1,
}
impl std::default::Default for WSModule {
fn default() -> Self { WSModule::Doc }
impl std::default::Default for WSChannel {
fn default() -> Self { WSChannel::Document }
}
impl ToString for WSModule {
impl ToString for WSChannel {
fn to_string(&self) -> String {
match self {
WSModule::Doc => "0".to_string(),
WSModule::Folder => "1".to_string(),
WSChannel::Document => "0".to_string(),
WSChannel::Folder => "1".to_string(),
}
}
}

View File

@ -26,7 +26,7 @@
#[derive(PartialEq,Clone,Default)]
pub struct WebSocketRawMessage {
// message fields
pub module: WSModule,
pub channel: WSChannel,
pub data: ::std::vec::Vec<u8>,
// special fields
pub unknown_fields: ::protobuf::UnknownFields,
@ -44,19 +44,19 @@ impl WebSocketRawMessage {
::std::default::Default::default()
}
// .WSModule module = 1;
// .WSChannel channel = 1;
pub fn get_module(&self) -> WSModule {
self.module
pub fn get_channel(&self) -> WSChannel {
self.channel
}
pub fn clear_module(&mut self) {
self.module = WSModule::Doc;
pub fn clear_channel(&mut self) {
self.channel = WSChannel::Document;
}
// Param is passed by value, moved
pub fn set_module(&mut self, v: WSModule) {
self.module = v;
pub fn set_channel(&mut self, v: WSChannel) {
self.channel = v;
}
// bytes data = 2;
@ -96,7 +96,7 @@ impl ::protobuf::Message for WebSocketRawMessage {
let (field_number, wire_type) = is.read_tag_unpack()?;
match field_number {
1 => {
::protobuf::rt::read_proto3_enum_with_unknown_fields_into(wire_type, is, &mut self.module, 1, &mut self.unknown_fields)?
::protobuf::rt::read_proto3_enum_with_unknown_fields_into(wire_type, is, &mut self.channel, 1, &mut self.unknown_fields)?
},
2 => {
::protobuf::rt::read_singular_proto3_bytes_into(wire_type, is, &mut self.data)?;
@ -113,8 +113,8 @@ impl ::protobuf::Message for WebSocketRawMessage {
#[allow(unused_variables)]
fn compute_size(&self) -> u32 {
let mut my_size = 0;
if self.module != WSModule::Doc {
my_size += ::protobuf::rt::enum_size(1, self.module);
if self.channel != WSChannel::Document {
my_size += ::protobuf::rt::enum_size(1, self.channel);
}
if !self.data.is_empty() {
my_size += ::protobuf::rt::bytes_size(2, &self.data);
@ -125,8 +125,8 @@ impl ::protobuf::Message for WebSocketRawMessage {
}
fn write_to_with_cached_sizes(&self, os: &mut ::protobuf::CodedOutputStream<'_>) -> ::protobuf::ProtobufResult<()> {
if self.module != WSModule::Doc {
os.write_enum(1, ::protobuf::ProtobufEnum::value(&self.module))?;
if self.channel != WSChannel::Document {
os.write_enum(1, ::protobuf::ProtobufEnum::value(&self.channel))?;
}
if !self.data.is_empty() {
os.write_bytes(2, &self.data)?;
@ -169,10 +169,10 @@ impl ::protobuf::Message for WebSocketRawMessage {
static descriptor: ::protobuf::rt::LazyV2<::protobuf::reflect::MessageDescriptor> = ::protobuf::rt::LazyV2::INIT;
descriptor.get(|| {
let mut fields = ::std::vec::Vec::new();
fields.push(::protobuf::reflect::accessor::make_simple_field_accessor::<_, ::protobuf::types::ProtobufTypeEnum<WSModule>>(
"module",
|m: &WebSocketRawMessage| { &m.module },
|m: &mut WebSocketRawMessage| { &mut m.module },
fields.push(::protobuf::reflect::accessor::make_simple_field_accessor::<_, ::protobuf::types::ProtobufTypeEnum<WSChannel>>(
"channel",
|m: &WebSocketRawMessage| { &m.channel },
|m: &mut WebSocketRawMessage| { &mut m.channel },
));
fields.push(::protobuf::reflect::accessor::make_simple_field_accessor::<_, ::protobuf::types::ProtobufTypeBytes>(
"data",
@ -195,7 +195,7 @@ impl ::protobuf::Message for WebSocketRawMessage {
impl ::protobuf::Clear for WebSocketRawMessage {
fn clear(&mut self) {
self.module = WSModule::Doc;
self.channel = WSChannel::Document;
self.data.clear();
self.unknown_fields.clear();
}
@ -214,28 +214,28 @@ impl ::protobuf::reflect::ProtobufValue for WebSocketRawMessage {
}
#[derive(Clone,PartialEq,Eq,Debug,Hash)]
pub enum WSModule {
Doc = 0,
pub enum WSChannel {
Document = 0,
Folder = 1,
}
impl ::protobuf::ProtobufEnum for WSModule {
impl ::protobuf::ProtobufEnum for WSChannel {
fn value(&self) -> i32 {
*self as i32
}
fn from_i32(value: i32) -> ::std::option::Option<WSModule> {
fn from_i32(value: i32) -> ::std::option::Option<WSChannel> {
match value {
0 => ::std::option::Option::Some(WSModule::Doc),
1 => ::std::option::Option::Some(WSModule::Folder),
0 => ::std::option::Option::Some(WSChannel::Document),
1 => ::std::option::Option::Some(WSChannel::Folder),
_ => ::std::option::Option::None
}
}
fn values() -> &'static [Self] {
static values: &'static [WSModule] = &[
WSModule::Doc,
WSModule::Folder,
static values: &'static [WSChannel] = &[
WSChannel::Document,
WSChannel::Folder,
];
values
}
@ -243,43 +243,43 @@ impl ::protobuf::ProtobufEnum for WSModule {
fn enum_descriptor_static() -> &'static ::protobuf::reflect::EnumDescriptor {
static descriptor: ::protobuf::rt::LazyV2<::protobuf::reflect::EnumDescriptor> = ::protobuf::rt::LazyV2::INIT;
descriptor.get(|| {
::protobuf::reflect::EnumDescriptor::new_pb_name::<WSModule>("WSModule", file_descriptor_proto())
::protobuf::reflect::EnumDescriptor::new_pb_name::<WSChannel>("WSChannel", file_descriptor_proto())
})
}
}
impl ::std::marker::Copy for WSModule {
impl ::std::marker::Copy for WSChannel {
}
impl ::std::default::Default for WSModule {
impl ::std::default::Default for WSChannel {
fn default() -> Self {
WSModule::Doc
WSChannel::Document
}
}
impl ::protobuf::reflect::ProtobufValue for WSModule {
impl ::protobuf::reflect::ProtobufValue for WSChannel {
fn as_ref(&self) -> ::protobuf::reflect::ReflectValueRef {
::protobuf::reflect::ReflectValueRef::Enum(::protobuf::ProtobufEnum::descriptor(self))
}
}
static file_descriptor_proto_data: &'static [u8] = b"\
\n\tmsg.proto\"L\n\x13WebSocketRawMessage\x12!\n\x06module\x18\x01\x20\
\x01(\x0e2\t.WSModuleR\x06module\x12\x12\n\x04data\x18\x02\x20\x01(\x0cR\
\x04data*\x1f\n\x08WSModule\x12\x07\n\x03Doc\x10\0\x12\n\n\x06Folder\x10\
\x01J\x82\x02\n\x06\x12\x04\0\0\t\x01\n\x08\n\x01\x0c\x12\x03\0\0\x12\n\
\n\n\x02\x04\0\x12\x04\x02\0\x05\x01\n\n\n\x03\x04\0\x01\x12\x03\x02\x08\
\x1b\n\x0b\n\x04\x04\0\x02\0\x12\x03\x03\x04\x18\n\x0c\n\x05\x04\0\x02\0\
\x06\x12\x03\x03\x04\x0c\n\x0c\n\x05\x04\0\x02\0\x01\x12\x03\x03\r\x13\n\
\x0c\n\x05\x04\0\x02\0\x03\x12\x03\x03\x16\x17\n\x0b\n\x04\x04\0\x02\x01\
\x12\x03\x04\x04\x13\n\x0c\n\x05\x04\0\x02\x01\x05\x12\x03\x04\x04\t\n\
\x0c\n\x05\x04\0\x02\x01\x01\x12\x03\x04\n\x0e\n\x0c\n\x05\x04\0\x02\x01\
\x03\x12\x03\x04\x11\x12\n\n\n\x02\x05\0\x12\x04\x06\0\t\x01\n\n\n\x03\
\x05\0\x01\x12\x03\x06\x05\r\n\x0b\n\x04\x05\0\x02\0\x12\x03\x07\x04\x0c\
\n\x0c\n\x05\x05\0\x02\0\x01\x12\x03\x07\x04\x07\n\x0c\n\x05\x05\0\x02\0\
\x02\x12\x03\x07\n\x0b\n\x0b\n\x04\x05\0\x02\x01\x12\x03\x08\x04\x0f\n\
\x0c\n\x05\x05\0\x02\x01\x01\x12\x03\x08\x04\n\n\x0c\n\x05\x05\0\x02\x01\
\x02\x12\x03\x08\r\x0eb\x06proto3\
\n\tmsg.proto\"O\n\x13WebSocketRawMessage\x12$\n\x07channel\x18\x01\x20\
\x01(\x0e2\n.WSChannelR\x07channel\x12\x12\n\x04data\x18\x02\x20\x01(\
\x0cR\x04data*%\n\tWSChannel\x12\x0c\n\x08Document\x10\0\x12\n\n\x06Fold\
er\x10\x01J\x82\x02\n\x06\x12\x04\0\0\t\x01\n\x08\n\x01\x0c\x12\x03\0\0\
\x12\n\n\n\x02\x04\0\x12\x04\x02\0\x05\x01\n\n\n\x03\x04\0\x01\x12\x03\
\x02\x08\x1b\n\x0b\n\x04\x04\0\x02\0\x12\x03\x03\x04\x1a\n\x0c\n\x05\x04\
\0\x02\0\x06\x12\x03\x03\x04\r\n\x0c\n\x05\x04\0\x02\0\x01\x12\x03\x03\
\x0e\x15\n\x0c\n\x05\x04\0\x02\0\x03\x12\x03\x03\x18\x19\n\x0b\n\x04\x04\
\0\x02\x01\x12\x03\x04\x04\x13\n\x0c\n\x05\x04\0\x02\x01\x05\x12\x03\x04\
\x04\t\n\x0c\n\x05\x04\0\x02\x01\x01\x12\x03\x04\n\x0e\n\x0c\n\x05\x04\0\
\x02\x01\x03\x12\x03\x04\x11\x12\n\n\n\x02\x05\0\x12\x04\x06\0\t\x01\n\n\
\n\x03\x05\0\x01\x12\x03\x06\x05\x0e\n\x0b\n\x04\x05\0\x02\0\x12\x03\x07\
\x04\x11\n\x0c\n\x05\x05\0\x02\0\x01\x12\x03\x07\x04\x0c\n\x0c\n\x05\x05\
\0\x02\0\x02\x12\x03\x07\x0f\x10\n\x0b\n\x04\x05\0\x02\x01\x12\x03\x08\
\x04\x0f\n\x0c\n\x05\x05\0\x02\x01\x01\x12\x03\x08\x04\n\n\x0c\n\x05\x05\
\0\x02\x01\x02\x12\x03\x08\r\x0eb\x06proto3\
";
static file_descriptor_proto_lazy: ::protobuf::rt::LazyV2<::protobuf::descriptor::FileDescriptorProto> = ::protobuf::rt::LazyV2::INIT;

View File

@ -1,10 +1,10 @@
syntax = "proto3";
message WebSocketRawMessage {
WSModule module = 1;
WSChannel channel = 1;
bytes data = 2;
}
enum WSModule {
Doc = 0;
enum WSChannel {
Document = 0;
Folder = 1;
}

View File

@ -2,7 +2,7 @@
use crate::{
connect::{WSConnectionFuture, WSStream},
errors::WSError,
WSModule,
WSChannel,
WebSocketRawMessage,
};
use backend_service::errors::ServerError;
@ -30,10 +30,10 @@ use tokio_tungstenite::tungstenite::{
pub type MsgReceiver = UnboundedReceiver<Message>;
pub type MsgSender = UnboundedSender<Message>;
type Handlers = DashMap<WSModule, Arc<dyn WSMessageReceiver>>;
type Handlers = DashMap<WSChannel, Arc<dyn WSMessageReceiver>>;
pub trait WSMessageReceiver: Sync + Send + 'static {
fn source(&self) -> WSModule;
fn source(&self) -> WSChannel;
fn receive_message(&self, msg: WebSocketRawMessage);
}
@ -175,7 +175,7 @@ impl WSHandlerFuture {
fn handle_binary_message(&self, bytes: Vec<u8>) {
let bytes = Bytes::from(bytes);
match WebSocketRawMessage::try_from(bytes) {
Ok(message) => match self.handlers.get(&message.module) {
Ok(message) => match self.handlers.get(&message.channel) {
None => log::error!("Can't find any handler for message: {:?}", message),
Some(handler) => handler.receive_message(message.clone()),
},
@ -215,17 +215,17 @@ impl WSSender {
Ok(())
}
pub fn send_text(&self, source: &WSModule, text: &str) -> Result<(), WSError> {
pub fn send_text(&self, source: &WSChannel, text: &str) -> Result<(), WSError> {
let msg = WebSocketRawMessage {
module: source.clone(),
channel: source.clone(),
data: text.as_bytes().to_vec(),
};
self.send_msg(msg)
}
pub fn send_binary(&self, source: &WSModule, bytes: Vec<u8>) -> Result<(), WSError> {
pub fn send_binary(&self, source: &WSChannel, bytes: Vec<u8>) -> Result<(), WSError> {
let msg = WebSocketRawMessage {
module: source.clone(),
channel: source.clone(),
data: bytes,
};
self.send_msg(msg)