karyon_jsonrpc/client/
mod.rs

1pub mod builder;
2mod message_dispatcher;
3mod multiplexed;
4mod subscriptions;
5
6#[cfg(feature = "http")]
7mod http;
8#[cfg(feature = "quic")]
9mod quic_stream;
10
11use std::{
12    marker::PhantomData,
13    sync::{atomic::AtomicBool, Arc},
14};
15
16use async_channel::{Receiver, Sender};
17use log::info;
18use serde::{de::DeserializeOwned, Serialize};
19use serde_json::json;
20
21use karyon_core::{async_runtime::Executor, async_util::TaskGroup, util::random_32};
22
23use karyon_net::Endpoint;
24
25#[cfg(feature = "tcp")]
26use karyon_net::tcp::TcpConfig;
27
28use crate::{
29    codec::{JsonCodec, JsonRpcCodec},
30    error::{Error, Result},
31    message::{self, SubscriptionID},
32};
33
34#[cfg(feature = "quic")]
35use karyon_net::quic::ClientQuicConfig;
36#[cfg(feature = "quic")]
37use parking_lot::Mutex;
38#[cfg(feature = "quic")]
39use std::collections::HashMap;
40
41#[cfg(feature = "ws")]
42use crate::codec::JsonRpcWsCodec;
43
44pub use builder::ClientBuilder;
45#[cfg(feature = "http")]
46pub use http::HttpConfig;
47pub use subscriptions::{NotificationDecoder, Subscription};
48
49use message_dispatcher::MessageDispatcher;
50use subscriptions::Subscriptions;
51
52type RequestID = u32;
53
54/// Bound on the WebSocket codec generic. With the `ws` feature it
55/// requires `JsonRpcWsCodec`; otherwise it accepts any clonable type
56/// (the codec is unused) so callers can pass `JsonCodec` unchanged.
57#[cfg(feature = "ws")]
58pub trait WsCodec: JsonRpcWsCodec {}
59#[cfg(feature = "ws")]
60impl<T: JsonRpcWsCodec> WsCodec for T {}
61
62#[cfg(not(feature = "ws"))]
63pub trait WsCodec: Clone + Send + Sync + 'static {}
64#[cfg(not(feature = "ws"))]
65impl<T: Clone + Send + Sync + 'static> WsCodec for T {}
66
67pub(crate) struct ClientConfig {
68    pub endpoint: Endpoint,
69    #[cfg(feature = "tcp")]
70    pub tcp_config: TcpConfig,
71    #[cfg(feature = "tls")]
72    pub tls_config: Option<karyon_net::tls::ClientTlsConfig>,
73    #[cfg(feature = "quic")]
74    pub quic_config: Option<ClientQuicConfig>,
75    #[cfg(feature = "http")]
76    pub http_config: HttpConfig,
77    pub timeout: Option<u64>,
78    pub subscription_buffer_size: usize,
79    /// Set via `ClientBuilder::with_notification_decoder`.
80    pub notification_decoder: NotificationDecoder,
81}
82
83pub(crate) enum ClientBackend {
84    /// One persistent message connection used by many concurrent calls
85    /// and subscriptions. Each request is tagged with a unique id; the
86    /// reader task multiplexes responses back to the right caller.
87    /// Used for TCP, TLS, WS, WSS, and Unix.
88    Multiplexed {
89        message_dispatcher: MessageDispatcher,
90        subscriptions: Arc<Subscriptions>,
91        // TODO what is the reason for this  ?
92        // AND why not using Async queue ?
93        send_chan: (Sender<serde_json::Value>, Receiver<serde_json::Value>),
94    },
95    #[cfg(feature = "quic")]
96    QuicStream {
97        quic_conn: Arc<karyon_net::quic::QuicConn>,
98        subscriptions: Arc<Subscriptions>,
99        /// Per-subscription unsubscribe signal: send the unsubscribe
100        /// JSON; the reader task forwards it and exits.
101        // TODO why not using async queue here ?
102        // TODO: move this map; entries leak when the stream ends.
103        unsub_chans: Mutex<HashMap<SubscriptionID, Sender<serde_json::Value>>>,
104    },
105    #[cfg(feature = "http")]
106    Http {
107        // Boxed because the HTTP backend (with hyper Client) is much
108        // larger than the other ClientBackend variants.
109        http_backend: Box<http::HttpClientBackend>,
110        #[cfg(feature = "http3")]
111        subscriptions: Arc<Subscriptions>,
112        /// Unsubscribe requests, sent on each subscription's own stream.
113        // TODO: move this map, same as `QuicStream::unsub_chans`.
114        #[cfg(feature = "http3")]
115        unsub_chans: Mutex<HashMap<SubscriptionID, Sender<serde_json::Value>>>,
116    },
117}
118
119/// An RPC client that connects to a JSON-RPC 2.0 server.
120///
121/// `B` is the byte-stream codec used for TCP/TLS/Unix/QUIC/HTTP.
122/// `W` is the WebSocket codec used for `ws://` / `wss://`.
123/// Both default to `JsonCodec`.
124pub struct Client<B = JsonCodec, W = JsonCodec> {
125    pub(crate) disconnect: AtomicBool,
126    pub(crate) task_group: Arc<TaskGroup>,
127    pub(crate) config: ClientConfig,
128    pub(crate) backend: ClientBackend,
129    /// Encodes and decodes HTTP bodies and HTTP/3 streams. The other
130    /// backends own their codec inside the connection.
131    #[cfg(feature = "http")]
132    pub(crate) codec: B,
133    _codecs: PhantomData<(B, W)>,
134}
135
136impl<B, W> Client<B, W>
137where
138    B: JsonRpcCodec,
139    W: WsCodec,
140{
141    /// Call a method, wait for response.
142    pub async fn call<T: Serialize + DeserializeOwned, V: DeserializeOwned>(
143        &self,
144        method: &str,
145        params: T,
146    ) -> Result<V> {
147        let response = match &self.backend {
148            #[cfg(feature = "http")]
149            ClientBackend::Http { http_backend, .. } => {
150                self.http_call(http_backend, method, params).await?
151            }
152            #[cfg(feature = "quic")]
153            ClientBackend::QuicStream { quic_conn, .. } => {
154                quic_stream::call(self, quic_conn, method, params).await?
155            }
156            _ => multiplexed::send_request(self, method, params).await?,
157        };
158
159        if let Some(error) = response.error {
160            return Err(Error::CallError(error.code, error.message));
161        }
162
163        match response.result {
164            Some(result) => Ok(serde_json::from_value::<V>(result)?),
165            None => Err(Error::InvalidMsg("Invalid response result".to_string())),
166        }
167    }
168
169    /// Subscribe to a method.
170    pub async fn subscribe<T: Serialize + DeserializeOwned>(
171        self: &Arc<Self>,
172        method: &str,
173        params: T,
174    ) -> Result<Arc<Subscription>> {
175        match &self.backend {
176            #[cfg(all(feature = "http", not(feature = "http3")))]
177            ClientBackend::Http { .. } => Err(Error::UnsupportedProtocol(
178                "Subscriptions not supported over HTTP/1-2".to_string(),
179            )),
180            #[cfg(feature = "http3")]
181            ClientBackend::Http { http_backend, .. } => {
182                self.h3_subscribe(http_backend, method, params).await
183            }
184            #[cfg(feature = "quic")]
185            ClientBackend::QuicStream {
186                quic_conn,
187                subscriptions,
188                unsub_chans,
189            } => {
190                quic_stream::subscribe(self, quic_conn, subscriptions, unsub_chans, method, params)
191                    .await
192            }
193            ClientBackend::Multiplexed { subscriptions, .. } => {
194                let response = multiplexed::send_request(self, method, params).await?;
195                if let Some(error) = response.error {
196                    return Err(Error::SubscribeError(error.code, error.message));
197                }
198                let sub_id = match response.result {
199                    Some(result) => serde_json::from_value::<SubscriptionID>(result)?,
200                    None => return Err(Error::InvalidMsg("Invalid subscription id".to_string())),
201                };
202                let sub = subscriptions.subscribe(sub_id).await;
203                Ok(sub)
204            }
205        }
206    }
207
208    /// Unsubscribe.
209    pub async fn unsubscribe(&self, method: &str, sub_id: SubscriptionID) -> Result<()> {
210        match &self.backend {
211            #[cfg(all(feature = "http", not(feature = "http3")))]
212            ClientBackend::Http { .. } => Err(Error::UnsupportedProtocol(
213                "Subscriptions not supported over HTTP/1-2".to_string(),
214            )),
215            #[cfg(feature = "http3")]
216            ClientBackend::Http { .. } => self.h3_unsubscribe(method, sub_id).await,
217            #[cfg(feature = "quic")]
218            ClientBackend::QuicStream {
219                subscriptions,
220                unsub_chans,
221                ..
222            } => {
223                // The QUIC subscription stream's reader and writer halves
224                // are owned by the spawned task. We can't write directly
225                // from here, so we hand the unsubscribe message to that
226                // task via a channel; the task forwards it on the wire
227                // and exits.
228                // Bind the removed sender in its own `let` so the parking_lot
229                // MutexGuard temporary doesn't live across the `.await`.
230                let ch = unsub_chans.lock().remove(&sub_id);
231                if let Some(ch) = ch {
232                    let request = message::Request {
233                        jsonrpc: message::JSONRPC_VERSION.to_string(),
234                        id: json!(random_32()),
235                        method: method.to_string(),
236                        params: Some(json!(sub_id)),
237                    };
238                    let _ = ch.send(serde_json::to_value(request)?).await;
239                }
240                subscriptions.unsubscribe(&sub_id).await;
241                Ok(())
242            }
243            ClientBackend::Multiplexed { subscriptions, .. } => {
244                let response = multiplexed::send_request(self, method, sub_id).await?;
245                subscriptions.unsubscribe(&sub_id).await;
246                match response.error {
247                    Some(error) => Err(Error::SubscribeError(error.code, error.message)),
248                    None => Ok(()),
249                }
250            }
251        }
252    }
253
254    /// Disconnect the client.
255    pub async fn stop(&self) {
256        self.task_group.cancel().await;
257    }
258
259    /// Pick the right backend based on endpoint kind, then build the
260    /// `Client`. Each backend module returns just the `ClientBackend`;
261    /// assembly happens here. Multiplexed mode also needs an extra
262    /// step to start its background reader/writer loop.
263    pub(super) async fn init(
264        config: ClientConfig,
265        byte_codec: B,
266        ws_codec: W,
267        executor: Option<Executor>,
268    ) -> Result<Arc<Self>> {
269        info!("Connecting to RPC endpoint: {}", config.endpoint);
270
271        // Build task_group early so backends (HTTP driver tasks) can
272        // spawn into the same group that Client::stop cancels.
273        let task_group = Arc::new(match executor {
274            Some(ex) => TaskGroup::with_executor(ex),
275            None => TaskGroup::new(),
276        });
277
278        #[cfg(feature = "http")]
279        if let Endpoint::Http(..) = &config.endpoint {
280            let backend = http::build_backend(&config, task_group.clone()).await?;
281            return Ok(Self::with_backend(config, task_group, backend, byte_codec));
282        }
283
284        #[cfg(feature = "quic")]
285        if let Endpoint::Quic(..) = &config.endpoint {
286            let backend = quic_stream::build_backend(&config).await?;
287            return Ok(Self::with_backend(config, task_group, backend, byte_codec));
288        }
289
290        #[cfg(feature = "ws")]
291        if matches!(&config.endpoint, Endpoint::Ws(..) | Endpoint::Wss(..)) {
292            let (backend, conn) = multiplexed::build_ws_backend(&config, ws_codec).await?;
293            let client = Self::with_backend(config, task_group, backend, byte_codec);
294            let (reader, writer) = conn.split();
295            multiplexed::start_io_loop(&client, reader, writer);
296            return Ok(client);
297        }
298
299        let (backend, conn) = multiplexed::build_byte_backend(&config, byte_codec.clone()).await?;
300        let client = Self::with_backend(config, task_group, backend, byte_codec);
301        let (reader, writer) = conn.split();
302        multiplexed::start_io_loop(&client, reader, writer);
303        Ok(client)
304    }
305
306    fn with_backend(
307        config: ClientConfig,
308        task_group: Arc<TaskGroup>,
309        backend: ClientBackend,
310        codec: B,
311    ) -> Arc<Self> {
312        // Only the HTTP backend reads the codec from the client.
313        #[cfg(not(feature = "http"))]
314        drop(codec);
315        Arc::new(Client {
316            disconnect: AtomicBool::new(false),
317            task_group,
318            config,
319            backend,
320            #[cfg(feature = "http")]
321            codec,
322            _codecs: PhantomData,
323        })
324    }
325}
karyon_jsonrpc/client/mod.rs

karyon_jsonrpc/client/
mod.rs
