Rename ServerFactory->ServerFact and decouple ServerTransport

Author: frostyplanet

stream/src/server/dispatch.rs

@@ -3,7 +3,7 @@ use super::task::*;
 use occams_rpc_core::Codec;
 use std::marker::PhantomData;
 
-/// ReqDispatch should be a user-defined struct initialized for every connection, by ServerFactory::new_dispatcher.
+/// ReqDispatch should be a user-defined struct initialized for every connection, by ServerFact::new_dispatcher.
 ///
 /// ReqDispatch must have Sync, because the connection reader and writer access concurrently.
 ///
@@ -31,8 +31,8 @@ pub trait ReqDispatch<R: ServerTaskResp>: Send + Sync + Sized + 'static {
 /// # Example
 ///
 /// ```no_compile,ignore
-/// use occams_rpc_stream::server::{ServerFactory, ReqDispatch};
-/// impl ServerFactory for YourServer {
+/// use occams_rpc_stream::server::{ServerFact, ReqDispatch};
+/// impl ServerFact for YourServer {
 ///
 ///     ...
 ///

stream/src/server/mod.rs

@@ -19,24 +19,19 @@ use occams_rpc_core::{Codec, error::*, io::*, runtime::AsyncIO};
 use std::{fmt, future::Future, io, sync::Arc};
 
 /// A central hub defined by the user for the server-side, to define the customizable plugin.
-pub trait ServerFactory: Sync + Send + 'static + Sized {
+pub trait ServerFact: Sync + Send + 'static + Sized {
     /// A [captains-log::filter::Filter](https://docs.rs/captains-log/latest/captains_log/filter/index.html) implementation.
     /// The type that new_logger returns.
     ///
     /// maybe a `Arc<LogFilter> or KeyFilter<Arc<LogFilter>>`
-    type Logger: Filter + Send + 'static;
-
-    /// Define the transport layer protocol
-    ///
-    /// Refers to [ServerTransport]
-    type Transport: ServerTransport<Self>;
+    type Logger: Filter + Send + Sync + 'static;
 
     /// Define the adaptor of async runtime
     ///
     /// Refers to [occams_rpc_core::runtime::AsyncIO](https://docs.rs/occams-rpc-core/latest/occams_rpc_core/runtime/index.html)
     type IO: AsyncIO;
 
-    /// You should keep ServerConfig inside ServerFactory, get_config() will return the reference.
+    /// You should keep ServerConfig inside ServerFact, get_config() will return the reference.
     fn get_config(&self) -> &ServerConfig;
 
     /// Construct a [captains_log::filter::Filter](https://docs.rs/captains-log/latest/captains_log/filter/trait.Filter.html) to oganize log of a client
@@ -65,37 +60,36 @@ pub trait ServerFactory: Sync + Send + 'static + Sized {
 /// The implementation can be found on:
 ///
 /// - [occams-rpc-tcp](https://docs.rs/occams-rpc-tcp): For TCP and Unix socket
-pub trait ServerTransport<F: ServerFactory>: Send + Sync + Sized + 'static + fmt::Debug {
+pub trait ServerTransport<IO: AsyncIO>: Send + Sync + Sized + 'static + fmt::Debug {
     type Listener: AsyncListener;
 
-    /// The ServerTransport holds a logger, the server will use it by reference.
-    fn get_logger(&self) -> &F::Logger;
-
     /// The implementation is expected to store the conn_count until dropped
     fn new_conn(
-        stream: <Self::Listener as AsyncListener>::Conn, f: &F, conn_count: Arc<()>,
+        stream: <Self::Listener as AsyncListener>::Conn, config: &ServerConfig, conn_count: Arc<()>,
     ) -> Self;
 
     /// Read a request from the socket
-    fn read_req<'a>(
-        &'a self, close_ch: &crossfire::MAsyncRx<()>,
+    fn read_req<'a, F: ServerFact>(
+        &'a self, logger: &F::Logger, close_ch: &crossfire::MAsyncRx<()>,
     ) -> impl Future<Output = Result<RpcSvrReq<'a>, RpcIntErr>> + Send;
 
     /// Write our user task response
