workflow_rpc/
messages.rs

1//!
2//! RPC message serialization module (header serialization and deserialization for `Borsh` and `JSON` data structures)
3//!
4
5pub mod serde_json {
6    //! RPC message serialization for JSON encoding
7    use serde::{Deserialize, Serialize};
8    use serde_json::{self, Value};
9
10    #[derive(Debug, Serialize, Deserialize)]
11    /// JSON-encoded request message sent from the client to the server.
12    pub struct JsonClientMessage<Ops, Id> {
13        // pub jsonrpc: String,
14        /// Optional request id used to correlate the server response.
15        pub id: Option<Id>,
16        /// Operation (method) being requested.
17        pub method: Ops,
18        /// Request params payload.
19        pub params: Value,
20    }
21
22    impl<Ops, Id> JsonClientMessage<Ops, Id> {
23        /// Create a new JSON client request from an optional request id, the
24        /// requested method and its params payload.
25        pub fn new(id: Option<Id>, method: Ops, payload: Value) -> Self {
26            JsonClientMessage {
27                // jsonrpc: "2.0".to_owned(),
28                id,
29                method,
30                params: payload,
31            }
32        }
33    }
34
35    #[derive(Debug, Serialize, Deserialize)]
36    /// JSON-encoded message sent from the server to the client, used for
37    /// responses, errors and notifications.
38    pub struct JSONServerMessage<Ops, Id> {
39        // pub jsonrpc: String,
40        /// Id of the request this message responds to, if any.
41        #[serde(skip_serializing_if = "Option::is_none")]
42        pub id: Option<Id>,
43        /// Operation (method) associated with the message, if any.
44        #[serde(skip_serializing_if = "Option::is_none")]
45        pub method: Option<Ops>,
46        /// Result or notification payload, if any.
47        #[serde(skip_serializing_if = "Option::is_none")]
48        pub params: Option<Value>,
49        // #[serde(skip_serializing_if = "Option::is_none")]
50        // pub result: Option<Value>,
51        /// Error detail, present when the message reports a failure.
52        #[serde(skip_serializing_if = "Option::is_none")]
53        pub error: Option<JsonServerError>,
54    }
55
56    impl<Ops, Id> JSONServerMessage<Ops, Id> {
57        /// Create a new JSON server message from an optional request id, method,
58        /// params and error.
59        pub fn new(
60            id: Option<Id>,
61            method: Option<Ops>,
62            params: Option<Value>,
63            // result: Option<Value>,
64            error: Option<JsonServerError>,
65        ) -> Self {
66            JSONServerMessage {
67                // jsonrpc: "2.0".to_owned(),
68                method,
69                params,
70                // result,
71                error,
72                id,
73            }
74        }
75    }
76
77    #[derive(Debug, Serialize, Deserialize)]
78    /// JSON-RPC style server error carrying a numeric code, a human-readable
79    /// message and an optional structured data payload.
80    pub struct JsonServerError {
81        code: u64,
82        message: String,
83        data: Option<Value>,
84    }
85
86    impl std::fmt::Display for JsonServerError {
87        fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::result::Result<(), std::fmt::Error> {
88            write!(
89                f,
90                "code:{}  message:`{}` data:{:?}",
91                self.code, self.message, self.data
92            )
93        }
94    }
95
96    impl From<crate::error::ServerError> for JsonServerError {
97        fn from(err: crate::error::ServerError) -> Self {
98            JsonServerError {
99                code: 0, //err.code,
100                message: err.to_string(),
101                data: None, //err.data,
102            }
103        }
104    }
105}
106
107pub mod borsh {
108    //! RPC message serialization for Borsh encoding
109
110    use crate::error::Error;
111    use borsh::{BorshDeserialize, BorshSerialize};
112    use workflow_websocket::client::message::Message as WebSocketMessage;
113    // use borsh::de::*;
114
115    /// Serialize a request header and payload into a single binary WebSocket
116    /// message (the serialized header followed by the raw payload bytes).
117    pub fn to_ws_msg<Ops, Id>(header: BorshReqHeader<Ops, Id>, payload: &[u8]) -> WebSocketMessage
118    where
119        Id: BorshSerialize + BorshDeserialize,
120        Ops: BorshSerialize + BorshDeserialize,
121    {
122        let header = borsh::to_vec(&header).expect("to_ws_msg header serialize error");
123        let header_len = header.len();
124        let len = payload.len() + header_len;
125        let mut buffer = Vec::with_capacity(len);
126        #[allow(clippy::uninit_vec)]
127        unsafe {
128            buffer.set_len(len);
129        }
130        buffer[0..header_len].copy_from_slice(&header);
131        buffer[header_len..].copy_from_slice(payload);
132        buffer.into()
133    }
134
135    #[derive(Debug, BorshSerialize, BorshDeserialize)]
136    /// Header of a Borsh-encoded client request, carrying the optional request
137    /// id (used to correlate the response) and the requested operation.
138    pub struct BorshReqHeader<Ops, Id>
139    where
140        Id: BorshSerialize + BorshDeserialize,
141        Ops: BorshSerialize + BorshDeserialize,
142    {
143        /// Optional request id used to correlate the server response.
144        pub id: Option<Id>, //u64,
145        /// Operation (method) being requested.
146        pub op: Ops,
147    }
148
149    impl<Ops, Id> BorshReqHeader<Ops, Id>
150    where
151        Id: BorshSerialize + BorshDeserialize,
152        Ops: BorshSerialize + BorshDeserialize,
153    {
154        /// Create a new request header from an optional request id and an operation.
155        pub fn new(id: Option<Id>, op: Ops) -> Self {
156            BorshReqHeader { id, op }
157        }
158    }
159
160    #[derive(Debug, BorshSerialize, BorshDeserialize)]
161    /// Header of a Borsh-encoded server message, identifying the request it
162    /// responds to (if any), its kind and the associated operation.
163    pub struct BorshServerMessageHeader<Ops, Id> {
164        /// Id of the request this message responds to, if any.
165        pub id: Option<Id>, //u64,
166        /// Whether the message is a success, error or notification.
167        pub kind: ServerMessageKind,
168        /// Operation associated with the message, if any.
169        pub op: Option<Ops>,
170    }
171
172    impl<Ops, Id> BorshServerMessageHeader<Ops, Id>
173    // where
174    //     Id: Default,
175    {
176        /// Create a new server message header from a request id, message kind
177        /// and an optional operation.
178        pub fn new(id: Option<Id>, kind: ServerMessageKind, op: Option<Ops>) -> Self {
179            Self { id, kind, op }
180        }
181    }
182
183    #[derive(Debug, Clone, Copy, BorshSerialize, BorshDeserialize)]
184    #[borsh(use_discriminant = true)]
185    /// Discriminant identifying the nature of a server message.
186    pub enum ServerMessageKind {
187        /// A successful response to a client request.
188        Success = 0,
189        /// An error response to a client request.
190        Error = 1,
191        /// A server-initiated notification (not tied to a request).
192        Notification = 0xff,
193    }
194
195    impl From<ServerMessageKind> for u32 {
196        fn from(kind: ServerMessageKind) -> u32 {
197            kind as u32
198        }
199    }
200
201    #[derive(Debug)]
202    /// Error outcome of a server response, distinguishing between a missing
203    /// payload, a typed error payload, and a transport/RPC-level error.
204    pub enum RespError<T>
205    where
206        T: BorshDeserialize,
207    {
208        /// The error response carried no associated data.
209        NoData,
210        /// The error response carried a deserialized typed payload.
211        Data(T),
212        /// A transport or RPC-level error occurred.
213        Rpc(Error),
214    }
215
216    #[derive(Debug)]
217    /// A Borsh-encoded client request message combining a header with a
218    /// borrowed, already-serialized payload slice.
219    pub struct BorshClientMessage<'data, Ops, Id>
220    where
221        Id: BorshSerialize + BorshDeserialize + 'data,
222        Ops: BorshSerialize + BorshDeserialize + 'data,
223    {
224        /// Request header carrying the optional request id and operation.
225        pub header: BorshReqHeader<Ops, Id>,
226        /// Raw (already serialized) request payload bytes.
227        pub payload: &'data [u8],
228    }
229
230    impl<'data, Ops, Id> TryFrom<&'data Vec<u8>> for BorshClientMessage<'data, Ops, Id>
231    where
232        Id: BorshSerialize + BorshDeserialize + 'data,
233        Ops: BorshSerialize + BorshDeserialize + 'data,
234    {
235        type Error = Error;
236
237        fn try_from(src: &'data Vec<u8>) -> Result<Self, Self::Error> {
238            let v: BorshClientMessage<Ops, Id> = src[..].try_into()?;
239            Ok(v)
240        }
241    }
242
243    impl<'data, Ops, Id> TryFrom<&'data [u8]> for BorshClientMessage<'data, Ops, Id>
244    where
245        Id: BorshSerialize + BorshDeserialize + 'data,
246        Ops: BorshSerialize + BorshDeserialize + 'data,
247    {
248        type Error = Error;
249
250        fn try_from(src: &'data [u8]) -> Result<Self, Self::Error> {
251            let mut payload = src;
252            let header = BorshReqHeader::<Ops, Id>::deserialize(&mut payload)?;
253            let message = BorshClientMessage { header, payload };
254            Ok(message)
255        }
256    }
257
258    #[derive(Debug)]
259    /// A Borsh-encoded server message combining a header with a borrowed,
260    /// already-serialized payload slice.
261    pub struct BorshServerMessage<'data, Ops, Id>
262    where
263        Id: BorshSerialize + BorshDeserialize + 'data,
264        Ops: BorshSerialize + BorshDeserialize + 'data,
265    {
266        /// Message header carrying the request id, message kind and operation.
267        pub header: BorshServerMessageHeader<Ops, Id>,
268        /// Raw (already serialized) message payload bytes.
269        pub payload: &'data [u8],
270    }
271
272    impl<'data, Ops, Id> BorshServerMessage<'data, Ops, Id>
273    where
274        Id: BorshSerialize + BorshDeserialize + 'data,
275        Ops: BorshSerialize + BorshDeserialize + 'data,
276    {
277        /// Create a new server message from a header and a borrowed payload slice.
278        pub fn new(
279            header: BorshServerMessageHeader<Ops, Id>,
280            payload: &'data [u8],
281        ) -> BorshServerMessage<'data, Ops, Id> {
282            BorshServerMessage { header, payload }
283        }
284
285        /// Serialize the header and payload into a single contiguous
286        /// byte buffer (header followed by the raw payload bytes).
287        pub fn try_to_vec(&self) -> Result<Vec<u8>, Error> {
288            let header = borsh::to_vec(&self.header)?;
289            let header_len = header.len();
290
291            let len = header_len + self.payload.len();
292            let mut buffer = Vec::with_capacity(len);
293            #[allow(clippy::uninit_vec)]
294            unsafe {
295                buffer.set_len(len);
296            }
297
298            buffer[0..header_len].copy_from_slice(&header);
299            if !self.payload.is_empty() {
300                buffer[header_len..].copy_from_slice(self.payload);
301            }
302            Ok(buffer)
303        }
304    }
305
306    impl<'data, Ops, Id> TryFrom<&'data [u8]> for BorshServerMessage<'data, Ops, Id>
307    where
308        Id: BorshSerialize + BorshDeserialize + 'data,
309        Ops: BorshSerialize + BorshDeserialize + 'data,
310    {
311        type Error = Error;
312
313        fn try_from(src: &'data [u8]) -> Result<Self, Self::Error> {
314            let mut payload = src;
315            let header = <BorshServerMessageHeader<Ops, Id>>::deserialize(&mut payload)?;
316            let message = BorshServerMessage { header, payload };
317            Ok(message)
318        }
319    }
320}
workflow_rpc/messages.rs

workflow_rpc/
messages.rs
