workflow_rpc/client/
mod.rs

1//!
2//! RPC client (operates uniformly in native and WASM-browser environments).
3//!
4//! # TLS crypto provider (native)
5//!
6//! Secure (`wss://`) RPC connections on native targets go through
7//! `workflow-websocket` → `tungstenite` → [`rustls`](https://docs.rs/rustls),
8//! which (since 0.23) requires a process-level crypto provider to be installed
9//! before the first connection — typically by the application or a higher-level
10//! SDK at startup:
11//!
12//! ```ignore
13//! rustls::crypto::ring::default_provider().install_default().unwrap();
14//! ```
15//!
16//! If exactly one rustls provider is compiled into the binary, it is selected
17//! automatically and no explicit install is needed; an install is only required
18//! when none — or more than one — provider is present. In the browser/WASM
19//! environment the host `WebSocket` handles TLS, so no provider is required.
20//!
21
22pub mod error;
23mod interface;
24pub mod prelude;
25mod protocol;
26pub mod result;
27pub use crate::client::error::Error;
28pub use crate::client::result::Result;
29
30use crate::imports::*;
31use futures_util::select_biased;
32pub use interface::{Interface, Notification};
33use protocol::ProtocolHandler;
34pub use protocol::{BorshProtocol, JsonProtocol};
35use std::fmt::Debug;
36use std::str::FromStr;
37use workflow_core::{channel::Multiplexer, task::yield_now};
38pub use workflow_websocket::client::{
39    ConnectOptions, ConnectResult, ConnectStrategy, Resolver, ResolverResult, WebSocketConfig,
40    WebSocketError,
41};
42
43#[cfg(feature = "wasm32-sdk")]
44pub use workflow_websocket::client::options::IConnectOptions;
45
46///
47/// notification!() macro for declaration of RPC notification handlers
48///
49/// This macro simplifies creation of async notification handler
50/// closures supplied to the RPC notification interface. An
51/// async notification closure requires to be *Box*ed
52/// and its result must be *Pin*ned, resulting in the following
53/// syntax:
54///
55/// ```ignore
56///
57/// interface.notification(Box::new(Notification::new(|msg: MyMsg|
58///     Box::pin(
59///         async move {
60///             // ...
61///             Ok(())
62///         }
63///     )
64/// )))
65///
66/// ```
67///
68/// The notification macro adds the required Box and Pin syntax,
69/// simplifying the declaration as follows:
70///
71/// ```ignore
72/// interface.notification(notification!(|msg: MyMsg| async move {
73///     // ...
74///     Ok(())
75/// }))
76/// ```
77///
78pub use workflow_rpc_macros::client_notification as notification;
79
80/// Client connection lifecycle event broadcast through the control multiplexer.
81#[derive(Debug, Clone, Copy, Eq, PartialEq, Hash)]
82pub enum Ctl {
83    /// The client has established a connection to the server.
84    Connect,
85    /// The client has lost or closed its connection to the server.
86    Disconnect,
87}
88
89impl std::fmt::Display for Ctl {
90    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
91        match self {
92            Ctl::Connect => write!(f, "connect"),
93            Ctl::Disconnect => write!(f, "disconnect"),
94        }
95    }
96}
97
98impl FromStr for Ctl {
99    type Err = Error;
100
101    fn from_str(s: &str) -> std::result::Result<Self, Self::Err> {
102        match s {
103            "connect" => Ok(Ctl::Connect),
104            "disconnect" => Ok(Ctl::Disconnect),
105            _ => Err(Error::InvalidEvent(s.to_string())),
106        }
107    }
108}
109
110/// Handler invoked with the raw payload of an inbound server notification.
111#[async_trait]
112pub trait NotificationHandler: Send + Sync + 'static {
113    /// Process a notification carrying the given raw `data` payload.
114    async fn handle_notification(&self, data: &[u8]) -> Result<()>;
115}
116
117/// Configuration options used when constructing an [`RpcClient`].
118#[derive(Default)]
119pub struct Options<'url> {
120    /// Optional multiplexer that receives [`Ctl`] connect/disconnect events.
121    pub ctl_multiplexer: Option<Multiplexer<Ctl>>,
122    /// Optional WebSocket endpoint URL the client should connect to.
123    pub url: Option<&'url str>,
124}
125
126impl<'url> Options<'url> {
127    /// Create a new set of default client options.
128    pub fn new() -> Self {
129        Self::default()
130    }
131
132    /// Set the WebSocket endpoint URL, returning the updated options.
133    pub fn with_url(mut self, url: &'url str) -> Self {
134        self.url = Some(url);
135        self
136    }
137
138    /// Set the [`Ctl`] event multiplexer, returning the updated options.
139    pub fn with_ctl_multiplexer(mut self, ctl_multiplexer: Multiplexer<Ctl>) -> Self {
140        self.ctl_multiplexer = Some(ctl_multiplexer);
141        self
142    }
143}
144
145struct Inner<Ops> {
146    ws: Arc<WebSocket>,
147    is_running: AtomicBool,
148    is_connected: AtomicBool,
149    receiver_is_running: AtomicBool,
150    timeout_is_running: AtomicBool,
151    receiver_shutdown: DuplexChannel,
152    timeout_shutdown: DuplexChannel,
153    timeout_timer_interval: AtomicU64,
154    timeout_duration: AtomicU64,
155    ctl_multiplexer: Option<Multiplexer<Ctl>>,
156    protocol: Arc<dyn ProtocolHandler<Ops>>,
157}
158
159impl<Ops> Inner<Ops>
160where
161    Ops: OpsT,
162{
163    fn new<T>(
164        ws: Arc<WebSocket>,
165        protocol: Arc<dyn ProtocolHandler<Ops>>,
166        options: Options,
167    ) -> Result<Self>
168    where
169        T: ProtocolHandler<Ops> + Send + Sync + 'static,
170    {
171        let inner = Inner {
172            ws,
173            is_running: AtomicBool::new(false),
174            is_connected: AtomicBool::new(false),
175            receiver_is_running: AtomicBool::new(false),
176            receiver_shutdown: DuplexChannel::oneshot(),
177            timeout_is_running: AtomicBool::new(false),
178            timeout_shutdown: DuplexChannel::oneshot(),
179            timeout_duration: AtomicU64::new(60_000),
180            timeout_timer_interval: AtomicU64::new(5_000),
181            ctl_multiplexer: options.ctl_multiplexer,
182            protocol,
183        };
184
185        Ok(inner)
186    }
187
188    #[inline]
189    pub fn is_running(&self) -> bool {
190        self.is_running.load(Ordering::SeqCst)
191    }
192
193    pub fn start(self: &Arc<Self>) -> Result<()> {
194        if !self.is_running.load(Ordering::Relaxed) {
195            self.is_running.store(true, Ordering::SeqCst);
196            self.clone().timeout_task();
197            self.clone().receiver_task();
198        } else {
199            log_warn!("wRPC services are already running: rpc::start() was called multiple times");
200        }
201        Ok(())
202    }
203
204    pub async fn shutdown(self: &Arc<Self>) -> Result<()> {
205        self.ws.disconnect().await?;
206        yield_now().await;
207        if self.is_running.load(Ordering::Relaxed) {
208            self.stop_timeout().await?;
209            self.stop_receiver().await?;
210            self.is_running.store(false, Ordering::SeqCst);
211        }
212        Ok(())
213    }
214
215    fn timeout_task(self: Arc<Self>) {
216        self.timeout_is_running.store(true, Ordering::SeqCst);
217        workflow_core::task::spawn(async move {
218            'outer: loop {
219                let timeout_timer_interval =
220                    Duration::from_millis(self.timeout_timer_interval.load(Ordering::SeqCst));
221                select_biased! {
222                    _ = workflow_core::task::sleep(timeout_timer_interval).fuse() => {
223                        let timeout = Duration::from_millis(self.timeout_duration.load(Ordering::Relaxed));
224                        self.protocol.handle_timeout(timeout).await;
225                    },
226                    _ = self.timeout_shutdown.request.receiver.recv().fuse() => {
227                        break 'outer;
228                    },
229                }
230            }
231
232            self.timeout_is_running.store(false, Ordering::SeqCst);
233            self.timeout_shutdown.response.sender.send(()).await.unwrap_or_else(|err|
234                log_error!("wRPC client - unable to signal shutdown completion for timeout task: `{err}`"));
235        });
236    }
237
238    fn receiver_task(self: Arc<Self>) {
239        self.receiver_is_running.store(true, Ordering::SeqCst);
240        let receiver_rx = self.ws.receiver_rx().clone();
241        workflow_core::task::spawn(async move {
242            'outer: loop {
243                select_biased! {
244                    msg = receiver_rx.recv().fuse() => {
245                        match msg {
246                            Ok(msg) => {
247                                match msg {
248                                    WebSocketMessage::Binary(_) | WebSocketMessage::Text(_) => {
249                                        self.protocol.handle_message(msg).await
250                                        .unwrap_or_else(|err|log_trace!("wRPC error: `{err}`"));
251                                    }
252                                    WebSocketMessage::Open => {
253                                        self.is_connected.store(true, Ordering::SeqCst);
254                                        if let Some(ctl_channel) = &self.ctl_multiplexer {
255                                            ctl_channel.try_broadcast(Ctl::Connect).expect("ctl_channel.try_broadcast(Ctl::Connect)");
256                                        }
257                                    }
258                                    WebSocketMessage::Close => {
259                                        self.is_connected.store(false, Ordering::SeqCst);
260
261                                        self.protocol.handle_disconnect().await.unwrap_or_else(|err|{
262                                            log_error!("wRPC error during protocol disconnect: {err}");
263                                        });
264
265                                        if let Some(ctl_channel) = &self.ctl_multiplexer {
266                                            ctl_channel.try_broadcast(Ctl::Disconnect).expect("ctl_channel.try_broadcast(Ctl::Disconnect)");
267                                        }
268                                    }
269                                }
270                            },
271                            Err(err) => {
272                                log_error!("wRPC client receiver channel error: {err}");
273                                break 'outer;
274                            }
275                        }
276                    },
277                    _ = self.receiver_shutdown.request.receiver.recv().fuse() => {
278                        break 'outer;
279                    },
280
281                }
282            }
283
284            self.receiver_is_running.store(false, Ordering::SeqCst);
285            self.receiver_shutdown.response.sender.send(()).await.unwrap_or_else(|err|
286                log_error!("wRPC client - unable to signal shutdown completion for receiver task: `{err}`")
287            );
288        });
289    }
290
291    async fn stop_receiver(&self) -> Result<()> {
292        if !self.receiver_is_running.load(Ordering::SeqCst) {
293            return Ok(());
294        }
295
296        self.receiver_shutdown
297            .signal(())
298            .await
299            .unwrap_or_else(|err| {
300                log_error!("wRPC client unable to signal receiver shutdown: `{err}`")
301            });
302
303        Ok(())
304    }
305
306    async fn stop_timeout(&self) -> Result<()> {
307        if !self.timeout_is_running.load(Ordering::SeqCst) {
308            return Ok(());
309        }
310
311        self.timeout_shutdown
312            .signal(())
313            .await
314            .unwrap_or_else(|err| {
315                log_error!("wRPC client unable to signal timeout shutdown: `{err}`")
316            });
317
318        Ok(())
319    }
320}
321
322#[derive(Clone)]
323enum Protocol<Ops, Id>
324where
325    Ops: OpsT,
326    Id: IdT,
327{
328    Borsh(Arc<BorshProtocol<Ops, Id>>),
329    Json(Arc<JsonProtocol<Ops, Id>>),
330}
331
332impl<Ops, Id> From<Arc<dyn ProtocolHandler<Ops>>> for Protocol<Ops, Id>
333where
334    Ops: OpsT,
335    Id: IdT,
336{
337    fn from(protocol: Arc<dyn ProtocolHandler<Ops>>) -> Self {
338        match protocol.clone().downcast_arc::<BorshProtocol<Ops, Id>>() {
339            Ok(protocol) => Protocol::Borsh(protocol),
340            _ => match protocol.clone().downcast_arc::<JsonProtocol<Ops, Id>>() {
341                Ok(protocol) => Protocol::Json(protocol),
342                _ => {
343                    panic!()
344                }
345            },
346        }
347    }
348}
349
350/// wRPC client capable of issuing requests and notifications and dispatching
351/// server-side notifications over a WebSocket connection using either the
352/// Borsh or JSON protocol. `Ops` is the application operation enum and `Id`
353/// is the message identifier type.
354#[derive(Clone)]
355pub struct RpcClient<Ops, Id = Id64>
356where
357    Ops: OpsT,
358    Id: IdT,
359{
360    inner: Arc<Inner<Ops>>,
361    protocol: Protocol<Ops, Id>,
362    ops: PhantomData<Ops>,
363    id: PhantomData<Id>,
364}
365
366impl<Ops, Id> RpcClient<Ops, Id>
367where
368    Ops: OpsT,
369    Id: IdT,
370{
371    ///
372    /// Create new wRPC client connecting to the supplied URL
373    ///
374    /// This function accepts the [`Encoding`] enum argument denoting the underlying
375    /// protocol that will be used by the client. Current variants supported
376    /// are:
377    ///
378    /// - [`Encoding::Borsh`]
379    /// - [`Encoding::SerdeJson`]
380    ///
381    ///
382    pub fn new_with_encoding(
383        encoding: Encoding,
384        interface: Option<Arc<Interface<Ops>>>,
385        options: Options,
386        config: Option<WebSocketConfig>,
387    ) -> Result<RpcClient<Ops, Id>> {
388        match encoding {
389            Encoding::Borsh => Self::new::<BorshProtocol<Ops, Id>>(interface, options, config),
390            Encoding::SerdeJson => Self::new::<JsonProtocol<Ops, Id>>(interface, options, config),
391        }
392    }
393
394    ///
395    /// Create new wRPC client connecting to the supplied URL.
396    ///
397    /// This function accepts a generic denoting the underlying
398    /// protocol that will be used by the client. Current protocols
399    /// supported are:
400    ///
401    /// - [`BorshProtocol`]
402    /// - [`JsonProtocol`]
403    ///
404    ///
405    pub fn new<T>(
406        interface: Option<Arc<Interface<Ops>>>,
407        options: Options,
408        config: Option<WebSocketConfig>,
409    ) -> Result<RpcClient<Ops, Id>>
410    where
411        T: ProtocolHandler<Ops> + Send + Sync + 'static,
412    {
413        let url = options.url.map(sanitize_url).transpose()?;
414
415        let ws = Arc::new(WebSocket::new(url.as_deref(), config)?);
416        let protocol: Arc<dyn ProtocolHandler<Ops>> = Arc::new(T::new(ws.clone(), interface));
417        let inner = Arc::new(Inner::new::<T>(ws, protocol.clone(), options)?);
418
419        let client = RpcClient::<Ops, Id> {
420            inner,
421            protocol: protocol.into(),
422            ops: PhantomData,
423            id: PhantomData,
424        };
425
426        Ok(client)
427    }
428
429    /// Connect to the target wRPC endpoint (websocket address)
430    pub async fn connect(&self, options: ConnectOptions) -> ConnectResult<Error> {
431        if !self.inner.is_running() {
432            self.inner.start()?;
433        }
434        Ok(self.inner.ws.connect(options).await?)
435    }
436
437    /// Stop wRPC client services
438    pub async fn shutdown(&self) -> Result<()> {
439        self.inner.shutdown().await?;
440        Ok(())
441    }
442
443    /// Access the optional connect/disconnect [`Ctl`] event multiplexer
444    /// configured for this client, if any.
445    pub fn ctl_multiplexer(&self) -> &Option<Multiplexer<Ctl>> {
446        &self.inner.ctl_multiplexer
447    }
448
449    /// Test if the underlying WebSocket is currently open
450    pub fn is_connected(&self) -> bool {
451        self.inner.ws.is_connected()
452    }
453
454    /// Obtain the current URL of the underlying WebSocket
455    pub fn url(&self) -> Option<String> {
456        self.inner.ws.url()
457    }
458
459    /// Change the URL of the underlying WebSocket
460    /// (applicable only to the next connection).
461    /// Alternatively, the new URL can be supplied
462    /// in the `connect()` method using [`ConnectOptions`].
463    pub fn set_url(&self, url: &str) -> Result<()> {
464        self.inner.ws.set_url(url);
465        Ok(())
466    }
467
468    /// Change the configuration of the underlying WebSocket.
469    /// This method can be used to alter the configuration
470    /// for the next connection.
471    pub fn configure(&self, config: WebSocketConfig) {
472        self.inner.ws.configure(config);
473    }
474
475    ///
476    /// Issue an async Notification to the server (no response is expected)
477    ///
478    /// Following are the trait requirements on the arguments:
479    /// - `Ops`: [`OpsT`]
480    /// - `Msg`: [`MsgT`]
481    ///
482    pub async fn notify<Msg>(&self, op: Ops, payload: Msg) -> Result<()>
483    where
484        Msg: BorshSerialize + Serialize + Send + Sync + 'static,
485    {
486        if !self.is_connected() {
487            return Err(WebSocketError::NotConnected.into());
488        }
489
490        match &self.protocol {
491            Protocol::Borsh(protocol) => {
492                protocol.notify(op, payload).await?;
493            }
494            Protocol::Json(protocol) => {
495                protocol.notify(op, payload).await?;
496            }
497        }
498
499        Ok(())
500    }
501
502    ///
503    /// Issue an async wRPC call and wait for response.
504    ///
505    /// Following are the trait requirements on the arguments:
506    /// - `Ops`: [`OpsT`]
507    /// - `Req`: [`MsgT`]
508    /// - `Resp`: [`MsgT`]
509    ///
510    pub async fn call<Req, Resp>(&self, op: Ops, req: Req) -> Result<Resp>
511    where
512        Req: MsgT,
513        Resp: MsgT,
514    {
515        if !self.is_connected() {
516            return Err(WebSocketError::NotConnected.into());
517        }
518
519        match &self.protocol {
520            Protocol::Borsh(protocol) => Ok(protocol.request(op, req).await?),
521            Protocol::Json(protocol) => Ok(protocol.request(op, req).await?),
522        }
523    }
524
525    /// Triggers a disconnection on the underlying WebSocket.
526    /// This is intended for debug purposes only.
527    /// Can be used to test application reconnection logic.
528    pub fn trigger_abort(&self) -> Result<()> {
529        Ok(self.inner.ws.trigger_abort()?)
530    }
531}
532
533fn sanitize_url(url: &str) -> Result<String> {
534    let url = url
535        .replace("wrpc://", "ws://")
536        .replace("wrpcs://", "wss://");
537    Ok(url)
538}
workflow_rpc/client/mod.rs

workflow_rpc/client/
mod.rs
