karyon_jsonrpc/server/
builder.rs

1use std::{collections::HashMap, sync::Arc, time::Duration};
2
3use karyon_core::async_runtime::Executor;
4
5use karyon_net::ToEndpoint;
6
7#[cfg(any(feature = "tcp", feature = "tls", feature = "quic"))]
8use karyon_net::Endpoint;
9
10#[cfg(feature = "tcp")]
11use karyon_net::tcp::TcpConfig;
12
13#[cfg(feature = "quic")]
14use karyon_net::quic::ServerQuicConfig;
15
16use crate::{
17    codec::{JsonCodec, JsonRpcCodec},
18    error::Result,
19    message::{Notification, NotificationResult, JSONRPC_VERSION},
20    server::channel::NewNotification,
21    server::PubSubRPCService,
22    server::RPCService,
23    server::WsCodec,
24};
25
26#[cfg(any(feature = "tcp", feature = "tls", feature = "quic"))]
27use crate::error::Error;
28
29use super::{Server, ServerConfig};
30
31#[cfg(any(feature = "tls", feature = "ws"))]
32const DEFAULT_HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(10);
33
34/// Builder for constructing an RPC [`Server`].
35///
36/// # Example
37///
38/// ```no_run
39/// use std::sync::Arc;
40/// use serde_json::Value;
41/// use karyon_jsonrpc::{error::RPCError, rpc_impl, server::ServerBuilder};
42///
43/// struct Ping {}
44///
45/// #[rpc_impl]
46/// impl Ping {
47///     async fn ping(&self, _params: Value) -> Result<Value, RPCError> {
48///         Ok(serde_json::json!("pong"))
49///     }
50/// }
51///
52/// async {
53///     let server = ServerBuilder::new("tcp://127.0.0.1:60000")
54///         .expect("create builder")
55///         .service(Arc::new(Ping {}))
56///         .build().await
57///         .expect("build server");
58///
59///     server.start_block().await.expect("run server");
60/// };
61/// ```
62pub struct ServerBuilder<B, W = JsonCodec> {
63    config: ServerConfig,
64    byte_codec: B,
65    ws_codec: W,
66    executor: Option<Executor>,
67}
68
69impl<B, W> ServerBuilder<B, W>
70where
71    B: JsonRpcCodec,
72    W: WsCodec,
73{
74    /// Add an RPC service.
75    pub fn service(mut self, service: Arc<dyn RPCService>) -> Self {
76        self.config.services.insert(service.name(), service);
77        self
78    }
79
80    /// Add a PubSub RPC service.
81    pub fn pubsub_service(mut self, service: Arc<dyn PubSubRPCService>) -> Self {
82        self.config.pubsub_services.insert(service.name(), service);
83        self
84    }
85
86    /// Set TCP config.
87    #[cfg(feature = "tcp")]
88    pub fn tcp_config(mut self, config: TcpConfig) -> Result<Self> {
89        match self.config.endpoint {
90            Endpoint::Tcp(..) | Endpoint::Tls(..) | Endpoint::Ws(..) | Endpoint::Wss(..) => {
91                self.config.tcp_config = config;
92                Ok(self)
93            }
94            _ => Err(Error::UnsupportedProtocol(self.config.endpoint.to_string())),
95        }
96    }
97
98    /// Set TLS config.
99    #[cfg(feature = "tls")]
100    pub fn tls_config(mut self, config: karyon_net::tls::ServerTlsConfig) -> Result<Self> {
101        match self.config.endpoint {
102            Endpoint::Tls(..) | Endpoint::Wss(..) => {
103                self.config.tls_config = Some(config);
104                Ok(self)
105            }
106            _ => Err(Error::UnsupportedProtocol(format!(
107                "Invalid tls config for endpoint: {}",
108                self.config.endpoint
109            ))),
110        }
111    }
112
113    /// Set QUIC config.
114    #[cfg(feature = "quic")]
115    pub fn quic_config(mut self, config: ServerQuicConfig) -> Result<Self> {
116        match self.config.endpoint {
117            Endpoint::Quic(..) => {
118                self.config.quic_config = Some(config);
119                Ok(self)
120            }
121            #[cfg(feature = "http3")]
122            Endpoint::Http(..) => {
123                self.config.quic_config = Some(config);
124                Ok(self)
125            }
126            _ => Err(Error::UnsupportedProtocol(format!(
127                "Invalid quic config for endpoint: {}",
128                self.config.endpoint
129            ))),
130        }
131    }
132
133    /// Set an executor.
134    pub fn with_executor(mut self, ex: Executor) -> Self {
135        self.executor = Some(ex);
136        self
137    }
138
139    /// Drop a connection that sends nothing for this long. Off by
140    /// default, because pubsub subscribers idle by design.
141    pub fn read_timeout(mut self, duration: Duration) -> Self {
142        self.config.read_timeout = Some(duration);
143        self
144    }
145
146    /// Drop a client that does not finish the TLS and/or WebSocket
147    /// handshake within this duration. Defaults to 10 seconds.
148    #[cfg(any(feature = "tls", feature = "ws"))]
149    pub fn handshake_timeout(mut self, duration: Duration) -> Self {
150        self.config.handshake_timeout = duration;
151        self
152    }
153
154    /// Set a custom notification encoder.
155    pub fn with_notification_encoder(
156        mut self,
157        encoder: fn(NewNotification) -> Notification,
158    ) -> Self {
159        self.config.notification_encoder = encoder;
160        self
161    }
162
163    /// Override the WebSocket codec. Only meaningful for `ws://` /
164    /// `wss://` endpoints.
165    #[cfg(feature = "ws")]
166    pub fn with_ws_codec<W2: WsCodec>(self, ws_codec: W2) -> ServerBuilder<B, W2> {
167        ServerBuilder {
168            config: self.config,
169            byte_codec: self.byte_codec,
170            ws_codec,
171            executor: self.executor,
172        }
173    }
174
175    /// Build the server.
176    pub async fn build(self) -> Result<Arc<Server>> {
177        Server::init(self.config, self.executor, self.byte_codec, self.ws_codec).await
178    }
179}
180
181impl<B: JsonRpcCodec> ServerBuilder<B, JsonCodec> {
182    /// Create a builder with a custom byte-stream codec. The default
183    /// `JsonCodec` is used for `ws://` / `wss://` endpoints; override
184    /// with `with_ws_codec`.
185    pub fn new_with_codec(
186        endpoint: impl ToEndpoint,
187        codec: B,
188    ) -> Result<ServerBuilder<B, JsonCodec>> {
189        let endpoint = endpoint.to_endpoint()?;
190        Ok(ServerBuilder {
191            config: ServerConfig {
192                endpoint,
193                services: HashMap::new(),
194                pubsub_services: HashMap::new(),
195                #[cfg(feature = "tcp")]
196                tcp_config: Default::default(),
197                #[cfg(feature = "tls")]
198                tls_config: None,
199                #[cfg(feature = "quic")]
200                quic_config: None,
201                notification_encoder: default_notification_encoder,
202                read_timeout: None,
203                #[cfg(any(feature = "tls", feature = "ws"))]
204                handshake_timeout: DEFAULT_HANDSHAKE_TIMEOUT,
205            },
206            byte_codec: codec,
207            ws_codec: JsonCodec::default(),
208            executor: None,
209        })
210    }
211}
212
213impl ServerBuilder<JsonCodec, JsonCodec> {
214    /// Create a builder with the default JSON codec.
215    pub fn new(endpoint: impl ToEndpoint) -> Result<ServerBuilder<JsonCodec, JsonCodec>> {
216        Self::new_with_codec(endpoint, JsonCodec::default())
217    }
218}
219
220fn default_notification_encoder(nt: NewNotification) -> Notification {
221    let params = Some(serde_json::json!(NotificationResult {
222        subscription: nt.sub_id,
223        result: Some(nt.result),
224    }));
225
226    Notification {
227        jsonrpc: JSONRPC_VERSION.to_string(),
228        method: nt.method,
229        params,
230    }
231}
karyon_jsonrpc/server/builder.rs

karyon_jsonrpc/server/
builder.rs
