razor_stream/client/
mod.rs

1//! The module contains traits defined for the client-side
2
3use crate::{Codec, error::RpcIntErr};
4use captains_log::filter::LogFilter;
5use crossfire::{AsyncRx, mpsc};
6use std::future::Future;
7use std::sync::Arc;
8use std::time::{Duration, SystemTime, UNIX_EPOCH};
9use std::{fmt, io};
10
11pub mod task;
12use task::{ClientTask, ClientTaskDone};
13
14pub mod stream;
15pub mod timer;
16use timer::ClientTaskTimer;
17
18mod pool;
19pub use pool::ClientPool;
20mod failover;
21pub use failover::FailoverPool;
22
23mod throttler;
24
25/// General config for client-side
26#[derive(Clone)]
27pub struct ClientConfig {
28    /// timeout of RpcTask waiting for response, in seconds.
29    pub task_timeout: usize,
30    /// socket read timeout
31    pub read_timeout: Duration,
32    /// Socket write timeout
33    pub write_timeout: Duration,
34    /// Socket idle time to be close. for connection pool.
35    pub idle_timeout: Duration,
36    /// connect timeout
37    pub connect_timeout: Duration,
38    /// How many async RpcTask in the queue, prevent overflow server capacity
39    pub thresholds: usize,
40    /// In bytes. when non-zero, overwrite the default DEFAULT_BUF_SIZE of transport
41    pub stream_buf_size: usize,
42}
43
44impl Default for ClientConfig {
45    fn default() -> Self {
46        Self {
47            task_timeout: 20,
48            read_timeout: Duration::from_secs(5),
49            write_timeout: Duration::from_secs(5),
50            idle_timeout: Duration::from_secs(120),
51            connect_timeout: Duration::from_secs(10),
52            thresholds: 128,
53            stream_buf_size: 0,
54        }
55    }
56}
57
58/// A trait implemented by the user for the client-side, to define the customizable plugin.
59pub trait ClientFacts: Send + Sync + Sized + 'static {
60    /// Define the codec to serialization and deserialization
61    ///
62    /// Refers to [Codec]
63    type Codec: Codec;
64
65    /// Define the RPC task from client-side
66    ///
67    /// Either one ClientTask or an enum of multiple ClientTask.
68    /// If you have multiple task type, recommend to use the `enum_dispatch` crate.
69    ///
70    /// You can use macro [client_task_enum](crate::client::task::client_task_enum) and [client_task](crate::client::task::client_task) on task type
71    type Task: ClientTask;
72
73    /// You should keep ClientConfig inside, get_config() will return the reference.
74    fn get_config(&self) -> &ClientConfig;
75
76    /// Construct a [captains_log::filter::Filter](https://docs.rs/captains-log/latest/captains_log/filter/trait.Filter.html) to oganize log of a client
77    ///
78    /// TODO: Fix the logger interface
79    fn new_logger(&self) -> Arc<LogFilter>;
80
81    /// How to deal with error
82    ///
83    /// The FailoverPool will overwrite this to implement retry logic
84    #[inline(always)]
85    fn error_handle(&self, task: Self::Task) {
86        task.done();
87    }
88
89    /// You can overwrite this to assign a client_id
90    #[inline(always)]
91    fn get_client_id(&self) -> u64 {
92        0
93    }
94
95    /// NOTE: you may overwrite this to use coarstime or quanta
96    #[inline]
97    fn get_timestamp(&self) -> u64 {
98        SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_secs()
99    }
100}
101
102/// A trait to support sending request task in async text, for all router and connection pool
103/// implementations
104pub trait ClientCaller: Send {
105    type Facts: ClientFacts;
106    fn send_req(&self, task: <Self::Facts as ClientFacts>::Task)
107    -> impl Future<Output = ()> + Send;
108}
109
110/// A trait to support sending request task in blocking text, for all router and connection pool
111/// implementations
112pub trait ClientCallerBlocking: Send {
113    type Facts: ClientFacts;
114    fn send_req_blocking(&self, task: <Self::Facts as ClientFacts>::Task);
115}
116
117impl<C: ClientCaller + Send + Sync> ClientCaller for Arc<C> {
118    type Facts = C::Facts;
119    #[inline(always)]
120    async fn send_req(&self, task: <Self::Facts as ClientFacts>::Task) {
121        self.as_ref().send_req(task).await
122    }
123}
124
125impl<C: ClientCallerBlocking + Send + Sync> ClientCallerBlocking for Arc<C> {
126    type Facts = C::Facts;
127
128    #[inline(always)]
129    fn send_req_blocking(&self, task: <Self::Facts as ClientFacts>::Task) {
130        self.as_ref().send_req_blocking(task);
131    }
132}
133
134/// This trait is for client-side transport layer protocol.
135///
136/// The implementation can be found on:
137///
138/// - [razor-rpc-tcp](https://docs.rs/razor-rpc-tcp): For TCP and Unix socket
139pub trait ClientTransport: fmt::Debug + Send + Sized + 'static {
140    /// How to establish an async connection.
141    ///
142    /// conn_id: used for log fmt, can by the same of addr.
143    fn connect(
144        addr: &str, conn_id: &str, config: &ClientConfig,
145    ) -> impl Future<Output = Result<Self, RpcIntErr>> + Send;
146
147    /// Shutdown the write direction of the connection
148    fn close_conn<F: ClientFacts>(&self, logger: &LogFilter) -> impl Future<Output = ()> + Send;
149
150    /// Flush the request for the socket writer, if the transport has buffering logic
151    fn flush_req<F: ClientFacts>(
152        &self, logger: &LogFilter,
153    ) -> impl Future<Output = io::Result<()>> + Send;
154
155    /// Write out the encoded request task
156    fn write_req<'a, F: ClientFacts>(
157        &'a self, logger: &LogFilter, buf: &'a [u8], blob: Option<&'a [u8]>, need_flush: bool,
158    ) -> impl Future<Output = io::Result<()>> + Send;
159
160    /// Read the response and decode it from the socket, find and notify the registered ClientTask
161    fn read_resp<F: ClientFacts>(
162        &self, facts: &F, logger: &LogFilter, codec: &F::Codec,
163        close_ch: Option<&mut AsyncRx<mpsc::Null>>, task_reg: &mut ClientTaskTimer<F>,
164    ) -> impl std::future::Future<Output = Result<bool, RpcIntErr>> + Send;
165}
166
167/// An example ClientFacts for general use
168pub struct ClientDefault<T: ClientTask, C: Codec> {
169    pub logger: Arc<LogFilter>,
170    config: ClientConfig,
171    _phan: std::marker::PhantomData<fn(&C, &T)>,
172}
173
174impl<T: ClientTask, C: Codec> ClientDefault<T, C> {
175    pub fn new(config: ClientConfig) -> Arc<Self> {
176        Arc::new(Self { logger: Arc::new(LogFilter::new()), config, _phan: Default::default() })
177    }
178
179    #[inline]
180    pub fn set_log_level(&self, level: log::Level) {
181        self.logger.set_level(level);
182    }
183}
184
185impl<T: ClientTask, C: Codec> ClientFacts for ClientDefault<T, C> {
186    type Codec = C;
187    type Task = T;
188
189    #[inline]
190    fn new_logger(&self) -> Arc<LogFilter> {
191        self.logger.clone()
192    }
193
194    #[inline]
195    fn get_config(&self) -> &ClientConfig {
196        &self.config
197    }
198}
razor_stream/client/mod.rs

razor_stream/client/
mod.rs
