add tests module and stop the system

.idea/appflowy_client.iml

@@ -6,6 +6,7 @@
       <sourceFolder url="file://$MODULE_DIR$/rust-lib/flowy-derive/src" isTestSource="false" />
       <sourceFolder url="file://$MODULE_DIR$/rust-lib/flowy-derive/tests" isTestSource="true" />
       <sourceFolder url="file://$MODULE_DIR$/rust-lib/flowy-sys/src" isTestSource="false" />
+      <sourceFolder url="file://$MODULE_DIR$/rust-lib/flowy-sys/tests" isTestSource="true" />
       <excludeFolder url="file://$MODULE_DIR$/app_flowy/packages/af_protobuf/.pub" />
       <excludeFolder url="file://$MODULE_DIR$/app_flowy/packages/af_protobuf/.dart_tool" />
       <excludeFolder url="file://$MODULE_DIR$/app_flowy/packages/af_protobuf/build" />

rust-lib/flowy-sys/src/error/error.rs

@@ -1,10 +1,14 @@
-use crate::response::{FlowyResponse, StatusCode};
-use std::{cell::RefCell, fmt};
+use crate::{
+    request::EventRequest,
+    response::{EventResponse, EventResponseBuilder, StatusCode},
+};
+use std::{fmt, option::NoneError};
+use tokio::sync::mpsc::error::SendError;
 
 pub trait Error: fmt::Debug + fmt::Display {
     fn status_code(&self) -> StatusCode;
 
-    fn as_response(&self) -> FlowyResponse { FlowyResponse::new(self.status_code()) }
+    fn as_response(&self) -> EventResponse { EventResponse::new(self.status_code()) }
 }
 
 impl<T: Error + 'static> From<T> for SystemError {
@@ -33,6 +37,57 @@ impl std::error::Error for SystemError {
     fn cause(&self) -> Option<&dyn std::error::Error> { None }
 }
 
-impl From<SystemError> for FlowyResponse {
+impl<T> From<SendError<T>> for SystemError
+where
+    T: fmt::Display + fmt::Debug + 'static,
+{
+    fn from(err: SendError<T>) -> Self { InternalError { inner: err }.into() }
+}
+
+impl From<SendError<EventRequest>> for SystemError {
+    fn from(err: SendError<EventRequest>) -> Self { InternalError { inner: err }.into() }
+}
+
+impl From<NoneError> for SystemError {
+    fn from(s: NoneError) -> Self {
+        InternalError {
+            inner: format!("Unexpected none: {:?}", s),
+        }
+        .into()
+    }
+}
+
+impl From<SystemError> for EventResponse {
     fn from(err: SystemError) -> Self { err.inner_error().as_response() }
 }
+
+pub struct InternalError<T> {
+    inner: T,
+}
+
+impl<T> InternalError<T> {
+    pub fn new(inner: T) -> Self { InternalError { inner } }
+}
+
+impl<T> fmt::Debug for InternalError<T>
+where
+    T: fmt::Debug + 'static,
+{
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { fmt::Debug::fmt(&self.inner, f) }
+}
+
+impl<T> fmt::Display for InternalError<T>
+where
+    T: fmt::Display + 'static,
+{
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { fmt::Display::fmt(&self.inner, f) }
+}
+
+impl<T> Error for InternalError<T>
+where
+    T: fmt::Debug + fmt::Display + 'static,
+{
+    fn status_code(&self) -> StatusCode { StatusCode::Err }
+
+    fn as_response(&self) -> EventResponse { EventResponseBuilder::Err().data(format!("{}", self.inner)).build() }
+}

rust-lib/flowy-sys/src/lib.rs

@@ -1,3 +1,5 @@
+#![feature(try_trait)]
+
 mod data;
 mod error;
 mod module;
@@ -6,3 +8,7 @@ mod response;
 mod rt;
 mod service;
 mod util;
+
+pub mod prelude {
+    pub use crate::{error::*, module::*, request::*, response::*, rt::*};
+}

rust-lib/flowy-sys/src/module/data.rs

@@ -1,6 +1,6 @@
 use crate::{
     error::SystemError,
-    request::{payload::Payload, FlowyRequest, FromRequest},
+    request::{payload::Payload, EventRequest, FromRequest},
     util::ready::Ready,
 };
 use std::{ops::Deref, sync::Arc};
@@ -32,5 +32,5 @@ impl<T: ?Sized + 'static> FromRequest for ModuleData<T> {
     type Future = Ready<Result<Self, SystemError>>;
 
     #[inline]
-    fn from_request(req: &FlowyRequest, _: &mut Payload) -> Self::Future { unimplemented!() }
+    fn from_request(_req: &EventRequest, _: &mut Payload) -> Self::Future { unimplemented!() }
 }

