airup_sdk/
rpc.rs

1//! # Airup RPC Protocol
2//! This module contains major structures and definitions for Airup's RPC protocol.
3//!
4//! ## Stream to Datagram
5//! Airup uses a very simple protocol to wrap streaming protocols to datagram protocols.
6//!
7//! ### The Conversation Model
8//! The simple protocol is a half-duplex, synchronous "blocking" protocol. The client sends a request and the server sends a
9//! response. If an serious protocol error occured, the connection will be simply shut down.
10//!
11//! ### Datagram Layout
12//! A datagram begins with a 64-bit long, little-endian ordered integer, which represents to the datagram's length. The length
13//! should be less than 6MiB and cannot be zero, or a serious protocol error will be occured. Then follows content of the
14//! datagram.
15
16use crate::error::ApiError;
17use serde::{Deserialize, Serialize, de::DeserializeOwned};
18
19/// A request object in the Airup RPC protocol.
20///
21/// Interpreted as CBOR, a serialized request object looks like (converted to human-readable JSON):
22///
23/// ```json
24/// {
25///     "status": "<ok | err>",
26///     "payload": <payload>,
27/// }
28/// ```
29///
30/// If the method requires no parameters, `params` can be `null` or even not present. If the method requires only one parameter,
31/// the field is the parameter itself. If the method requires more than one parameters, the field is an array filled with
32/// parameters.
33#[derive(Debug, Clone, Serialize, Deserialize)]
34pub struct Request {
35    pub method: String,
36
37    #[serde(alias = "param")]
38    pub params: Option<ciborium::Value>,
39}
40impl Request {
41    /// Creates a new [`Request`] with given method name and parameters.
42    pub fn new<M: Into<String>, C: Serialize, P: Into<Option<C>>>(method: M, params: P) -> Self {
43        let method = method.into();
44        let params = params
45            .into()
46            .map(|x| ciborium::Value::serialized(&x).unwrap());
47
48        Self { method, params }
49    }
50
51    /// Extracts parameters from the request.
52    pub fn extract_params<T: DeserializeOwned>(self) -> Result<T, ApiError> {
53        let value = self.params.unwrap_or(ciborium::Value::Null);
54        value.deserialized().map_err(ApiError::invalid_params)
55    }
56}
57
58/// A response object in the Airup RPC protocol.
59///
60/// Interpreted as CBOR, a serialized response object looks like (converted to human-readable JSON):
61///
62/// ```json
63/// {
64///     "status": "<ok | err>",
65///     "payload": <payload>,
66/// }
67/// ```
68///
69/// On success, `status` is `ok`, and `payload` is the return value of the requested method.
70///
71/// On failure, `status` is `err`, and `payload` is an [`Error`] object.
72#[derive(Debug, Clone, Serialize, Deserialize)]
73#[serde(rename_all = "kebab-case", tag = "status", content = "payload")]
74pub enum Response {
75    Ok(ciborium::Value),
76    Err(ApiError),
77}
78impl Response {
79    /// Creates a new `Response` from given `Result`.
80    ///
81    /// # Panics
82    /// Panics when CBOR serialization fails. This always assumes that the passed value is always interpreted as a value
83    /// CBOR object.
84    pub fn new<T: Serialize>(result: Result<T, ApiError>) -> Self {
85        match result {
86            Ok(val) => Self::Ok(ciborium::Value::serialized(&val).unwrap()),
87            Err(err) => Self::Err(err),
88        }
89    }
90
91    /// Converts from `Response` to a `Result`.
92    pub fn into_result<T: DeserializeOwned>(self) -> Result<T, ApiError> {
93        match self {
94            Self::Ok(val) => Ok(val
95                .deserialized()
96                .map_err(|err| ApiError::bad_response("TypeError", format!("{:?}", err)))?),
97            Self::Err(err) => Err(err),
98        }
99    }
100}
101
102/// A middle layer that splits a stream into messages.
103#[derive(Debug)]
104pub struct MessageProto<T> {
105    pub(crate) inner: T,
106    pub(crate) size_limit: usize,
107}
108impl<T> MessageProto<T> {
109    pub const DEFAULT_SIZE_LIMIT: usize = 128 * 1024;
110
111    /// Creates a new [`MessageProto`] with provided stream.
112    pub fn new(inner: T, size_limit: usize) -> Self {
113        Self { inner, size_limit }
114    }
115
116    /// Sets received message size limitation.
117    pub fn set_size_limit(&mut self, new: usize) -> usize {
118        std::mem::replace(&mut self.size_limit, new)
119    }
120
121    /// Gets received message size limitation.
122    pub fn size_limit(&self) -> usize {
123        self.size_limit
124    }
125
126    /// Gets the inner stream.
127    pub fn into_inner(self) -> T {
128        self.inner
129    }
130}
131impl<T> AsRef<T> for MessageProto<T> {
132    fn as_ref(&self) -> &T {
133        &self.inner
134    }
135}
136impl<T> AsMut<T> for MessageProto<T> {
137    fn as_mut(&mut self) -> &mut T {
138        &mut self.inner
139    }
140}
141impl<T> From<T> for MessageProto<T> {
142    fn from(value: T) -> Self {
143        Self::new(value, Self::DEFAULT_SIZE_LIMIT)
144    }
145}
146
147#[derive(Debug, thiserror::Error)]
148pub enum Error {
149    #[error("{0}")]
150    Io(std::io::Error),
151
152    #[error("{0}")]
153    CborSer(ciborium::ser::Error<std::io::Error>),
154
155    #[error("{0}")]
156    CborDe(ciborium::de::Error<std::io::Error>),
157
158    #[error("message too long: {0} bytes")]
159    MessageTooLong(usize),
160}
161impl From<std::io::Error> for Error {
162    fn from(value: std::io::Error) -> Self {
163        Self::Io(value)
164    }
165}
166impl From<ciborium::de::Error<std::io::Error>> for Error {
167    fn from(value: ciborium::de::Error<std::io::Error>) -> Self {
168        Self::CborDe(value)
169    }
170}
171impl From<ciborium::ser::Error<std::io::Error>> for Error {
172    fn from(value: ciborium::ser::Error<std::io::Error>) -> Self {
173        Self::CborSer(value)
174    }
175}
airup_sdk/rpc.rs

airup_sdk/
rpc.rs
