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

karyon_jsonrpc/server/
builder.rs