rust-lib/flowy-sys/src/module/module.rs

@@ -8,44 +8,35 @@ use crate::{
 };
 
 use crate::{
-    request::{payload::Payload, FlowyRequest},
-    response::{FlowyResponse, FlowyResponseBuilder},
+    request::{payload::Payload, EventRequest},
+    response::EventResponse,
     service::{factory, BoxServiceFactory, HandlerService},
 };
 use futures_core::{future::LocalBoxFuture, ready};
 use pin_project::pin_project;
 use std::{
-    cell::RefCell,
     collections::HashMap,
-    fmt::Debug,
     future::Future,
-    hash::Hash,
-    marker::PhantomData,
     pin::Pin,
-    rc::Rc,
-    sync::Arc,
     task::{Context, Poll},
 };
-use tokio::sync::{
-    mpsc,
-    mpsc::{unbounded_channel, UnboundedReceiver, UnboundedSender},
-};
+use tokio::sync::mpsc::{unbounded_channel, UnboundedReceiver, UnboundedSender};
 
-pub type Command = String;
-pub type CommandServiceFactory = BoxServiceFactory<(), ServiceRequest, ServiceResponse, SystemError>;
+pub type Event = String;
+pub type EventServiceFactory = BoxServiceFactory<(), ServiceRequest, ServiceResponse, SystemError>;
 
 pub struct Module {
     name: String,
     data: DataContainer,
-    service_map: HashMap<Command, CommandServiceFactory>,
-    req_tx: UnboundedSender<FlowyRequest>,
-    req_rx: UnboundedReceiver<FlowyRequest>,
-    resp_tx: UnboundedSender<FlowyResponse>,
+    service_map: HashMap<Event, EventServiceFactory>,
+    req_tx: UnboundedSender<EventRequest>,
+    req_rx: UnboundedReceiver<EventRequest>,
+    resp_tx: UnboundedSender<EventResponse>,
 }
 
 impl Module {
-    pub fn new(resp_tx: UnboundedSender<FlowyResponse>) -> Self {
-        let (req_tx, req_rx) = unbounded_channel::<FlowyRequest>();
+    pub fn new(resp_tx: UnboundedSender<EventResponse>) -> Self {
+        let (req_tx, req_rx) = unbounded_channel::<EventRequest>();
         Self {
             name: "".to_owned(),
             data: DataContainer::new(),
@@ -62,36 +53,38 @@ impl Module {
     }
 
     pub fn data<D: 'static>(mut self, data: D) -> Self {
-        let module_data = ModuleData::new(data);
-        self.data.insert(module_data);
+        self.data.insert(ModuleData::new(data));
         self
     }
 
-    pub fn event<H, T, R>(mut self, command: Command, handler: H) -> Self
+    pub fn event<H, T, R>(mut self, event: Event, handler: H) -> Self
     where
         H: Handler<T, R>,
         T: FromRequest + 'static,
         R: Future + 'static,
         R::Output: Responder + 'static,
     {
-        self.service_map.insert(command, factory(HandlerService::new(handler)));
+        if self.service_map.contains_key(&event) {
+            log::error!("Duplicate Event: {}", &event);
+        }
+
+        self.service_map.insert(event, factory(HandlerService::new(handler)));
         self
     }
 
-    pub fn can_handle(&self, cmd: &Command) -> bool { self.service_map.contains_key(cmd) }
+    pub fn req_tx(&self) -> UnboundedSender<EventRequest> { self.req_tx.clone() }
 
-    pub fn req_tx(&self) -> UnboundedSender<FlowyRequest> { self.req_tx.clone() }
-
-    pub fn handle(&self, request: FlowyRequest) {
+    pub fn handle(&self, request: EventRequest) {
+        log::trace!("Module: {} receive request: {:?}", self.name, request);
         match self.req_tx.send(request) {
             Ok(_) => {},
             Err(e) => {
-                log::error!("{:?}", e);
+                log::error!("Module: {} with error: {:?}", self.name, e);
             },
         }
     }
 
-    pub fn service_sender_map(&self) -> HashMap<Command, UnboundedSender<FlowyRequest>> {
+    pub fn forward_map(&self) -> HashMap<Event, UnboundedSender<EventRequest>> {
         self.service_map
             .keys()
             .map(|key| (key.clone(), self.req_tx()))
@@ -105,7 +98,7 @@ impl Future for Module {
         loop {
             match ready!(Pin::new(&mut self.req_rx).poll_recv(cx)) {
                 None => return Poll::Ready(()),
-                Some(request) => match self.service_map.get(request.get_cmd()) {
+                Some(request) => match self.service_map.get(request.get_event()) {
                     Some(factory) => {
                         let fut = ModuleServiceFuture {
                             request,
@@ -113,14 +106,14 @@ impl Future for Module {
                         };
                         let resp_tx = self.resp_tx.clone();
                         tokio::task::spawn_local(async move {
-                            let resp = fut.await.unwrap_or_else(|e| panic!());
+                            let resp = fut.await.unwrap_or_else(|_e| panic!());
                             if let Err(e) = resp_tx.send(resp) {
                                 log::error!("{:?}", e);
                             }
                         });
                     },
                     None => {
-                        log::error!("Command: {} handler not found", request.get_cmd());
+                        log::error!("Event: {} handler not found", request.get_event());
                     },
                 },
             }
@@ -131,18 +124,19 @@ impl Future for Module {
 type BoxModuleService = BoxService<ServiceRequest, ServiceResponse, SystemError>;
 #[pin_project]
 pub struct ModuleServiceFuture {
-    request: FlowyRequest,
+    request: EventRequest,
     #[pin]
     fut: LocalBoxFuture<'static, Result<BoxModuleService, SystemError>>,
 }
 
 impl Future for ModuleServiceFuture {
-    type Output = Result<FlowyResponse, SystemError>;
+    type Output = Result<EventResponse, SystemError>;
 
     fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
         loop {
             let service = ready!(self.as_mut().project().fut.poll(cx))?;
             let req = ServiceRequest::new(self.as_mut().request.clone(), Payload::None);
+            log::trace!("Call service to handle request {:?}", self.request);
             let (_, resp) = ready!(Pin::new(&mut service.call(req)).poll(cx))?.into_parts();
             return Poll::Ready(Ok(resp));
         }
@@ -156,29 +150,23 @@ mod tests {
     use futures_util::{future, pin_mut};
     use tokio::sync::mpsc::unbounded_channel;
 
-    pub async fn hello_service() -> String {
-        println!("no params");
-        "hello".to_string()
-    }
-
-    // #[tokio::test]
+    pub async fn hello_service() -> String { "hello".to_string() }
 
     #[test]
     fn test() {
         let mut runtime = Runtime::new().unwrap();
         runtime.block_on(async {
-            let (resp_tx, mut resp_rx) = unbounded_channel::<FlowyResponse>();
-            let command = "hello".to_string();
-            let mut module = Module::new(resp_tx).event(command.clone(), hello_service);
-            assert_eq!(module.can_handle(&command), true);
+            let (resp_tx, mut resp_rx) = unbounded_channel::<EventResponse>();
+            let event = "hello".to_string();
+            let mut module = Module::new(resp_tx).event(event.clone(), hello_service);
             let req_tx = module.req_tx();
             let mut event = async move {
-                let request = FlowyRequest::new(command.clone());
+                let request = EventRequest::new(event.clone());
                 req_tx.send(request).unwrap();
 
                 match resp_rx.recv().await {
                     Some(resp) => {
-                        println!("{}", resp);
+                        log::info!("{}", resp);
                     },
                     None => panic!(""),
                 }

rust-lib/flowy-sys/src/request/mod.rs

@@ -1,4 +1,5 @@
-pub use request::*;
-
 pub mod payload;
 mod request;
+
+pub use payload::*;
+pub use request::*;

rust-lib/flowy-sys/src/request/request.rs

@@ -5,32 +5,31 @@ use crate::{
     request::payload::Payload,
     util::ready::{ready, Ready},
 };
-use std::hash::Hash;
 
 #[derive(Clone, Debug)]
-pub struct FlowyRequest {
+pub struct EventRequest {
     id: String,
-    cmd: String,
+    event: String,
 }
 
-impl FlowyRequest {
-    pub fn new(cmd: String) -> FlowyRequest {
+impl EventRequest {
+    pub fn new(event: String) -> EventRequest {
         Self {
             id: uuid::Uuid::new_v4().to_string(),
-            cmd,
+            event,
         }
     }
 }
 
-impl FlowyRequest {
-    pub fn get_cmd(&self) -> &str { &self.cmd }
+impl EventRequest {
+    pub fn get_event(&self) -> &str { &self.event }
 }
 
 pub trait FromRequest: Sized {
     type Error: Into<SystemError>;
     type Future: Future<Output = Result<Self, Self::Error>>;
 
-    fn from_request(req: &FlowyRequest, payload: &mut Payload) -> Self::Future;
+    fn from_request(req: &EventRequest, payload: &mut Payload) -> Self::Future;
 }
 
 #[doc(hidden)]
@@ -38,7 +37,7 @@ impl FromRequest for () {
     type Error = SystemError;
     type Future = Ready<Result<(), SystemError>>;
 
-    fn from_request(_req: &FlowyRequest, _payload: &mut Payload) -> Self::Future { ready(Ok(())) }
+    fn from_request(_req: &EventRequest, _payload: &mut Payload) -> Self::Future { ready(Ok(())) }
 }
 
 #[doc(hidden)]
@@ -46,5 +45,5 @@ impl FromRequest for String {
     type Error = SystemError;
     type Future = Ready<Result<String, SystemError>>;
 
-    fn from_request(_req: &FlowyRequest, _payload: &mut Payload) -> Self::Future { ready(Ok("".to_string())) }
+    fn from_request(_req: &EventRequest, _payload: &mut Payload) -> Self::Future { ready(Ok("".to_string())) }
 }

rust-lib/flowy-sys/src/response/builder.rs

@@ -1,24 +1,24 @@
 use crate::{
     error::SystemError,
-    response::{data::ResponseData, FlowyResponse, StatusCode},
+    response::{data::ResponseData, EventResponse, StatusCode},
 };
 
 macro_rules! static_response {
     ($name:ident, $status:expr) => {
         #[allow(non_snake_case, missing_docs)]
-        pub fn $name() -> FlowyResponseBuilder { FlowyResponseBuilder::new($status) }
+        pub fn $name() -> EventResponseBuilder { EventResponseBuilder::new($status) }
     };
 }
 
-pub struct FlowyResponseBuilder<T = ResponseData> {
+pub struct EventResponseBuilder<T = ResponseData> {
     pub data: T,
     pub status: StatusCode,
     pub error: Option<SystemError>,
 }
 
-impl FlowyResponseBuilder {
+impl EventResponseBuilder {
     pub fn new(status: StatusCode) -> Self {
-        FlowyResponseBuilder {
+        EventResponseBuilder {
             data: ResponseData::None,
             status,
             error: None,
@@ -35,13 +35,14 @@ impl FlowyResponseBuilder {
         self
     }
 
-    pub fn build(self) -> FlowyResponse {
-        FlowyResponse {
+    pub fn build(self) -> EventResponse {
+        EventResponse {
             data: self.data,
             status: self.status,
             error: self.error,
         }
     }
 
-    static_response!(Ok, StatusCode::Success);
+    static_response!(Ok, StatusCode::Ok);
+    static_response!(Err, StatusCode::Err);
 }

rust-lib/flowy-sys/src/response/responder.rs

@@ -1,17 +1,16 @@
-use crate::request::FlowyRequest;
-use crate::response::FlowyResponse;
-use crate::response::FlowyResponseBuilder;
+use crate::{
+    request::EventRequest,
+    response::{EventResponse, EventResponseBuilder},
+};
 
 pub trait Responder {
-    fn respond_to(self, req: &FlowyRequest) -> FlowyResponse;
+    fn respond_to(self, req: &EventRequest) -> EventResponse;
 }
 
 macro_rules! impl_responder {
     ($res: ty) => {
         impl Responder for $res {
-            fn respond_to(self, _: &FlowyRequest) -> FlowyResponse {
-                FlowyResponseBuilder::Ok().data(self).build()
-            }
+            fn respond_to(self, _: &EventRequest) -> EventResponse { EventResponseBuilder::Ok().data(self).build() }
         }
     };
 }

rust-lib/flowy-sys/src/response/response.rs

@@ -1,30 +1,31 @@
 use crate::{
     error::SystemError,
-    request::FlowyRequest,
+    request::EventRequest,
     response::{data::ResponseData, Responder},
 };
-use serde::{Deserialize, Serialize};
-use serde_with::skip_serializing_none;
+use serde::{Deserialize, Serialize, Serializer};
+
 use std::{fmt, fmt::Formatter};
 
-#[derive(Clone, Copy, Debug, Serialize, Deserialize)]
+#[derive(Clone, Debug, Serialize, Deserialize)]
 pub enum StatusCode {
-    Success,
-    Error,
+    Ok,
+    Err,
 }
 
-#[skip_serializing_none]
-#[derive(Serialize, Deserialize, Debug)]
-pub struct FlowyResponse<T = ResponseData> {
-    pub data: T,
+// serde user guide: https://serde.rs/field-attrs.html
+#[derive(Serialize, Debug)]
+pub struct EventResponse {
+    #[serde(serialize_with = "serialize_data")]
+    pub data: ResponseData,
     pub status: StatusCode,
-    #[serde(skip)]
+    #[serde(serialize_with = "serialize_error")]
     pub error: Option<SystemError>,
 }
 
-impl FlowyResponse {
+impl EventResponse {
     pub fn new(status: StatusCode) -> Self {
-        FlowyResponse {
+        EventResponse {
             data: ResponseData::None,
             status,
             error: None,
@@ -32,7 +33,7 @@ impl FlowyResponse {
     }
 }
 
-impl std::fmt::Display for FlowyResponse {
+impl std::fmt::Display for EventResponse {
     fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
         match serde_json::to_string(self) {
             Ok(json) => f.write_fmt(format_args!("{:?}", json))?,
@@ -42,7 +43,27 @@ impl std::fmt::Display for FlowyResponse {
     }
 }
 
-impl Responder for FlowyResponse {
+impl Responder for EventResponse {
     #[inline]
-    fn respond_to(self, _: &FlowyRequest) -> FlowyResponse { self }
+    fn respond_to(self, _: &EventRequest) -> EventResponse { self }
+}
+
+fn serialize_error<S>(error: &Option<SystemError>, serializer: S) -> Result<S::Ok, S::Error>
+where
+    S: Serializer,
+{
+    match error {
+        Some(e) => serializer.serialize_str(&format!("{:?}", e)),
+        None => serializer.serialize_str(""),
+    }
+}
+
+fn serialize_data<S>(data: &ResponseData, serializer: S) -> Result<S::Ok, S::Error>
+where
+    S: Serializer,
+{
+    match data {
+        ResponseData::Bytes(bytes) => serializer.serialize_str(&format!("{} bytes", bytes.len())),
+        ResponseData::None => serializer.serialize_str(""),
+    }
 }

rust-lib/flowy-sys/src/rt/mod.rs

@@ -2,3 +2,4 @@ mod runtime;
 mod system;
 
 pub use runtime::*;
+pub use system::*;

rust-lib/flowy-sys/src/rt/system.rs

@@ -1,12 +1,12 @@
 use crate::{
-    module::{Command, CommandServiceFactory, Module},
-    request::FlowyRequest,
-    response::FlowyResponse,
-    rt::Runtime,
-    service::BoxServiceFactory,
+    module::{Event, Module},
+    request::EventRequest,
+    response::EventResponse,
+    rt::runtime::Runtime,
 };
-use futures_core::{future::LocalBoxFuture, ready, task::Context};
-use futures_util::{future, pin_mut};
+use futures_core::{ready, task::Context};
+
+use crate::error::{InternalError, SystemError};
 use std::{cell::RefCell, collections::HashMap, future::Future, io, sync::Arc};
 use tokio::{
     macros::support::{Pin, Poll},
@@ -20,30 +20,40 @@ thread_local!(
     static CURRENT: RefCell<Option<Arc<FlowySystem>>> = RefCell::new(None);
 );
 
+enum SystemCommand {
+    Exit(i8),
+}
+
 pub struct FlowySystem {
-    resp_tx: UnboundedSender<FlowyResponse>,
-    sender_map: HashMap<Command, UnboundedSender<FlowyRequest>>,
+    sys_tx: UnboundedSender<SystemCommand>,
+    forward_map: HashMap<Event, UnboundedSender<EventRequest>>,
 }
 
 impl FlowySystem {
     pub fn construct<F>(module_factory: F) -> SystemRunner
     where
-        F: FnOnce(UnboundedSender<FlowyResponse>) -> Vec<Module>,
+        F: FnOnce(UnboundedSender<EventResponse>) -> Vec<Module>,
     {
         let runtime = Runtime::new().unwrap();
-        let (resp_tx, mut resp_rx) = unbounded_channel::<FlowyResponse>();
+        let (resp_tx, resp_rx) = unbounded_channel::<EventResponse>();
+
+        let (sys_tx, sys_rx) = unbounded_channel::<SystemCommand>();
         let (stop_tx, stop_rx) = oneshot::channel();
-        let controller = SystemController { resp_rx, stop_tx };
-        runtime.spawn(controller);
+
+        runtime.spawn(SystemFFI { resp_rx });
+        runtime.spawn(SystemController {
+            stop_tx: Some(stop_tx),
+            sys_rx,
+        });
 
         let mut system = Self {
-            resp_tx: resp_tx.clone(),
-            sender_map: HashMap::default(),
+            sys_tx,
+            forward_map: HashMap::default(),
         };
 
         let factory = module_factory(resp_tx.clone());
         factory.into_iter().for_each(|m| {
-            system.sender_map.extend(m.service_sender_map());
+            system.forward_map.extend(m.forward_map());
             runtime.spawn(m);
         });
 
@@ -52,9 +62,18 @@ impl FlowySystem {
         runner
     }
 
-    pub fn handle_command(&self, cmd: Command, request: FlowyRequest) {
-        if let Some(sender) = self.sender_map.get(&cmd) {
-            sender.send(request);
+    pub fn sink(&self, event: Event, request: EventRequest) -> Result<(), SystemError> {
+        log::trace!("Sink event: {}", event);
+        let _ = self.forward_map.get(&event)?.send(request)?;
+        Ok(())
+    }
+
+    pub fn stop(&self) {
+        match self.sys_tx.send(SystemCommand::Exit(0)) {
+            Ok(_) => {},
+            Err(e) => {
+                log::error!("Stop system error: {}", e);
+            },
         }
     }
 
@@ -71,24 +90,43 @@ impl FlowySystem {
             None => panic!("System is not running"),
         })
     }
+}
+
+struct SystemFFI {
+    resp_rx: UnboundedReceiver<EventResponse>,
+}
 
-    pub(crate) fn resp_tx(&self) -> UnboundedSender<FlowyResponse> { self.resp_tx.clone() }
+impl Future for SystemFFI {
+    type Output = ();
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        loop {
+            match ready!(Pin::new(&mut self.resp_rx).poll_recv(cx)) {
+                None => return Poll::Ready(()),
+                Some(resp) => {
+                    log::trace!("Response: {:?}", resp);
+                },
+            }
+        }
+    }
 }
 
 struct SystemController {
-    resp_rx: UnboundedReceiver<FlowyResponse>,
-    stop_tx: oneshot::Sender<i32>,
+    stop_tx: Option<oneshot::Sender<i8>>,
+    sys_rx: UnboundedReceiver<SystemCommand>,
 }
 
 impl Future for SystemController {
     type Output = ();
     fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
         loop {
-            match ready!(Pin::new(&mut self.resp_rx).poll_recv(cx)) {
+            match ready!(Pin::new(&mut self.sys_rx).poll_recv(cx)) {
                 None => return Poll::Ready(()),
-                Some(resp) => {
-                    // FF
-                    println!("Receive response: {:?}", resp);
+                Some(cmd) => match cmd {
+                    SystemCommand::Exit(code) => {
+                        if let Some(tx) = self.stop_tx.take() {
+                            let _ = tx.send(code);
+                        }
+                    },
                 },
             }
         }
@@ -97,7 +135,7 @@ impl Future for SystemController {
 
 pub struct SystemRunner {
     rt: Runtime,
-    stop_rx: oneshot::Receiver<i32>,
+    stop_rx: oneshot::Receiver<i8>,
 }
 
 impl SystemRunner {
@@ -126,28 +164,3 @@ impl SystemRunner {
         self
     }
 }
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-
-    pub async fn hello_service() -> String { "hello".to_string() }
-
-    #[test]
-    fn test() {
-        let command = "Hello".to_string();
-
-        FlowySystem::construct(|tx| {
-            vec![
-                Module::new(tx.clone()).event(command.clone(), hello_service),
-                // Module::new(tx.clone()).event(command.clone(), hello_service),
-            ]
-        })
-        .spawn(async {
-            let request = FlowyRequest::new(command.clone());
-            FlowySystem::current().handle_command(command, request);
-        })
-        .run()
-        .unwrap();
-    }
-}

rust-lib/flowy-sys/src/service/boxed.rs

@@ -1,8 +1,4 @@
-use crate::{
-    service::{Service, ServiceFactory, ServiceRequest, ServiceResponse},
-    util::ready::*,
-};
-
+use crate::service::{Service, ServiceFactory};
 use futures_core::future::LocalBoxFuture;
 
 pub fn factory<SF, Req>(factory: SF) -> BoxServiceFactory<SF::Config, Req, SF::Response, SF::Error>
@@ -47,6 +43,7 @@ type Inner<Cfg, Req, Res, Err> = Box<
 pub type BoxService<Req, Res, Err> =
     Box<dyn Service<Req, Response = Res, Error = Err, Future = LocalBoxFuture<'static, Result<Res, Err>>>>;
 
+#[allow(dead_code)]
 pub fn service<S, Req>(service: S) -> BoxService<Req, S::Response, S::Error>
 where
     S: Service<Req> + 'static,

rust-lib/flowy-sys/src/service/handler.rs

@@ -10,8 +10,8 @@ use pin_project::pin_project;
 
 use crate::{
     error::SystemError,
-    request::{payload::Payload, FlowyRequest, FromRequest},
-    response::{FlowyResponse, Responder},
+    request::{payload::Payload, EventRequest, FromRequest},
+    response::{EventResponse, Responder},
     service::{Service, ServiceFactory, ServiceRequest, ServiceResponse},
     util::ready::*,
 };
@@ -107,8 +107,8 @@ where
     R: Future,
     R::Output: Responder,
 {
-    Extract(#[pin] T::Future, Option<FlowyRequest>, H),
-    Handle(#[pin] R, Option<FlowyRequest>),
+    Extract(#[pin] T::Future, Option<EventRequest>, H),
+    Handle(#[pin] R, Option<EventRequest>),
 }
 
 impl<F, T, R> Future for HandlerServiceFuture<F, T, R>
@@ -133,7 +133,7 @@ where
                         Err(err) => {
                             let req = req.take().unwrap();
                             let system_err: SystemError = err.into();
-                            let res: FlowyResponse = system_err.into();
+                            let res: EventResponse = system_err.into();
                             return Poll::Ready(Ok(ServiceResponse::new(req, res)));
                         },
                     };
@@ -178,7 +178,7 @@ macro_rules! tuple_from_req ({$tuple_type:ident, $(($n:tt, $T:ident)),+} => {
             type Error = SystemError;
             type Future = $tuple_type<$($T),+>;
 
-            fn from_request(req: &FlowyRequest, payload: &mut Payload) -> Self::Future {
+            fn from_request(req: &EventRequest, payload: &mut Payload) -> Self::Future {
                 $tuple_type {
                     items: <($(Option<$T>,)+)>::default(),
                     futs: FromRequestFutures($($T::from_request(req, payload),)+),

rust-lib/flowy-sys/src/service/service.rs

@@ -1,8 +1,8 @@
 use std::future::Future;
 
 use crate::{
-    request::{payload::Payload, FlowyRequest},
-    response::{data::ResponseData, FlowyResponse},
+    request::{payload::Payload, EventRequest},
+    response::EventResponse,
 };
 
 pub trait Service<Request> {
@@ -24,24 +24,24 @@ pub trait ServiceFactory<Request> {
 }
 
 pub struct ServiceRequest {
-    req: FlowyRequest,
+    req: EventRequest,
     payload: Payload,
 }
 
 impl ServiceRequest {
-    pub fn new(req: FlowyRequest, payload: Payload) -> Self { Self { req, payload } }
+    pub fn new(req: EventRequest, payload: Payload) -> Self { Self { req, payload } }
 
     #[inline]
-    pub fn into_parts(self) -> (FlowyRequest, Payload) { (self.req, self.payload) }
+    pub fn into_parts(self) -> (EventRequest, Payload) { (self.req, self.payload) }
 }
 
-pub struct ServiceResponse<T = ResponseData> {
-    request: FlowyRequest,
-    response: FlowyResponse<T>,
+pub struct ServiceResponse {
+    request: EventRequest,
+    response: EventResponse,
 }
 
-impl<T> ServiceResponse<T> {
-    pub fn new(request: FlowyRequest, response: FlowyResponse<T>) -> Self { ServiceResponse { request, response } }
+impl ServiceResponse {
+    pub fn new(request: EventRequest, response: EventResponse) -> Self { ServiceResponse { request, response } }
 
-    pub fn into_parts(self) -> (FlowyRequest, FlowyResponse<T>) { (self.request, self.response) }
+    pub fn into_parts(self) -> (EventRequest, EventResponse) { (self.request, self.response) }
 }

rust-lib/flowy-sys/tests/api/helper.rs

@@ -0,0 +1,16 @@
+use std::sync::Once;
+
+#[allow(dead_code)]
+pub fn setup_env() {
+    static INIT: Once = Once::new();
+    INIT.call_once(|| {
+        std::env::set_var("RUST_LOG", "flowy_sys=trace,trace");
+        env_logger::init();
+    });
+}
+
+pub struct ExecutorAction {
+    command: String,
+}
+
+pub struct FlowySystemExecutor {}

rust-lib/flowy-sys/tests/api/main.rs

@@ -0,0 +1,2 @@
+mod helper;
+mod module_event;

rust-lib/flowy-sys/tests/api/module_event.rs

@@ -0,0 +1,29 @@
+use crate::helper::*;
+use flowy_sys::prelude::*;
+
+pub async fn no_params() -> String { "no params function call".to_string() }
+pub async fn one_params(s: String) -> String { "one params function call".to_string() }
+pub async fn two_params(s1: String, s2: String) -> String { "two params function call".to_string() }
+
+#[test]
+fn test() {
+    setup_env();
+
+    let no_params_command = "no params".to_string();
+    let one_params_command = "one params".to_string();
+    let two_params_command = "two params".to_string();
+    FlowySystem::construct(|tx| {
+        vec![Module::new(tx.clone())
+            .event(no_params_command.clone(), no_params)
+            .event(one_params_command.clone(), one_params)
+            .event(two_params_command.clone(), two_params)]
+    })
+    .spawn(async {
+        let request = EventRequest::new(no_params_command.clone());
+        FlowySystem::current().sink(no_params_command, request);
+
+        FlowySystem::current().stop();
+    })
+    .run()
+    .unwrap();
+}

rust-lib/rustfmt.toml

@@ -5,12 +5,14 @@ fn_single_line = true
 match_block_trailing_comma = true
 normalize_comments = true
 wrap_comments = true
-merge_imports = true
 use_field_init_shorthand = true
 use_try_shorthand = true
 normalize_doc_attributes = true
 report_todo = "Always"
 report_fixme = "Always"
 imports_layout = "HorizontalVertical"
+merge_imports = true
+reorder_modules = true
+reorder_imports = true
 enum_discrim_align_threshold = 20
 edition = "2018"