workflow_rpc/server/protocol/
borsh.rs

1//!
2//! Module containing [`BorshProtocol`] responsible for server-side
3//! dispatch of RPC methods and notifications when using `Borsh`
4//! protocol.
5//!
6
7use super::Encoding;
8use crate::imports::*;
9use crate::messages::borsh::*;
10use crate::server::Interface;
11use crate::server::ProtocolHandler;
12pub use crate::server::result::Result;
13use workflow_websocket::server::{
14    Error as WebSocketError, Message, Result as WebSocketResult, WebSocketSink,
15};
16
17/// Server-side message serializer and dispatcher when using `Borsh` protocol.
18pub struct BorshProtocol<ServerContext, ConnectionContext, Ops, Id>
19where
20    ServerContext: Clone + Send + Sync + 'static,
21    ConnectionContext: Clone + Send + Sync + 'static,
22    Ops: OpsT,
23    Id: IdT,
24{
25    id: PhantomData<Id>,
26    ops: PhantomData<Ops>,
27    interface: Arc<Interface<ServerContext, ConnectionContext, Ops>>,
28}
29
30#[async_trait]
31impl<ServerContext, ConnectionContext, Ops, Id>
32    ProtocolHandler<ServerContext, ConnectionContext, Ops>
33    for BorshProtocol<ServerContext, ConnectionContext, Ops, Id>
34where
35    ServerContext: Clone + Send + Sync + 'static,
36    ConnectionContext: Clone + Send + Sync + 'static,
37    Ops: OpsT,
38    Id: IdT,
39{
40    fn new(interface: Arc<Interface<ServerContext, ConnectionContext, Ops>>) -> Self
41    where
42        Self: Sized,
43    {
44        BorshProtocol {
45            id: PhantomData,
46            ops: PhantomData,
47            interface,
48        }
49    }
50
51    fn encoding(&self) -> Encoding {
52        Encoding::Borsh
53    }
54
55    async fn handle_message(
56        &self,
57        connection_ctx: ConnectionContext,
58        msg: Message,
59        sink: &WebSocketSink,
60    ) -> WebSocketResult<()> {
61        let data = msg.into_data();
62        let req: BorshClientMessage<Ops, Id> = data
63            .as_ref()
64            .try_into()
65            .map_err(|_| WebSocketError::MalformedMessage)?;
66
67        if req.header.id.is_some() {
68            let result = self
69                .interface
70                .call_method_with_borsh(&req.header.op, connection_ctx, req.payload)
71                .await;
72
73            match result {
74                Ok(data) => {
75                    if let Ok(msg) = BorshServerMessage::<Ops, Id>::new(
76                        BorshServerMessageHeader::new(
77                            req.header.id,
78                            ServerMessageKind::Success,
79                            Some(req.header.op),
80                        ),
81                        &data,
82                    )
83                    .try_to_vec()
84                        && let Err(e) = sink.send(msg.into())
85                    {
86                        log_trace!("Sink error: {:?}", e);
87                    }
88                }
89                Err(err) => {
90                    // log_trace!("RPC server error: {:?} req: {:#?}", err, req);
91                    if err == ServerError::Close {
92                        return Err(WebSocketError::ServerClose);
93                    } else if let Ok(err_vec) = borsh::to_vec(&err)
94                        && let Ok(msg) = BorshServerMessage::new(
95                            BorshServerMessageHeader::<Ops, Id>::new(
96                                req.header.id,
97                                ServerMessageKind::Error,
98                                None,
99                            ),
100                            &err_vec,
101                        )
102                        .try_to_vec()
103                        && let Err(e) = sink.send(msg.into())
104                    {
105                        log_trace!("Sink error: {:?}", e);
106                    }
107                }
108            }
109        } else {
110            self.interface
111                .call_notification_with_borsh(&req.header.op, connection_ctx, req.payload)
112                .await
113                .unwrap_or_else(|err| {
114                    log_trace!("error handling client-side notification {}", err)
115                });
116        }
117
118        Ok(())
119    }
120
121    fn serialize_notification_message<Msg>(&self, op: Ops, msg: Msg) -> Result<tungstenite::Message>
122    where
123        Msg: BorshSerialize + Send + Sync + 'static,
124    {
125        create_serialized_notification_message(op, msg)
126    }
127}
128
129/// Build a Borsh-encoded notification WebSocket message for the given operation
130/// and message payload.
131pub fn create_serialized_notification_message<Ops, Msg>(op: Ops, msg: Msg) -> Result<Message>
132where
133    Ops: OpsT,
134    Msg: BorshSerialize + Send + Sync + 'static,
135{
136    let payload = borsh::to_vec(&msg)?;
137    let data = BorshServerMessage::new(
138        BorshServerMessageHeader::<Ops, ()>::new(None, ServerMessageKind::Notification, Some(op)),
139        &payload,
140    )
141    .try_to_vec()?;
142    Ok(Message::Binary(data.into()))
143}
workflow_rpc/server/protocol/borsh.rs

workflow_rpc/server/protocol/
borsh.rs
