workflow_rpc/client/interface/
mod.rs

1pub mod notification;
2use crate::imports::*;
3pub use notification::*;
4
5/// Collection of server-side notification handlers
6pub struct Interface<Ops>
7where
8    Ops: OpsT,
9{
10    notifications: AHashMap<Ops, Box<dyn NotificationTrait>>,
11}
12
13impl<Ops> Default for Interface<Ops>
14where
15    Ops: OpsT,
16{
17    fn default() -> Self {
18        Self::new()
19    }
20}
21
22impl<Ops> Interface<Ops>
23where
24    Ops: OpsT,
25{
26    /// Create a new, empty notification interface.
27    pub fn new() -> Interface<Ops> {
28        Interface {
29            notifications: AHashMap::new(),
30        }
31    }
32
33    /// Register a notification handler for the given operation. Panics if a
34    /// handler for the same `op` has already been registered.
35    pub fn notification<Msg>(&mut self, op: Ops, method: Notification<Msg>)
36    where
37        Ops: OpsT,
38        Msg: BorshDeserialize + DeserializeOwned + Send + Sync + 'static,
39    {
40        let method: Box<dyn NotificationTrait> = Box::new(method);
41        if self.notifications.insert(op.clone(), method).is_some() {
42            panic!("RPC notification {op:?} is declared multiple times")
43        }
44    }
45
46    /// Dispatch a Borsh-encoded notification payload to the handler registered
47    /// for `op`, returning [`ServerError::NotFound`] if none exists.
48    pub async fn call_notification_with_borsh(&self, op: &Ops, payload: &[u8]) -> ServerResult<()> {
49        if let Some(notification) = self.notifications.get(op) {
50            notification.call_with_borsh(payload).await
51        } else {
52            Err(ServerError::NotFound)
53        }
54    }
55
56    /// Dispatch a JSON notification payload to the handler registered for `op`,
57    /// returning [`ServerError::NotFound`] if none exists.
58    pub async fn call_notification_with_serde_json(
59        &self,
60        op: &Ops,
61        payload: Value,
62    ) -> ServerResult<()> {
63        if let Some(notification) = self.notifications.get(op) {
64            notification.call_with_serde_json(payload).await
65        } else {
66            Err(ServerError::NotFound)
67        }
68    }
69}
70
71impl<Ops> From<Interface<Ops>> for Option<Arc<Interface<Ops>>>
72where
73    Ops: OpsT,
74{
75    fn from(interface: Interface<Ops>) -> Self {
76        Some(Arc::new(interface))
77    }
78}
workflow_rpc/client/interface/mod.rs

workflow_rpc/client/interface/
mod.rs