-    fn write_resp<C: Codec, T: ServerTaskEncode>(
-        &self, codec: &C, task: T,
+    fn write_resp<F: ServerFact, T: ServerTaskEncode>(
+        &self, logger: &F::Logger, codec: &impl Codec, task: T,
     ) -> impl Future<Output = io::Result<()>> + Send;
 
     /// Write out ping resp or error
-    fn write_resp_internal(
-        &self, seq: u64, err: Option<RpcIntErr>,
+    fn write_resp_internal<F: ServerFact>(
+        &self, logger: &F::Logger, seq: u64, err: Option<RpcIntErr>,
     ) -> impl Future<Output = io::Result<()>> + Send;
 
     /// Flush the response for the socket writer, if the transport has buffering logic
-    fn flush_resp(&self) -> impl Future<Output = io::Result<()>> + Send;
+    fn flush_resp<F: ServerFact>(
+        &self, logger: &F::Logger,
+    ) -> impl Future<Output = io::Result<()>> + Send;
 
     /// Shutdown the write direction of the connection
-    fn close_conn(&self) -> impl Future<Output = ()> + Send;
+    fn close_conn<F: ServerFact>(&self, logger: &F::Logger) -> impl Future<Output = ()> + Send;
 }
 
 /// A temporary struct to hold data buffer return by ServerTransport

stream/src/server/server.rs

@@ -6,10 +6,10 @@ use std::io;
 use std::sync::{Arc, Mutex};
 use std::time::{Duration, Instant};
 
-/// An RpcServer that listen, accept, and server connections, according to ServerFactory interface.
+/// An RpcServer that listen, accept, and server connections, according to ServerFact interface.
 pub struct RpcServer<F>
 where
-    F: ServerFactory,
+    F: ServerFact,
 {
     listeners_abort: Vec<(AbortHandle, String)>,
     logger: F::Logger,
@@ -21,7 +21,7 @@ where
 
 impl<F> RpcServer<F>
 where
-    F: ServerFactory,
+    F: ServerFact,
 {
     pub fn new(factory: Arc<F>) -> Self {
         let (tx, rx) = crossfire::mpmc::unbounded_async();
@@ -35,8 +35,8 @@ where
         }
     }
 
-    pub fn listen(&mut self, addr: &str) -> io::Result<String> {
-        match <<F::Transport as ServerTransport<F>>::Listener as AsyncListener>::bind(addr) {
+    pub fn listen<T: ServerTransport<F::IO>>(&mut self, addr: &str) -> io::Result<String> {
+        match <T::Listener as AsyncListener>::bind(addr) {
             Err(e) => {
                 error!("bind addr {:?} err: {}", addr, e);
                 return Err(e);
@@ -68,12 +68,12 @@ where
                                     return;
                                 }
                                 Ok(stream) => {
-                                    let conn = F::Transport::new_conn(
+                                    let conn = T::new_conn(
                                         stream,
-                                        &factory,
+                                        factory.get_config(),
                                         conn_ref_count.clone(),
                                     );
-                                    Self::server_conn(conn, &factory, server_close_rx.clone())
+                                    Self::server_conn::<T>(conn, &factory, server_close_rx.clone())
                                 }
                             }
                         }
@@ -88,24 +88,34 @@ where
         }
     }
 
-    fn server_conn(conn: F::Transport, factory: &F, server_close_rx: crossfire::MAsyncRx<()>) {
+    fn server_conn<T: ServerTransport<F::IO>>(
+        conn: T, factory: &F, server_close_rx: crossfire::MAsyncRx<()>,
+    ) {
         let conn = Arc::new(conn);
 
         let dispatch = Arc::new(factory.new_dispatcher());
         let (done_tx, done_rx) = crossfire::mpsc::unbounded_async();
 
         let noti = RespNoti(done_tx);
-        struct Reader<F: ServerFactory, D: ReqDispatch<R>, R: ServerTaskResp> {
+        struct Reader<
+            T: ServerTransport<F::IO>,
+            F: ServerFact,
+            D: ReqDispatch<R>,
+            R: ServerTaskResp,
+        > {
             noti: RespNoti<R>,
-            conn: Arc<F::Transport>,
+            conn: Arc<T>,
             server_close_rx: crossfire::MAsyncRx<()>,
             dispatch: Arc<D>,
+            logger: F::Logger,
         }
 
-        impl<F: ServerFactory, D: ReqDispatch<R>, R: ServerTaskResp> Reader<F, D, R> {
+        impl<T: ServerTransport<F::IO>, F: ServerFact, D: ReqDispatch<R>, R: ServerTaskResp>
+            Reader<T, F, D, R>
+        {
             async fn run(self) -> Result<(), ()> {
                 loop {
-                    match self.conn.read_req(&self.server_close_rx).await {
+                    match self.conn.read_req::<F>(&self.logger, &self.server_close_rx).await {
                         Ok(req) => {
                             if req.action == RpcAction::Num(0) && req.msg.len() == 0 {
                                 // ping request
@@ -129,41 +139,52 @@ where
             #[inline]
             fn send_quick_resp(&self, seq: u64, err: Option<RpcIntErr>) -> Result<(), ()> {
                 if self.noti.send_err(seq, err).is_err() {
-                    logger_warn!(
-                        self.conn.get_logger(),
-                        "{:?} reader abort due to writer has err",
-                        self.conn
-                    );
+                    logger_warn!(self.logger, "{:?} reader abort due to writer has err", self.conn);
                     return Err(());
                 }
                 Ok(())
             }
         }
-        let reader = Reader::<F, _, _> {
+        let reader = Reader::<T, F, _, _> {
             noti,
             conn: conn.clone(),
             server_close_rx,
             dispatch: dispatch.clone(),
+            logger: factory.new_logger(),
         };
         factory.spawn_detach(async move { reader.run().await });
 
-        struct Writer<F: ServerFactory, D: ReqDispatch<R>, R: ServerTaskResp> {
+        struct Writer<
+            T: ServerTransport<F::IO>,
+            F: ServerFact,
+            D: ReqDispatch<R>,
+            R: ServerTaskResp,
+        > {
             dispatch: Arc<D>,
             done_rx: crossfire::AsyncRx<Result<R, (u64, Option<RpcIntErr>)>>,
-            conn: Arc<F::Transport>,
+            conn: Arc<T>,
+            logger: F::Logger,
         }
 
-        impl<F: ServerFactory, D: ReqDispatch<R>, R: ServerTaskResp> Writer<F, D, R> {
+        impl<T: ServerTransport<F::IO>, F: ServerFact, D: ReqDispatch<R>, R: ServerTaskResp>
+            Writer<T, F, D, R>
+        {
             async fn run(self) -> Result<(), io::Error> {
                 macro_rules! process {
                     ($task: expr) => {{
                         match $task {
                             Ok(_task) => {
-                                logger_trace!(self.conn.get_logger(), "write_resp {:?}", _task);
-                                self.conn.write_resp(self.dispatch.get_codec(), _task).await?;
+                                logger_trace!(self.logger, "write_resp {:?}", _task);
+                                self.conn
+                                    .write_resp::<F, R>(
+                                        &self.logger,
+                                        self.dispatch.get_codec(),
+                                        _task,
+                                    )
+                                    .await?;
                             }
                             Err((seq, err)) => {
-                                self.conn.write_resp_internal(seq, err).await?;
+                                self.conn.write_resp_internal::<F>(&self.logger, seq, err).await?;
                             }
                         }
                     }};
@@ -173,14 +194,14 @@ where
                     while let Ok(task) = self.done_rx.try_recv() {
                         process!(task);
                     }
-                    self.conn.flush_resp().await?;
+                    self.conn.flush_resp::<F>(&self.logger).await?;
                 }
-                logger_trace!(self.conn.get_logger(), "{:?} writer exits", self.conn);
-                self.conn.close_conn().await;
+                logger_trace!(self.logger, "{:?} writer exits", self.conn);
+                self.conn.close_conn::<F>(&self.logger).await;
                 Ok(())
             }
         }
-        let writer = Writer::<F, _, _> { done_rx, conn, dispatch };
+        let writer = Writer::<T, F, _, _> { done_rx, conn, dispatch, logger: factory.new_logger() };
         factory.spawn_detach(async move { writer.run().await });
     }
 

stream/test/src/server.rs

@@ -1,6 +1,7 @@
 use super::client::{FileAction, FileIOReq, FileIOResp, FileOpenReq};
 use occams_rpc_codec::MsgpCodec;
 use occams_rpc_stream::server::{dispatch::*, task::*, *};
+use occams_rpc_tcp::TcpServer;
 use std::sync::Arc;
 
 use captains_log::filter::LogFilter;
@@ -15,7 +16,7 @@ where
 {
     let factory = Arc::new(FileServer::new(server_handle, config));
     let mut server = RpcServer::new(factory);
-    let local_addr = server.listen(addr)?;
+    let local_addr = server.listen::<TcpServer<crate::RT>>(addr)?;
     Ok((server, local_addr))
 }
 
@@ -39,15 +40,13 @@ where
     }
 }
 
-impl<H, FH> ServerFactory for FileServer<H, FH>
+impl<H, FH> ServerFact for FileServer<H, FH>
 where
     H: FnOnce(FileServerTask) -> FH + Send + Sync + 'static + Clone,
     FH: Future<Output = Result<(), ()>> + Send + 'static,
 {
     type Logger = Arc<LogFilter>;
 
-    type Transport = occams_rpc_tcp::TcpServer<Self>;
-
     type IO = crate::RT;
 
     type RespTask = FileServerTask;