Put RT as associate type of ClientTransport & ServerTransport

frostyplanet · frostyplanet · commit e71d15dff4e8 · 2026-03-16T14:10:04.000+08:00
Let FailoverPool store the runtime handle (so that we can reduce one
generic parameter)
diff --git a/src/client/mod.rs b/src/client/mod.rs
@@ -1,7 +1,6 @@
 pub mod task;
 use captains_log::filter::LogFilter;
 use crossfire::oneshot::oneshot;
-use orb::AsyncRuntime;
 pub use razor_rpc_macros::{endpoint_async, endpoint_client};
 pub use razor_stream::client::ClientCaller;
 pub use task::*;
@@ -33,16 +32,16 @@ impl<C: Codec> APIFact<C> {
         self.logger.set_level(level);
     }
 
-    pub fn new_conn_pool<P: ClientTransport, RT: AsyncRuntime + Clone>(
-        self: Arc<Self>, rt: &RT, addr: &str,
+    pub fn new_conn_pool<P: ClientTransport>(
+        self: Arc<Self>, rt: &P::RT, addr: &str,
     ) -> APIConnPool<C, P> {
-        ConnPool::<APIFact<C>, P>::new::<RT>(self.clone(), rt, addr, 0)
+        ConnPool::<APIFact<C>, P>::new(self.clone(), rt, addr, 0)
     }
 
-    pub fn new_failover<P: ClientTransport, RT: AsyncRuntime + Clone>(
-        self: Arc<Self>, rt: &RT, addrs: Vec<String>, stateless: bool, retry_limit: usize,
+    pub fn new_failover<P: ClientTransport>(
+        self: Arc<Self>, rt: &P::RT, addrs: Vec<String>, stateless: bool, retry_limit: usize,
     ) -> APIFailoverPool<C, P> {
-        FailoverPool::<APIFact<C>, P>::new::<RT>(self.clone(), rt, addrs, stateless, retry_limit, 0)
+        FailoverPool::<APIFact<C>, P>::new(self.clone(), rt, addrs, stateless, retry_limit, 0)
     }
 }
 
diff --git a/stream/src/client/failover.rs b/stream/src/client/failover.rs
@@ -11,7 +11,7 @@ use ahash::AHashMap;
 use arc_swap::ArcSwap;
 use captains_log::filter::LogFilter;
 use crossfire::{AsyncRx, MTx, SendError, mpsc};
-use orb::AsyncRuntime;
+use orb::prelude::AsyncExec;
 use std::fmt;
 use std::sync::{
     Arc, Weak,
@@ -57,6 +57,7 @@ where
     next_node: AtomicUsize,
     pool_channel_size: usize,
     facts: Arc<FailoverFacts<F>>,
+    rt: P::RT,
 }
 
 struct ClusterConfig<F, P>
@@ -76,8 +77,8 @@ where
     /// Initiate the pool with multiple addresses.
     /// When stateless == true, all addresses in the pool will be selected with equal chance (round-robin);
     /// When stateless == false, the leader address will always be picked unless error happens.
-    pub fn new<RT: AsyncRuntime + Clone>(
-        facts: Arc<F>, rt: &RT, addrs: Vec<String>, stateless: bool, retry_limit: usize,
+    pub fn new(
+        facts: Arc<F>, rt: &P::RT, addrs: Vec<String>, stateless: bool, retry_limit: usize,
         pool_channel_size: usize,
     ) -> Self {
         let (retry_tx, retry_rx) = mpsc::unbounded_async();
@@ -86,7 +87,7 @@ where
             Arc::new(FailoverFacts { retry_limit, retry_tx, logger: facts.new_logger(), facts });
         let mut pools = Vec::with_capacity(addrs.len());
         for addr in addrs.iter() {
-            let pool = ConnPool::new::<RT>(wrapped_facts.clone(), rt, addr, pool_channel_size);
+            let pool = ConnPool::new(wrapped_facts.clone(), rt, addr, pool_channel_size);
             pools.push(pool);
         }
         // NOTE: the ConnPool has cycle reference with FailoverPoolInner
@@ -97,6 +98,7 @@ where
             ver: AtomicU64::new(1),
             next_node: AtomicUsize::new(0),
             pool_channel_size,
+            rt: rt.clone(),
         });
         let weak_self = Arc::downgrade(&inner);
         rt.spawn_detach(async move {
@@ -174,7 +176,7 @@ where
         }
     }
 
-    pub fn update_addrs<RT: AsyncRuntime + Clone>(&self, addrs: Vec<String>, rt: &RT) {
+    pub fn update_addrs(&self, addrs: Vec<String>) {
         let inner = &self.inner;
         let old_pools = inner.pools.load_full();
         let mut new_pools: Vec<ConnPool<FailoverFacts<F>, P>> = Vec::with_capacity(addrs.len());
@@ -190,7 +192,7 @@ where
             } else {
                 // Create a new pool for the new address
                 let new_pool =
-                    ConnPool::new::<RT>(inner.facts.clone(), rt, &addr, inner.pool_channel_size);
+                    ConnPool::new(inner.facts.clone(), &inner.rt, &addr, inner.pool_channel_size);
                 new_pools.push(new_pool);
             }
         }
diff --git a/stream/src/client/mod.rs b/stream/src/client/mod.rs
@@ -3,6 +3,7 @@
 use crate::{Codec, error::RpcIntErr};
 use captains_log::filter::LogFilter;
 use crossfire::{AsyncRx, mpsc};
+use orb::AsyncRuntime;
 use std::future::Future;
 use std::sync::Arc;
 use std::time::{Duration, SystemTime, UNIX_EPOCH};
@@ -146,6 +147,8 @@ impl<C: ClientCallerBlocking + Send + Sync> ClientCallerBlocking for Arc<C> {
 ///
 /// - [razor-rpc-tcp](https://docs.rs/razor-rpc-tcp): For TCP and Unix socket
 pub trait ClientTransport: fmt::Debug + Send + Sized + 'static {
+    type RT: AsyncRuntime + Clone;
+
     /// How to establish an async connection.
     ///
     /// conn_id: used for log fmt, can by the same of addr.
diff --git a/stream/src/client/pool.rs b/stream/src/client/pool.rs
@@ -5,7 +5,7 @@ use crate::client::{
 use crate::error::RpcIntErr;
 use captains_log::filter::LogFilter;
 use crossfire::{MAsyncRx, MAsyncTx, MTx, RecvTimeoutError, mpmc};
-use orb::AsyncRuntime;
+use orb::prelude::{AsyncExec, AsyncTime};
 use std::fmt;
 use std::marker::PhantomData;
 use std::sync::Arc;
@@ -65,9 +65,7 @@ struct ConnPoolInner<F: ClientFacts, P: ClientTransport> {
 const ONE_SEC: Duration = Duration::from_secs(1);
 
 impl<F: ClientFacts, P: ClientTransport> ConnPool<F, P> {
-    pub fn new<RT: AsyncRuntime + Clone>(
-        facts: Arc<F>, rt: &RT, addr: &str, mut channel_size: usize,
-    ) -> Self {
+    pub fn new(facts: Arc<F>, rt: &P::RT, addr: &str, mut channel_size: usize) -> Self {
         let config = facts.get_config();
         if config.thresholds > 0 {
             if channel_size < config.thresholds {
@@ -91,7 +89,7 @@ impl<F: ClientFacts, P: ClientTransport> ConnPool<F, P> {
             _phan: Default::default(),
         });
         let s = Self { tx_async, tx, inner };
-        s.spawn::<RT>(rt);
+        s.spawn(rt);
         s
     }
 
@@ -118,7 +116,7 @@ impl<F: ClientFacts, P: ClientTransport> ConnPool<F, P> {
     /// by default there's one worker thread after initiation, but you can pre-spawn more thread if
     /// the connection is not enough to achieve desired throughput.
     #[inline]
-    pub fn spawn<RT: AsyncRuntime + Clone>(&self, rt: &RT) {
+    pub fn spawn(&self, rt: &P::RT) {
         let worker_id = self.inner.worker_count.fetch_add(1, Acquire);
         self.inner.clone().spawn_worker(rt, worker_id);
     }
@@ -155,7 +153,7 @@ impl<F: ClientFacts, P: ClientTransport> fmt::Display for ConnPoolInner<F, P> {
 }
 
 impl<F: ClientFacts, P: ClientTransport> ConnPoolInner<F, P> {
-    fn spawn_worker<RT: AsyncRuntime + Clone>(self: Arc<Self>, rt: &RT, worker_id: usize) {
+    fn spawn_worker(self: Arc<Self>, rt: &P::RT, worker_id: usize) {
         let _rt = rt.clone();
         rt.spawn_detach(async move {
             logger_trace!(&self.logger, "{} worker_id={} running", self, worker_id);
@@ -181,7 +179,7 @@ impl<F: ClientFacts, P: ClientTransport> ConnPoolInner<F, P> {
     }
 
     #[inline]
-    async fn connect<RT: AsyncRuntime>(&self, rt: &RT) -> Result<ClientStream<F, P>, RpcIntErr> {
+    async fn connect(&self, rt: &P::RT) -> Result<ClientStream<F, P>, RpcIntErr> {
         ClientStream::connect(self.facts.clone(), rt, &self.addr, &self.conn_id, None).await
     }
 
@@ -220,23 +218,27 @@ impl<F: ClientFacts, P: ClientTransport> ConnPoolInner<F, P> {
     /// connection attempts happens after we spawn.
     /// If the address is dead, the thread might exit after multiple attempts, and later re-spawn
     /// when the needs arrives.
-    async fn run<RT: AsyncRuntime + Clone>(self: &Arc<Self>, rt: RT, mut worker_id: usize) {
+    async fn run(self: &Arc<Self>, rt: P::RT, mut worker_id: usize) {
         'CONN_LOOP: loop {
-            match self.connect::<RT>(&rt).await {
+            match self.connect(&rt).await {
                 Ok(mut stream) => {
                     logger_trace!(self.logger, "{} worker={} connected", self, worker_id);
                     if worker_id == 0 {
                         // act as monitor
                         'MONITOR: loop {
                             if self.get_workers() > 1 {
-                                RT::sleep(ONE_SEC).await;
+                                <P::RT as AsyncTime>::sleep(ONE_SEC).await;
                                 if stream.ping().await.is_err() {
                                     self.set_err();
                                     // don't cleanup the channel unless only one worker left
                                     continue 'CONN_LOOP;
                                 }
                             } else {
-                                match self.rx.recv_with_timer(RT::sleep(ONE_SEC)).await {
+                                match self
+                                    .rx
+                                    .recv_with_timer(<P::RT as AsyncTime>::sleep(ONE_SEC))
+                                    .await
+                                {
                                     Err(RecvTimeoutError::Disconnected) => {
                                         return;
                                     }
@@ -258,13 +260,13 @@ impl<F: ClientFacts, P: ClientTransport> ConnPoolInner<F, P> {
                                             // there's might be a lag to connect,
                                             // so we are spawning identity with new worker,
                                             worker_id = 1;
-                                            self.clone().spawn_worker::<RT>(&rt, 0);
+                                            self.clone().spawn_worker(&rt, 0);
                                         }
                                         if stream.send_task(task, true).await.is_err() {
                                             self.set_err();
                                             if worker_id == 0 {
                                                 self.cleanup();
-                                                RT::sleep(ONE_SEC).await;
+                                                <P::RT as AsyncTime>::sleep(ONE_SEC).await;
                                                 continue 'CONN_LOOP;
                                             } else {
                                                 return;
@@ -297,7 +299,7 @@ impl<F: ClientFacts, P: ClientTransport> ConnPoolInner<F, P> {
                     self.set_err();
                     error!("connect failed to {}: {}", self.addr, e);
                     self.cleanup();
-                    RT::sleep(ONE_SEC).await;
+                    <P::RT as AsyncTime>::sleep(ONE_SEC).await;
                 }
             }
         }
diff --git a/stream/src/server/mod.rs b/stream/src/server/mod.rs
@@ -65,6 +65,8 @@ pub trait ServerFacts: Sync + Send + 'static + Sized {
 ///
 /// - [razor-rpc-tcp](https://docs.rs/razor-rpc-tcp): For TCP and Unix socket
 pub trait ServerTransport: Send + Sync + Sized + 'static + fmt::Debug {
+    type RT: AsyncRuntime + Clone;
+
     type Listener: AsyncListener;
 
     fn bind(addr: &str) -> impl Future<Output = io::Result<Self::Listener>> + Send;
diff --git a/test-suite/src/stream/client.rs b/test-suite/src/stream/client.rs
@@ -165,7 +165,7 @@ pub async fn init_failover_client(
     // since client may be drop by test logic, it's not allow
     // to drop a tokio runtime inside async code.
     let facts = MyClient::new(config);
-    FailoverPool::new::<crate::RT>(
+    FailoverPool::new(
         facts,
         rt,
         addrs,
diff --git a/transport/tcp/src/client.rs b/transport/tcp/src/client.rs
@@ -203,7 +203,9 @@ impl<RT: AsyncRuntime> TcpClient<RT> {
     }
 }
 
-impl<RT: AsyncRuntime> ClientTransport for TcpClient<RT> {
+impl<RT: AsyncRuntime + Clone> ClientTransport for TcpClient<RT> {
+    type RT = RT;
+
     async fn connect(addr: &str, conn_id: &str, config: &ClientConfig) -> Result<Self, RpcIntErr> {
         let stream: UnifyStream<RT> =
             match UnifyStream::<RT>::connect_timeout(addr, config.connect_timeout).await {
diff --git a/transport/tcp/src/server.rs b/transport/tcp/src/server.rs
@@ -62,7 +62,8 @@ impl<RT: AsyncRuntime> fmt::Debug for TcpServer<RT> {
     }
 }
 
-impl<RT: AsyncRuntime> ServerTransport for TcpServer<RT> {
+impl<RT: AsyncRuntime + Clone> ServerTransport for TcpServer<RT> {
+    type RT = RT;
     type Listener = UnifyListener<RT>;
 
     async fn bind(addr: &str) -> io::Result<Self::Listener> {

Original file line number	Diff line number	Diff line change
`@@ -203,7 +203,9 @@ impl<RT: AsyncRuntime> TcpClient<RT> {`
`203`	`203`	`}`
`204`	`204`	`}`
`205`	`205`
`206`		`-impl<RT: AsyncRuntime> ClientTransport for TcpClient<RT> {`
	`206`	`+impl<RT: AsyncRuntime + Clone> ClientTransport for TcpClient<RT> {`
	`207`	`+ type RT = RT;`
	`208`	`+`
`207`	`209`	`async fn connect(addr: &str, conn_id: &str, config: &ClientConfig) -> Result<Self, RpcIntErr> {`
`208`	`210`	`let stream: UnifyStream<RT> =`
`209`	`211`	`match UnifyStream::<RT>::connect_timeout(addr, config.connect_timeout).await {`
Original file line number	Diff line number	Diff line change
`@@ -62,7 +62,8 @@ impl<RT: AsyncRuntime> fmt::Debug for TcpServer<RT> {`
`62`	`62`	`}`
`63`	`63`	`}`
`64`	`64`
`65`		`-impl<RT: AsyncRuntime> ServerTransport for TcpServer<RT> {`
	`65`	`+impl<RT: AsyncRuntime + Clone> ServerTransport for TcpServer<RT> {`
	`66`	`+ type RT = RT;`
`66`	`67`	`type Listener = UnifyListener<RT>;`
`67`	`68`
`68`	`69`	`async fn bind(addr: &str) -> io::Result<Self::Listener> {`