AppFlowy-IO/AppFlowy
1
0
Fork 0
mirror of https://github.com/AppFlowy-IO/AppFlowy.git synced 2024-09-20 23:47:20 +03:00
Code Issues Projects Releases Wiki Activity

fix init folder manager bugs

Browse Source
This commit is contained in:
appflowy 2022-01-23 22:33:47 +08:00
parent 23c4924532
commit 10d99bdd8b
36 changed files with 205 additions and 157 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
backend/src/services/document/persistence.rs
View File

@ -11,7 +11,7 @@ use crate::services::kv::revision_kv::RevisionKVPersistence;
use std::sync::Arc;
use uuid::Uuid;
#[tracing::instrument(level = "debug", skip(document_store, params), err)]
#[tracing::instrument(level = "trace", skip(document_store, params), err)]
pub(crate) async fn create_document(
document_store: &Arc<RevisionKVPersistence>,
mut params: CreateDocParams,
@ -21,7 +21,7 @@ pub(crate) async fn create_document(
Ok(())
}
#[tracing::instrument(level = "debug", skip(document_store), err)]
#[tracing::instrument(level = "trace", skip(document_store), err)]
pub async fn read_document(
document_store: &Arc<RevisionKVPersistence>,
params: DocumentId,
@ -52,7 +52,7 @@ pub async fn reset_document(
Ok(())
}
#[tracing::instrument(level = "debug", skip(document_store), err)]
#[tracing::instrument(level = "trace", skip(document_store), err)]
pub(crate) async fn delete_document(
document_store: &Arc<RevisionKVPersistence>,
doc_id: Uuid,

12
backend/src/services/document/ws_actor.rs
View File

@ -31,27 +31,27 @@ pub enum DocumentWSActorMessage {
}
pub struct DocumentWebSocketActor {
receiver: Option<mpsc::Receiver<DocumentWSActorMessage>>,
actor_msg_receiver: Option<mpsc::Receiver<DocumentWSActorMessage>>,
doc_manager: Arc<ServerDocumentManager>,
}
impl DocumentWebSocketActor {
pub fn new(receiver: mpsc::Receiver<DocumentWSActorMessage>, manager: Arc<ServerDocumentManager>) -> Self {
Self {
receiver: Some(receiver),
actor_msg_receiver: Some(receiver),
doc_manager: manager,
}
}
pub async fn run(mut self) {
let mut receiver = self
.receiver
let mut actor_msg_receiver = self
.actor_msg_receiver
.take()
.expect("DocumentWebSocketActor's receiver should only take one time");
let stream = stream! {
loop {
match receiver.recv().await {
match actor_msg_receiver.recv().await {
Some(msg) => yield msg,
None => break,
}
@ -79,7 +79,7 @@ impl DocumentWebSocketActor {
.await
.map_err(internal_error)??;
tracing::debug!(
tracing::trace!(
"[DocumentWebSocketActor]: receive: {}:{}, {:?}",
document_client_data.object_id,
document_client_data.data_id,

21
backend/src/services/document/ws_receiver.rs
View File

@ -34,28 +34,31 @@ pub fn make_document_ws_receiver(
persistence: Arc<FlowyPersistence>,
document_manager: Arc<ServerDocumentManager>,
) -> Arc<DocumentWebSocketReceiver> {
let (ws_sender, rx) = tokio::sync::mpsc::channel(1000);
let (actor_msg_sender, rx) = tokio::sync::mpsc::channel(1000);
let actor = DocumentWebSocketActor::new(rx, document_manager);
tokio::task::spawn(actor.run());
Arc::new(DocumentWebSocketReceiver::new(persistence, ws_sender))
Arc::new(DocumentWebSocketReceiver::new(persistence, actor_msg_sender))
}
pub struct DocumentWebSocketReceiver {
ws_sender: mpsc::Sender<DocumentWSActorMessage>,
actor_msg_sender: mpsc::Sender<DocumentWSActorMessage>,
persistence: Arc<FlowyPersistence>,
}
impl DocumentWebSocketReceiver {
pub fn new(persistence: Arc<FlowyPersistence>, ws_sender: mpsc::Sender<DocumentWSActorMessage>) -> Self {
Self { ws_sender, persistence }
pub fn new(persistence: Arc<FlowyPersistence>, actor_msg_sender: mpsc::Sender<DocumentWSActorMessage>) -> Self {
Self {
actor_msg_sender,
persistence,
}
}
}
impl WebSocketReceiver for DocumentWebSocketReceiver {
fn receive(&self, data: WSClientData) {
let (ret, rx) = oneshot::channel();
let sender = self.ws_sender.clone();
let actor_msg_sender = self.actor_msg_sender.clone();
let persistence = self.persistence.clone();
actix_rt::spawn(async move {
@ -65,13 +68,13 @@ impl WebSocketReceiver for DocumentWebSocketReceiver {
ret,
};
match sender.send(msg).await {
match actor_msg_sender.send(msg).await {
Ok(_) => {},
Err(e) => log::error!("{}", e),
Err(e) => log::error!("[DocumentWebSocketReceiver]: send message to actor failed: {}", e),
}
match rx.await {
Ok(_) => {},
Err(e) => log::error!("{:?}", e),
Err(e) => log::error!("[DocumentWebSocketReceiver]: message ret failed {:?}", e),
};
});
}

16
backend/src/services/folder/ws_actor.rs
View File

@ -29,28 +29,30 @@ pub enum FolderWSActorMessage {
}
pub struct FolderWebSocketActor {
receiver: Option<mpsc::Receiver<FolderWSActorMessage>>,
actor_msg_receiver: Option<mpsc::Receiver<FolderWSActorMessage>>,
folder_manager: Arc<ServerFolderManager>,
}
impl FolderWebSocketActor {
pub fn new(receiver: mpsc::Receiver<FolderWSActorMessage>, folder_manager: Arc<ServerFolderManager>) -> Self {
Self {
receiver: Some(receiver),
actor_msg_receiver: Some(receiver),
folder_manager,
}
}
pub async fn run(mut self) {
let mut receiver = self
.receiver
let mut actor_msg_receiver = self
.actor_msg_receiver
.take()
.expect("FolderWebSocketActor's receiver should only take one time");
let stream = stream! {
loop {
match receiver.recv().await {
match actor_msg_receiver.recv().await {
Some(msg) => yield msg,
None => break,
None => {
break
},
}
}
};
@ -76,7 +78,7 @@ impl FolderWebSocketActor {
.map_err(internal_error)??;
tracing::debug!(
"[DocumentWebSocketActor]: receive: {}:{}, {:?}",
"[FolderWebSocketActor]: receive: {}:{}, {:?}",
folder_client_data.object_id,
folder_client_data.data_id,
folder_client_data.ty

23
backend/src/services/folder/ws_receiver.rs
View File

@ -23,27 +23,30 @@ pub fn make_folder_ws_receiver(
persistence: Arc<FlowyPersistence>,
folder_manager: Arc<ServerFolderManager>,
) -> Arc<FolderWebSocketReceiver> {
let (ws_sender, rx) = tokio::sync::mpsc::channel(1000);
let (actor_msg_sender, rx) = tokio::sync::mpsc::channel(1000);
let actor = FolderWebSocketActor::new(rx, folder_manager);
tokio::task::spawn(actor.run());
Arc::new(FolderWebSocketReceiver::new(persistence, ws_sender))
Arc::new(FolderWebSocketReceiver::new(persistence, actor_msg_sender))
}
pub struct FolderWebSocketReceiver {
ws_sender: mpsc::Sender<FolderWSActorMessage>,
actor_msg_sender: mpsc::Sender<FolderWSActorMessage>,
persistence: Arc<FlowyPersistence>,
}
impl FolderWebSocketReceiver {
pub fn new(persistence: Arc<FlowyPersistence>, ws_sender: mpsc::Sender<FolderWSActorMessage>) -> Self {
Self { ws_sender, persistence }
pub fn new(persistence: Arc<FlowyPersistence>, actor_msg_sender: mpsc::Sender<FolderWSActorMessage>) -> Self {
Self {
actor_msg_sender,
persistence,
}
}
}
impl WebSocketReceiver for FolderWebSocketReceiver {
fn receive(&self, data: WSClientData) {
let (ret, rx) = oneshot::channel();
let sender = self.ws_sender.clone();
let actor_msg_sender = self.actor_msg_sender.clone();
let persistence = self.persistence.clone();
actix_rt::spawn(async move {
@ -53,13 +56,15 @@ impl WebSocketReceiver for FolderWebSocketReceiver {
ret,
};
match sender.send(msg).await {
match actor_msg_sender.send(msg).await {
Ok(_) => {},
Err(e) => log::error!("{}", e),
Err(e) => {
log::error!("[FolderWebSocketReceiver]: send message to actor failed: {}", e);
},
}
match rx.await {
Ok(_) => {},
Err(e) => log::error!("{:?}", e),
Err(e) => log::error!("[FolderWebSocketReceiver]: message ret failed {:?}", e),
};
});
}

2
backend/src/services/log.rs
View File

@ -28,7 +28,7 @@ impl Builder {
let env_filter = EnvFilter::new(self.env_filter);
let subscriber = tracing_subscriber::fmt()
.with_target(true)
.with_max_level(tracing::Level::DEBUG)
.with_max_level(tracing::Level::TRACE)
.with_writer(std::io::stderr)
.with_thread_ids(true)
.compact()

5
backend/src/services/web_socket/ws_client.rs
View File

@ -29,8 +29,9 @@ impl std::default::Default for WebSocketReceivers {
impl WebSocketReceivers {
pub fn new() -> Self { WebSocketReceivers::default() }
pub fn set(&mut self, source: WSChannel, receiver: Arc<dyn WebSocketReceiver>) {
self.inner.insert(source, receiver);
pub fn set(&mut self, channel: WSChannel, receiver: Arc<dyn WebSocketReceiver>) {
tracing::trace!("Add {:?} receiver", channel);
self.inner.insert(channel, receiver);
}
pub fn get(&self, source: &WSChannel) -> Option<Arc<dyn WebSocketReceiver>> { self.inner.get(source).cloned() }

2
frontend/Makefile.toml
View File

@ -15,7 +15,7 @@ CARGO_MAKE_EXTEND_WORKSPACE_MAKEFILE = true
CARGO_MAKE_CRATE_FS_NAME = "dart_ffi"
CARGO_MAKE_CRATE_NAME = "dart-ffi"
VERSION = "0.0.2"
FEATURES = "flutter"
FEATURES = "flutter, http_server"
PRODUCT_NAME = "AppFlowy"
#CRATE_TYPE: https://doc.rust-lang.org/reference/linkage.html
CRATE_TYPE = "staticlib"

2
frontend/app_flowy/lib/user/presentation/splash_screen.dart
View File

@ -2,6 +2,7 @@ import 'package:app_flowy/startup/startup.dart';
import 'package:app_flowy/user/application/splash_bloc.dart';
import 'package:app_flowy/user/domain/auth_state.dart';
import 'package:app_flowy/user/domain/i_splash.dart';
import 'package:flowy_log/flowy_log.dart';
import 'package:flowy_sdk/dispatch/dispatch.dart';
import 'package:flowy_sdk/protobuf/flowy-core-data-model/errors.pb.dart';
import 'package:flutter/material.dart';
@ -48,6 +49,7 @@ class SplashScreen extends StatelessWidget {
return result.fold(
(workspaceSetting) => getIt<ISplashRoute>().pushHomeScreen(context, userProfile, workspaceSetting),
(error) async {
Log.error(error);
assert(error.code == ErrorCode.RecordNotFound.value);
getIt<ISplashRoute>().pushWelcomeScreen(context, userProfile);
},

17
frontend/rust-lib/flowy-core/src/controller.rs
View File

@ -7,7 +7,7 @@ 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, fmt::Formatter, sync::Arc};
use tokio::sync::RwLock as TokioRwLock;
@ -28,10 +28,10 @@ use crate::{
};
lazy_static! {
static ref INIT_FOLDER_FLAG: RwLock<HashMap<String, bool>> = RwLock::new(HashMap::new());
static ref INIT_FOLDER_FLAG: TokioRwLock<HashMap<String, bool>> = TokioRwLock::new(HashMap::new());
}
const FOLDER_ID: &str = "user_folder";
const FOLDER_ID: &str = "folder";
const FOLDER_ID_SPLIT: &str = ":";
#[derive(Clone)]
pub struct FolderId(String);
@ -64,17 +64,13 @@ pub struct FolderManager {
}
impl FolderManager {
pub fn new(
pub async fn new(
user: Arc<dyn WorkspaceUser>,
cloud_service: Arc<dyn FolderCouldServiceV1>,
database: Arc<dyn WorkspaceDatabase>,
document_manager: Arc<FlowyDocumentManager>,
web_socket: Arc<dyn RevisionWebSocket>,
) -> Self {
if let Ok(token) = user.token() {
INIT_FOLDER_FLAG.write().insert(token, false);
}
let folder_editor = Arc::new(TokioRwLock::new(None));
let persistence = Arc::new(FolderPersistence::new(database.clone(), folder_editor.clone()));
@ -145,7 +141,8 @@ impl FolderManager {
}
pub async fn initialize(&self, user_id: &str) -> FlowyResult<()> {
if let Some(is_init) = INIT_FOLDER_FLAG.read().get(user_id) {
let mut write_guard = INIT_FOLDER_FLAG.write().await;
if let Some(is_init) = write_guard.get(user_id) {
if *is_init {
return Ok(());
}
@ -160,7 +157,7 @@ impl FolderManager {
let _ = self.app_controller.initialize()?;
let _ = self.view_controller.initialize()?;
INIT_FOLDER_FLAG.write().insert(user_id.to_owned(), true);
write_guard.insert(user_id.to_owned(), true);
Ok(())
}

4
frontend/rust-lib/flowy-core/src/dart_notification.rs
View File

@ -27,12 +27,12 @@ impl std::convert::From<WorkspaceNotification> for i32 {
fn from(notification: WorkspaceNotification) -> Self { notification as i32 }
}
#[tracing::instrument(level = "debug")]
#[tracing::instrument(level = "trace")]
pub(crate) fn send_dart_notification(id: &str, ty: WorkspaceNotification) -> DartNotifyBuilder {
DartNotifyBuilder::new(id, ty, OBSERVABLE_CATEGORY)
}
#[tracing::instrument(level = "debug")]
#[tracing::instrument(level = "trace")]
pub(crate) fn send_anonymous_dart_notification(ty: WorkspaceNotification) -> DartNotifyBuilder {
DartNotifyBuilder::new("", ty, OBSERVABLE_CATEGORY)
}

6
frontend/rust-lib/flowy-core/src/services/app/controller.rs
View File

@ -112,14 +112,14 @@ impl AppController {
}
impl AppController {
#[tracing::instrument(level = "debug", skip(self), err)]
#[tracing::instrument(level = "trace", skip(self), err)]
async fn create_app_on_server(&self, params: CreateAppParams) -> Result<App, FlowyError> {
let token = self.user.token()?;
let app = self.cloud_service.create_app(&token, params).await?;
Ok(app)
}
#[tracing::instrument(level = "debug", skip(self), err)]
#[tracing::instrument(level = "trace", skip(self), err)]
fn update_app_on_server(&self, params: UpdateAppParams) -> Result<(), FlowyError> {
let token = self.user.token()?;
let server = self.cloud_service.clone();
@ -135,7 +135,7 @@ impl AppController {
Ok(())
}
#[tracing::instrument(level = "debug", skip(self), err)]
#[tracing::instrument(level = "trace", skip(self), err)]
fn read_app_on_server(&self, params: AppId) -> Result<(), FlowyError> {
let token = self.user.token()?;
let server = self.cloud_service.clone();

2
frontend/rust-lib/flowy-core/src/services/folder_editor.rs
View File

@ -104,7 +104,7 @@ struct FolderRevisionCloudServiceImpl {
}
impl RevisionCloudService for FolderRevisionCloudServiceImpl {
#[tracing::instrument(level = "debug", skip(self))]
#[tracing::instrument(level = "trace", skip(self))]
fn fetch_object(&self, _user_id: &str, _object_id: &str) -> FutureResult<Vec<Revision>, FlowyError> {
FutureResult::new(async move { Ok(vec![]) })
}

2
frontend/rust-lib/flowy-core/src/services/mod.rs
View File

@ -11,4 +11,4 @@ pub(crate) mod view;
mod web_socket;
pub(crate) mod workspace;
pub const FOLDER_SYNC_INTERVAL_IN_MILLIS: u64 = 1000;
pub const FOLDER_SYNC_INTERVAL_IN_MILLIS: u64 = 5000;

8
frontend/rust-lib/flowy-core/src/services/trash/controller.rs
View File

@ -205,7 +205,7 @@ impl TrashController {
}
impl TrashController {
#[tracing::instrument(level = "debug", skip(self, trash), err)]
#[tracing::instrument(level = "trace", skip(self, trash), err)]
fn create_trash_on_server<T: Into<RepeatedTrashId>>(&self, trash: T) -> FlowyResult<()> {
let token = self.user.token()?;
let trash_identifiers = trash.into();
@ -220,7 +220,7 @@ impl TrashController {
Ok(())
}
#[tracing::instrument(level = "debug", skip(self, trash), err)]
#[tracing::instrument(level = "trace", skip(self, trash), err)]
fn delete_trash_on_server<T: Into<RepeatedTrashId>>(&self, trash: T) -> FlowyResult<()> {
let token = self.user.token()?;
let trash_identifiers = trash.into();
@ -234,7 +234,7 @@ impl TrashController {
Ok(())
}
#[tracing::instrument(level = "debug", skip(self), err)]
#[tracing::instrument(level = "trace", skip(self), err)]
fn read_trash_on_server(&self) -> FlowyResult<()> {
let token = self.user.token()?;
let server = self.cloud_service.clone();
@ -264,7 +264,7 @@ impl TrashController {
Ok(())
}
#[tracing::instrument(level = "debug", skip(self), err)]
#[tracing::instrument(level = "trace", skip(self), err)]
async fn delete_all_trash_on_server(&self) -> FlowyResult<()> {
let token = self.user.token()?;
let server = self.cloud_service.clone();

2
frontend/rust-lib/flowy-core/src/services/view/controller.rs
View File

@ -60,7 +60,7 @@ impl ViewController {
Ok(())
}
#[tracing::instrument(level = "debug", skip(self, params), fields(name = %params.name), err)]
#[tracing::instrument(level = "trace", skip(self, params), fields(name = %params.name), err)]
pub(crate) async fn create_view_from_params(&self, params: CreateViewParams) -> Result<View, FlowyError> {
let view_data = if params.view_data.is_empty() {
initial_delta_string()

1
frontend/rust-lib/flowy-core/src/services/web_socket.rs
View File

@ -37,6 +37,7 @@ pub(crate) async fn make_folder_ws_manager(
let sink_provider = Arc::new(FolderWSSinkDataProviderAdapter(composite_sink_provider));
let ping_duration = Duration::from_millis(FOLDER_SYNC_INTERVAL_IN_MILLIS);
Arc::new(RevisionWebSocketManager::new(
"Folder",
folder_id,
web_socket,
sink_provider,

6
frontend/rust-lib/flowy-core/src/services/workspace/controller.rs
View File

@ -150,14 +150,14 @@ impl WorkspaceController {
}
impl WorkspaceController {
#[tracing::instrument(level = "debug", skip(self), err)]
#[tracing::instrument(level = "trace", skip(self), err)]
async fn create_workspace_on_server(&self, params: CreateWorkspaceParams) -> Result<Workspace, FlowyError> {
let token = self.user.token()?;
let workspace = self.cloud_service.create_workspace(&token, params).await?;
Ok(workspace)
}
#[tracing::instrument(level = "debug", skip(self), err)]
#[tracing::instrument(level = "trace", skip(self), err)]
fn update_workspace_on_server(&self, params: UpdateWorkspaceParams) -> Result<(), FlowyError> {
let (token, server) = (self.user.token()?, self.cloud_service.clone());
tokio::spawn(async move {
@ -172,7 +172,7 @@ impl WorkspaceController {
Ok(())
}
#[tracing::instrument(level = "debug", skip(self), err)]
#[tracing::instrument(level = "trace", skip(self), err)]
fn delete_workspace_on_server(&self, workspace_id: &str) -> Result<(), FlowyError> {
let params = WorkspaceId {
workspace_id: Some(workspace_id.to_string()),

2
frontend/rust-lib/flowy-core/src/services/workspace/event_handler.rs
View File

@ -94,7 +94,7 @@ pub async fn read_cur_workspace_handler(
data_result(setting)
}
#[tracing::instrument(level = "debug", skip(folder_manager), err)]
#[tracing::instrument(level = "trace", skip(folder_manager), err)]
fn read_workspaces_on_server(
folder_manager: Unit<Arc<FolderManager>>,
user_id: String,

7
frontend/rust-lib/flowy-document/src/controller.rs
View File

@ -21,7 +21,6 @@ pub trait DocumentUser: Send + Sync {
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>;
@ -67,7 +66,7 @@ impl FlowyDocumentManager {
self.get_editor(doc_id).await
}
#[tracing::instrument(level = "debug", skip(self, doc_id), fields(doc_id), err)]
#[tracing::instrument(level = "trace", skip(self, doc_id), fields(doc_id), err)]
pub fn close_document<T: AsRef<str>>(&self, doc_id: T) -> Result<(), FlowyError> {
let doc_id = doc_id.as_ref();
tracing::Span::current().record("doc_id", &doc_id);
@ -178,7 +177,7 @@ struct DocumentRevisionCloudServiceImpl {
}
impl RevisionCloudService for DocumentRevisionCloudServiceImpl {
#[tracing::instrument(level = "debug", skip(self))]
#[tracing::instrument(level = "trace", skip(self))]
fn fetch_object(&self, user_id: &str, doc_id: &str) -> FutureResult<Vec<Revision>, FlowyError> {
let params = DocumentId {
doc_id: doc_id.to_string(),
@ -235,7 +234,7 @@ impl OpenDocCache {
}
}
#[tracing::instrument(level = "debug", skip(state_receiver, receivers))]
#[tracing::instrument(level = "trace", skip(state_receiver, receivers))]
fn listen_ws_state_changed(mut state_receiver: WSStateReceiver, receivers: WebSocketDataReceivers) {
tokio::spawn(async move {
while let Ok(state) = state_receiver.recv().await {

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

@ -147,7 +147,7 @@ impl ClientDocumentEditor {
Ok(json)
}
#[tracing::instrument(level = "debug", skip(self, data), err)]
#[tracing::instrument(level = "trace", skip(self, data), err)]
pub(crate) async fn compose_local_delta(&self, data: Bytes) -> Result<(), FlowyError> {
let delta = RichTextDelta::from_bytes(&data)?;
let (ret, rx) = oneshot::channel::<CollaborateResult<()>>();

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

@ -48,6 +48,7 @@ pub(crate) async fn make_document_ws_manager(
let sink_provider = Arc::new(DocumentWSSinkDataProviderAdapter(composite_sink_provider));
let ping_duration = Duration::from_millis(DOCUMENT_SYNC_INTERVAL_IN_MILLIS);
let ws_manager = Arc::new(RevisionWebSocketManager::new(
"Document",
&doc_id,
web_socket,
sink_provider,

7
frontend/rust-lib/flowy-net/src/local_server/ws.rs
View File

@ -60,15 +60,14 @@ 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> {
fn add_msg_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(())
}
fn sender(&self) -> Result<Arc<dyn FlowyWebSocket>, FlowyError> {
let ws = LocalWebSocketAdaptor(self.server_ws_sender.clone());
Ok(Arc::new(ws))
fn ws_msg_sender(&self) -> Result<Arc<dyn FlowyWebSocket>, FlowyError> {
Ok(Arc::new(LocalWebSocketAdaptor(self.server_ws_sender.clone())))
}
}

8
frontend/rust-lib/flowy-net/src/ws/connection.rs
View File

@ -15,8 +15,8 @@ pub trait FlowyRawWebSocket: Send + Sync {
fn stop_connect(&self) -> FutureResult<(), FlowyError>;
fn subscribe_connect_state(&self) -> broadcast::Receiver<WSConnectState>;
fn reconnect(&self, count: usize) -> FutureResult<(), FlowyError>;
fn add_receiver(&self, receiver: Arc<dyn WSMessageReceiver>) -> Result<(), FlowyError>;
fn sender(&self) -> Result<Arc<dyn FlowyWebSocket>, FlowyError>;
fn add_msg_receiver(&self, receiver: Arc<dyn WSMessageReceiver>) -> Result<(), FlowyError>;
fn ws_msg_sender(&self) -> Result<Arc<dyn FlowyWebSocket>, FlowyError>;
}
pub trait FlowyWebSocket: Send + Sync {
@ -97,11 +97,11 @@ impl FlowyWebSocketConnect {
pub fn subscribe_network_ty(&self) -> broadcast::Receiver<NetworkType> { self.status_notifier.subscribe() }
pub fn add_ws_message_receiver(&self, receiver: Arc<dyn WSMessageReceiver>) -> Result<(), FlowyError> {
let _ = self.inner.add_receiver(receiver)?;
let _ = self.inner.add_msg_receiver(receiver)?;
Ok(())
}
pub fn web_socket(&self) -> Result<Arc<dyn FlowyWebSocket>, FlowyError> { self.inner.sender() }
pub fn web_socket(&self) -> Result<Arc<dyn FlowyWebSocket>, FlowyError> { self.inner.ws_msg_sender() }
}
#[tracing::instrument(level = "debug", skip(ws_conn))]

4
frontend/rust-lib/flowy-net/src/ws/http_ws.rs
View File

@ -37,12 +37,12 @@ impl FlowyRawWebSocket for Arc<WSController> {
})
}
fn add_receiver(&self, receiver: Arc<dyn WSMessageReceiver>) -> Result<(), FlowyError> {
fn add_msg_receiver(&self, receiver: Arc<dyn WSMessageReceiver>) -> Result<(), FlowyError> {
let _ = self.add_ws_message_receiver(receiver).map_err(internal_error)?;
Ok(())
}
fn sender(&self) -> Result<Arc<dyn FlowyWebSocket>, FlowyError> {
fn ws_msg_sender(&self) -> Result<Arc<dyn FlowyWebSocket>, FlowyError> {
let sender = self.ws_message_sender().map_err(internal_error)?;
Ok(sender)
}

11
frontend/rust-lib/flowy-sdk/src/deps_resolve/folder_deps.rs
View File

@ -20,7 +20,7 @@ use std::{convert::TryInto, sync::Arc};
pub struct FolderDepsResolver();
impl FolderDepsResolver {
pub fn resolve(
pub async fn resolve(
local_server: Option<Arc<LocalServer>>,
user_session: Arc<UserSession>,
server_config: &ClientServerConfiguration,
@ -35,13 +35,8 @@ impl FolderDepsResolver {
Some(local_server) => local_server,
};
let folder_manager = Arc::new(FolderManager::new(
user,
cloud_service,
database,
document_manager.clone(),
web_socket,
));
let folder_manager =
Arc::new(FolderManager::new(user, cloud_service, database, document_manager.clone(), web_socket).await);
let receiver = Arc::new(FolderWSMessageReceiverImpl(folder_manager.clone()));
ws_conn.add_ws_message_receiver(receiver).unwrap();

30
frontend/rust-lib/flowy-sdk/src/lib.rs
View File

@ -12,6 +12,7 @@ use flowy_user::services::{notifier::UserStatus, UserSession, UserSessionConfig}
use lib_dispatch::prelude::*;
use flowy_document::FlowyDocumentManager;
use lib_dispatch::util::tokio_default_runtime;
use module::mk_modules;
pub use module::*;
use std::{
@ -21,7 +22,7 @@ use std::{
Arc,
},
};
use tokio::sync::broadcast;
use tokio::{runtime::Runtime, sync::broadcast};
static INIT_LOG: AtomicBool = AtomicBool::new(false);
@ -95,7 +96,7 @@ impl FlowySDK {
init_log(&config);
init_kv(&config.root);
tracing::debug!("🔥 {:?}", config);
let runtime = tokio_default_runtime().unwrap();
let ws_addr = config.server_config.ws_addr();
let (local_server, ws_conn) = if cfg!(feature = "http_server") {
let ws_conn = Arc::new(FlowyWebSocketConnect::new(ws_addr));
@ -110,6 +111,7 @@ impl FlowySDK {
let user_session = mk_user_session(&config, &local_server, &config.server_config);
let document_manager = mk_document(&local_server, &ws_conn, &user_session, &config.server_config);
let folder_manager = mk_folder_manager(
&runtime,
&local_server,
&user_session,
&document_manager,
@ -117,9 +119,9 @@ impl FlowySDK {
&ws_conn,
);
//
let modules = mk_modules(&ws_conn, &folder_manager, &user_session);
let dispatcher = Arc::new(EventDispatcher::construct(|| modules));
let dispatcher = Arc::new(EventDispatcher::construct(runtime, || {
mk_modules(&ws_conn, &folder_manager, &user_session)
}));
_init(&local_server, &dispatcher, &ws_conn, &user_session, &folder_manager);
Self {
@ -244,19 +246,23 @@ fn mk_user_session(
}
fn mk_folder_manager(
runtime: &Runtime,
local_server: &Option<Arc<LocalServer>>,
user_session: &Arc<UserSession>,
document_manager: &Arc<FlowyDocumentManager>,
server_config: &ClientServerConfiguration,
ws_conn: &Arc<FlowyWebSocketConnect>,
) -> Arc<FolderManager> {
FolderDepsResolver::resolve(
local_server.clone(),
user_session.clone(),
server_config,
document_manager,
ws_conn.clone(),
)
runtime.block_on(async {
FolderDepsResolver::resolve(
local_server.clone(),
user_session.clone(),
server_config,
document_manager,
ws_conn.clone(),
)
.await
})
}
pub fn mk_document(

6
frontend/rust-lib/flowy-sdk/src/module.rs
View File

@ -10,13 +10,13 @@ pub fn mk_modules(
user_session: &Arc<UserSession>,
) -> Vec<Module> {
let user_module = mk_user_module(user_session.clone());
let core_module = mk_core_module(folder_manager.clone());
let folder_module = mk_folder_module(folder_manager.clone());
let network_module = mk_network_module(ws_conn.clone());
vec![user_module, core_module, network_module]
vec![user_module, folder_module, network_module]
}
fn mk_user_module(user_session: Arc<UserSession>) -> Module { flowy_user::module::create(user_session) }
fn mk_core_module(core: Arc<FolderManager>) -> Module { flowy_core::module::create(core) }
fn mk_folder_module(core: Arc<FolderManager>) -> Module { flowy_core::module::create(core) }
fn mk_network_module(ws_conn: Arc<FlowyWebSocketConnect>) -> Module { flowy_net::module::create(ws_conn) }

42
frontend/rust-lib/flowy-sync/src/rev_manager.rs
View File

@ -5,10 +5,8 @@ use flowy_collaboration::{
util::{pair_rev_id_from_revisions, RevIdCounter},
};
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::{broadcast, RwLock};
pub trait RevisionCloudService: Send + Sync {
@ -54,7 +52,7 @@ impl RevisionManager {
where
Builder: RevisionObjectBuilder,
{
let revisions = RevisionLoader {
let (revisions, rev_id) = RevisionLoader {
object_id: self.object_id.clone(),
user_id: self.user_id.clone(),
cloud,
@ -63,6 +61,7 @@ impl RevisionManager {
}
.load()
.await?;
self.rev_id_counter.set(rev_id);
Builder::build_with_revisions(&self.object_id, revisions)
}
@ -117,8 +116,6 @@ impl RevisionManager {
pub fn rev_id(&self) -> i64 { self.rev_id_counter.value() }
pub fn set_rev_id(&self, rev_id: i64) { self.rev_id_counter.set(rev_id); }
pub fn next_rev_id_pair(&self) -> (i64, i64) {
let cur = self.rev_id_counter.value();
let next = self.rev_id_counter.next();
@ -223,12 +220,14 @@ struct RevisionLoader {
}
impl RevisionLoader {
async fn load(&self) -> Result<Vec<Revision>, FlowyError> {
async fn load(&self) -> Result<(Vec<Revision>, i64), FlowyError> {
let records = self.revision_cache.batch_get(&self.object_id)?;
let revisions: Vec<Revision>;
let mut rev_id = 0;
if records.is_empty() {
let remote_revisions = self.cloud.fetch_object(&self.user_id, &self.object_id).await?;
for revision in &remote_revisions {
rev_id = revision.rev_id;
let _ = self
.revision_cache
.add(revision.clone(), RevisionState::Ack, true)
@ -236,25 +235,30 @@ impl RevisionLoader {
}
revisions = remote_revisions;
} else {
stream::iter(records.clone())
.filter(|record| future::ready(record.state == RevisionState::Sync))
.for_each(|record| async move {
let f = || async {
// Sync the records if their state is RevisionState::Local.
for record in records.clone() {
let f = || async {
rev_id = record.revision.rev_id;
if record.state == RevisionState::Sync {
// Sync the records if their state is RevisionState::Sync.
let _ = self.revision_sync_seq.add_revision_record(record.clone()).await?;
let _ = self.revision_cache.add(record.revision, record.state, false).await?;
Ok::<(), FlowyError>(())
};
match f().await {
Ok(_) => {},
Err(e) => tracing::error!("[RevisionLoader]: {}", e),
}
})
.await;
Ok::<(), FlowyError>(())
};
match f().await {
Ok(_) => {},
Err(e) => tracing::error!("[RevisionLoader]: {}", e),
}
}
revisions = records.into_iter().map(|record| record.revision).collect::<_>();
}
Ok(revisions)
if let Some(revision) = revisions.last() {
debug_assert_eq!(rev_id, revision.rev_id);
}
Ok((revisions, rev_id))
}
}

64
frontend/rust-lib/flowy-sync/src/ws_manager.rs
View File

@ -9,7 +9,7 @@ use flowy_error::{FlowyError, FlowyResult};
use futures_util::stream::StreamExt;
use lib_infra::future::{BoxResultFuture, FutureResult};
use lib_ws::WSConnectState;
use std::{collections::VecDeque, convert::TryFrom, sync::Arc};
use std::{collections::VecDeque, convert::TryFrom, fmt::Formatter, sync::Arc};
use tokio::{
sync::{
broadcast,
@ -41,6 +41,7 @@ pub trait RevisionWebSocket: Send + Sync + 'static {
}
pub struct RevisionWebSocketManager {
pub object_name: String,
pub object_id: String,
sink_provider: Arc<dyn RevisionWSSinkDataProvider>,
stream_consumer: Arc<dyn RevisionWSSteamConsumer>,
@ -51,8 +52,14 @@ pub struct RevisionWebSocketManager {
stop_sync_tx: SinkStopTx,
}
impl std::fmt::Display for RevisionWebSocketManager {
fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
f.write_fmt(format_args!("{}RevisionWebSocketManager", self.object_name))
}
}
impl RevisionWebSocketManager {
pub fn new(
object_name: &str,
object_id: &str,
web_socket: Arc<dyn RevisionWebSocket>,
sink_provider: Arc<dyn RevisionWSSinkDataProvider>,
@ -62,9 +69,11 @@ impl RevisionWebSocketManager {
let (ws_passthrough_tx, ws_passthrough_rx) = mpsc::channel(1000);
let (stop_sync_tx, _) = tokio::sync::broadcast::channel(2);
let object_id = object_id.to_string();
let object_name = object_name.to_string();
let (state_passthrough_tx, _) = broadcast::channel(2);
let mut manager = RevisionWebSocketManager {
object_id,
object_name,
sink_provider,
stream_consumer,
web_socket,
@ -81,12 +90,14 @@ impl RevisionWebSocketManager {
let ws_msg_rx = self.ws_passthrough_rx.take().expect("Only take once");
let sink = RevisionWSSink::new(
&self.object_id,
&self.object_name,
self.sink_provider.clone(),
self.web_socket.clone(),
self.stop_sync_tx.subscribe(),
ping_duration,
);
let stream = RevisionWSStream::new(
&self.object_name,
&self.object_id,
self.stream_consumer.clone(),
ws_msg_rx,
@ -106,28 +117,37 @@ impl RevisionWebSocketManager {
}
impl std::ops::Drop for RevisionWebSocketManager {
fn drop(&mut self) { tracing::trace!("{} RevisionWebSocketManager was dropped", self.object_id) }
fn drop(&mut self) { tracing::trace!("{} was dropped", self) }
}
pub struct RevisionWSStream {
object_name: String,
object_id: String,
consumer: Arc<dyn RevisionWSSteamConsumer>,
ws_msg_rx: Option<mpsc::Receiver<ServerRevisionWSData>>,
stop_rx: Option<SinkStopRx>,
}
impl std::fmt::Display for RevisionWSStream {
fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
f.write_fmt(format_args!("{}RevisionWSStream", self.object_name))
}
}
impl std::ops::Drop for RevisionWSStream {
fn drop(&mut self) { tracing::trace!("{} RevisionWSStream was dropped", self.object_id) }
fn drop(&mut self) { tracing::trace!("{} was dropped", self) }
}
impl RevisionWSStream {
pub fn new(
object_name: &str,
object_id: &str,
consumer: Arc<dyn RevisionWSSteamConsumer>,
ws_msg_rx: mpsc::Receiver<ServerRevisionWSData>,
stop_rx: SinkStopRx,
) -> Self {
RevisionWSStream {
object_name: object_name.to_string(),
object_id: object_id.to_owned(),
consumer,
ws_msg_rx: Some(ws_msg_rx),
@ -139,6 +159,7 @@ impl RevisionWSStream {
let mut receiver = self.ws_msg_rx.take().expect("Only take once");
let mut stop_rx = self.stop_rx.take().expect("Only take once");
let object_id = self.object_id.clone();
let name = format!("{}", &self);
let stream = stream! {
loop {
tokio::select! {
@ -148,13 +169,13 @@ impl RevisionWSStream {
yield msg
},
None => {
tracing::debug!("[RevisionWSStream]:{} loop exit", object_id);
tracing::debug!("[{}]:{} loop exit", name, object_id);
break;
},
}
},
_ = stop_rx.recv() => {
tracing::debug!("[RevisionWSStream]:{} loop exit", object_id);
tracing::debug!("[{}]:{} loop exit", name, object_id);
break
},
};
@ -165,7 +186,7 @@ impl RevisionWSStream {
.for_each(|msg| async {
match self.handle_message(msg).await {
Ok(_) => {},
Err(e) => tracing::error!("[RevisionWSStream]:{} error: {}", self.object_id, e),
Err(e) => tracing::error!("[{}]:{} error: {}", &self, self.object_id, e),
}
})
.await;
@ -174,7 +195,7 @@ impl RevisionWSStream {
async fn handle_message(&self, msg: ServerRevisionWSData) -> FlowyResult<()> {
let ServerRevisionWSData { object_id, ty, data } = msg;
let bytes = Bytes::from(data);
tracing::trace!("[RevisionWSStream]: new message: {}:{:?}", object_id, ty);
tracing::trace!("[{}]: new message: {}:{:?}", self, object_id, ty);
match ty {
ServerRevisionWSDataType::ServerPushRev => {
let _ = self.consumer.receive_push_revision(bytes).await?;
@ -199,26 +220,29 @@ impl RevisionWSStream {
type SinkStopRx = broadcast::Receiver<()>;
type SinkStopTx = broadcast::Sender<()>;
pub struct RevisionWSSink {
object_id: String,
object_name: String,
provider: Arc<dyn RevisionWSSinkDataProvider>,
ws_sender: Arc<dyn RevisionWebSocket>,
stop_rx: Option<SinkStopRx>,
object_id: String,
ping_duration: Duration,
}
impl RevisionWSSink {
pub fn new(
object_id: &str,
object_name: &str,
provider: Arc<dyn RevisionWSSinkDataProvider>,
ws_sender: Arc<dyn RevisionWebSocket>,
stop_rx: SinkStopRx,
ping_duration: Duration,
) -> Self {
Self {
object_id: object_id.to_owned(),
object_name: object_name.to_owned(),
provider,
ws_sender,
stop_rx: Some(stop_rx),
object_id: object_id.to_owned(),
ping_duration,
}
}
@ -228,6 +252,7 @@ impl RevisionWSSink {
let mut stop_rx = self.stop_rx.take().expect("Only take once");
let object_id = self.object_id.clone();
tokio::spawn(tick(tx, self.ping_duration));
let name = format!("{}", self);
let stream = stream! {
loop {
tokio::select! {
@ -238,7 +263,7 @@ impl RevisionWSSink {
}
},
_ = stop_rx.recv() => {
tracing::trace!("[RevisionWSSink:{}] loop exit", object_id);
tracing::trace!("[{}]:{} loop exit", name, object_id);
break
},
};
@ -248,7 +273,7 @@ impl RevisionWSSink {
.for_each(|_| async {
match self.send_next_revision().await {
Ok(_) => {},
Err(e) => tracing::error!("[RevisionWSSink] send failed, {:?}", e),
Err(e) => tracing::error!("[{}] send failed, {:?}", self, e),
}
})
.await;
@ -257,19 +282,25 @@ impl RevisionWSSink {
async fn send_next_revision(&self) -> FlowyResult<()> {
match self.provider.next().await? {
None => {
tracing::trace!("Finish synchronizing revisions");
tracing::trace!("[{}]: Finish synchronizing revisions", self);
Ok(())
},
Some(data) => {
tracing::trace!("[RevisionWSSink] send: {}:{}-{:?}", data.object_id, data.id(), data.ty);
tracing::trace!("[{}]: send {}:{}-{:?}", self, data.object_id, data.id(), data.ty);
self.ws_sender.send(data)
},
}
}
}
impl std::fmt::Display for RevisionWSSink {
fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
f.write_fmt(format_args!("{}RevisionWSSink", self.object_name))
}
}
impl std::ops::Drop for RevisionWSSink {
fn drop(&mut self) { tracing::trace!("{} RevisionWSSink was dropped", self.object_id) }
fn drop(&mut self) { tracing::trace!("{} was dropped", self) }
}
async fn tick(sender: mpsc::Sender<()>, duration: Duration) {
@ -330,11 +361,6 @@ impl CompositeWSSinkDataProvider {
}
},
};
if let Ok(Some(data)) = &data {
tracing::trace!("[CompositeWSSinkDataProvider]: {}:{:?}", data.object_id, data.ty);
}
data
}

4
frontend/rust-lib/lib-dispatch/src/dispatcher.rs
View File

@ -3,7 +3,6 @@ use crate::{
module::{as_module_map, Module, ModuleMap, ModuleRequest},
response::EventResponse,
service::{Service, ServiceFactory},
util::tokio_default_runtime,
};
use derivative::*;
use futures_core::future::BoxFuture;
@ -17,11 +16,10 @@ pub struct EventDispatcher {
}
impl EventDispatcher {
pub fn construct<F>(module_factory: F) -> EventDispatcher
pub fn construct<F>(runtime: tokio::runtime::Runtime, module_factory: F) -> EventDispatcher
where
F: FnOnce() -> Vec<Module>,
{
let runtime = tokio_default_runtime().unwrap();
let modules = module_factory();
tracing::trace!("{}", module_info(&modules));
let module_map = as_module_map(modules);

2
frontend/rust-lib/lib-dispatch/src/lib.rs
View File

@ -3,7 +3,7 @@ mod module;
mod request;
mod response;
mod service;
mod util;
pub mod util;
mod byte_trait;
mod data;

7
frontend/rust-lib/lib-dispatch/tests/api/module.rs
View File

@ -1,4 +1,4 @@
use lib_dispatch::prelude::*;
use lib_dispatch::{prelude::*, util::tokio_default_runtime};
use std::sync::Arc;
pub async fn hello() -> String { "say hello".to_string() }
@ -8,7 +8,10 @@ async fn test() {
env_logger::init();
let event = "1";
let dispatch = Arc::new(EventDispatcher::construct(|| vec![Module::new().event(event, hello)]));
let runtime = tokio_default_runtime().unwrap();
let dispatch = Arc::new(EventDispatcher::construct(runtime, || {
vec![Module::new().event(event, hello)]
}));
let request = ModuleRequest::new(event);
let _ = EventDispatcher::async_send_with_callback(dispatch.clone(), request, |resp| {
Box::pin(async move {

1
shared-lib/flowy-collaboration/src/server_document/document_manager.rs
View File

@ -88,6 +88,7 @@ impl ServerDocumentManager {
let result = match self.get_document_handler(&object_id).await {
None => {
tracing::trace!("Can't find the document. Creating the document {}", object_id);
let _ = self.create_document(&object_id, repeated_revision).await.map_err(|e| {
CollaborateError::internal().context(format!("Server create document failed: {}", e))
})?;

17
shared-lib/lib-ws/src/connect.rs
View File

@ -61,15 +61,19 @@ impl Future for WSConnectionFuture {
loop {
return match ready!(self.as_mut().project().fut.poll(cx)) {
Ok((stream, _)) => {
tracing::debug!("🐴 ws connect success");
tracing::debug!("[WebSocket]: connect success");
let (msg_tx, ws_rx) = (
self.msg_tx.take().expect("WsConnection should be call once "),
self.ws_rx.take().expect("WsConnection should be call once "),
self.msg_tx
.take()
.expect("[WebSocket]: WSConnection should be call once "),
self.ws_rx
.take()
.expect("[WebSocket]: WSConnection should be call once "),
);
Poll::Ready(Ok(WSStream::new(msg_tx, ws_rx, stream)))
},
Err(error) => {
tracing::debug!("🐴 ws connect failed: {:?}", error);
tracing::debug!("[WebSocket]: ❌ connect failed: {:?}", error);
Poll::Ready(Err(error.into()))
},
};
@ -99,7 +103,7 @@ impl WSStream {
.for_each(|message| async {
match tx.send(send_message(msg_tx.clone(), message)) {
Ok(_) => {},
Err(e) => log::error!("WsStream tx closed unexpectedly: {} ", e),
Err(e) => log::error!("[WebSocket]: WSStream sender closed unexpectedly: {} ", e),
}
})
.await;
@ -110,7 +114,8 @@ impl WSStream {
loop {
match rx.recv().await {
None => {
return Err(WSError::internal().context("WsStream rx closed unexpectedly"));
return Err(WSError::internal()
.context("[WebSocket]: WSStream receiver closed unexpectedly"));
},
Some(result) => {
if result.is_err() {

10
shared-lib/lib-ws/src/ws.rs
View File

@ -116,7 +116,7 @@ impl WSController {
}
pub async fn retry(&self, count: usize) -> Result<(), ServerError> {
if self.sender_ctrl.read().is_connecting() {
if !self.sender_ctrl.read().is_disconnected() {
return Ok(());
}
@ -125,7 +125,7 @@ impl WSController {
.addr
.read()
.as_ref()
.expect("must call start_connect first")
.expect("Retry web socket connection failed, should call start_connect first")
.clone();
self.connect(addr, strategy).await
@ -135,7 +135,7 @@ impl WSController {
pub fn ws_message_sender(&self) -> Result<Arc<WSSender>, WSError> {
match self.sender_ctrl.read().sender() {
None => Err(WSError::internal().context("WsSender is not initialized, should call connect first")),
None => Err(WSError::internal().context("WebSocket is not initialized, should call connect first")),
Some(sender) => Ok(sender),
}
}
@ -370,10 +370,10 @@ impl WSSenderController {
fn sender(&self) -> Option<Arc<WSSender>> { self.sender.clone() }
#[allow(dead_code)]
fn is_connecting(&self) -> bool { self.state == WSConnectState::Connecting }
#[allow(dead_code)]
fn is_connected(&self) -> bool { self.state == WSConnectState::Connected }
fn is_disconnected(&self) -> bool { self.state == WSConnectState::Disconnected }
}
impl std::default::Default for WSSenderController {