use backend_service::configuration::ClientServerConfiguration; use bytes::Bytes; use flowy_collaboration::entities::{ doc::{CreateDocParams, DocumentId, DocumentInfo, ResetDocumentParams}, ws::DocumentClientWSData, }; use flowy_database::ConnectionPool; use flowy_document::{ context::DocumentUser, core::{DocumentWSReceivers, DocumentWebSocket, WSStateReceiver}, errors::{internal_error, FlowyError}, DocumentCloudService, }; use flowy_net::{ cloud::document::{DocumentHttpCloudService, DocumentLocalCloudService}, services::ws_conn::FlowyWebSocketConnect, }; use flowy_user::services::UserSession; use lib_infra::future::FutureResult; use lib_ws::{WSMessageReceiver, WSModule, WebSocketRawMessage}; use std::{convert::TryInto, path::Path, sync::Arc}; pub struct DocumentDependencies { pub user: Arc, pub ws_receivers: Arc, pub ws_sender: Arc, pub cloud_service: Arc, } pub struct DocumentDepsResolver(); impl DocumentDepsResolver { pub fn resolve( ws_conn: Arc, user_session: Arc, server_config: &ClientServerConfiguration, ) -> DocumentDependencies { let user = Arc::new(DocumentUserImpl(user_session)); let ws_sender = Arc::new(DocumentWebSocketImpl(ws_conn.clone())); let ws_receivers = Arc::new(DocumentWSReceivers::new()); let receiver = Arc::new(WSMessageReceiverImpl(ws_receivers.clone())); ws_conn.add_ws_message_receiver(receiver).unwrap(); let cloud_service = make_document_cloud_service(server_config); DocumentDependencies { user, ws_receivers, ws_sender, cloud_service, } } } struct DocumentUserImpl(Arc); impl DocumentUser for DocumentUserImpl { fn user_dir(&self) -> Result { let dir = self.0.user_dir().map_err(|e| FlowyError::unauthorized().context(e))?; let doc_dir = format!("{}/document", dir); if !Path::new(&doc_dir).exists() { let _ = std::fs::create_dir_all(&doc_dir)?; } Ok(doc_dir) } fn user_id(&self) -> Result { self.0.user_id() } fn token(&self) -> Result { self.0.token() } fn db_pool(&self) -> Result, FlowyError> { self.0.db_pool() } } struct DocumentWebSocketImpl(Arc); impl DocumentWebSocket for DocumentWebSocketImpl { fn send(&self, data: DocumentClientWSData) -> Result<(), FlowyError> { let bytes: Bytes = data.try_into().unwrap(); let msg = WebSocketRawMessage { module: WSModule::Doc, data: bytes.to_vec(), }; let sender = self.0.ws_sender()?; sender.send(msg).map_err(internal_error)?; Ok(()) } fn subscribe_state_changed(&self) -> WSStateReceiver { self.0.subscribe_websocket_state() } } struct WSMessageReceiverImpl(Arc); impl WSMessageReceiver for WSMessageReceiverImpl { fn source(&self) -> WSModule { WSModule::Doc } fn receive_message(&self, msg: WebSocketRawMessage) { self.0.did_receive_data(Bytes::from(msg.data)); } } fn make_document_cloud_service(server_config: &ClientServerConfiguration) -> Arc { if cfg!(feature = "http_server") { Arc::new(DocumentHttpCloudServiceAdaptor::new(server_config.clone())) } else { Arc::new(DocumentLocalCloudServiceAdaptor::new()) } } struct DocumentHttpCloudServiceAdaptor(DocumentHttpCloudService); impl DocumentHttpCloudServiceAdaptor { fn new(config: ClientServerConfiguration) -> Self { DocumentHttpCloudServiceAdaptor(DocumentHttpCloudService::new(config)) } } impl DocumentCloudService for DocumentHttpCloudServiceAdaptor { fn create_document(&self, token: &str, params: CreateDocParams) -> FutureResult<(), FlowyError> { self.0.create_document_request(token, params) } fn read_document(&self, token: &str, params: DocumentId) -> FutureResult, FlowyError> { self.0.read_document_request(token, params) } fn update_document(&self, token: &str, params: ResetDocumentParams) -> FutureResult<(), FlowyError> { self.0.update_document_request(token, params) } } struct DocumentLocalCloudServiceAdaptor(DocumentLocalCloudService); impl DocumentLocalCloudServiceAdaptor { fn new() -> Self { Self(DocumentLocalCloudService {}) } } impl DocumentCloudService for DocumentLocalCloudServiceAdaptor { fn create_document(&self, token: &str, params: CreateDocParams) -> FutureResult<(), FlowyError> { self.0.create_document_request(token, params) } fn read_document(&self, token: &str, params: DocumentId) -> FutureResult, FlowyError> { self.0.read_document_request(token, params) } fn update_document(&self, token: &str, params: ResetDocumentParams) -> FutureResult<(), FlowyError> { self.0.update_document_request(token, params) } }