harmos_sidecar/
wire.rs

1/* ***
2 *
3 * Harmos
4 *
5 * SPDX-License-Identifier: Apache-2.0
6 *
7 * ***
8 */
9
10/* === Modules === */
11
12/* === Re-Exports === */
13
14/* === Imports === */
15
16// Standard Library Imports
17
18// Third Party Imports
19use tokio::io::{AsyncBufRead, AsyncBufReadExt, AsyncReadExt};
20// Local Imports
21use harmos_domain::Message;
22use harmos_sdk::operations::{CHILD_METHOD, OperationDeclaration, WIRE_METHOD, WireAnswer, WireChildCall, WireCommand};
23
24/* === Constants === */
25
26/* === Macros === */
27
28/* === Type Aliases === */
29
30/* === Structs === */
31
32#[derive(Clone, Debug, Default, PartialEq, Eq, serde::Deserialize, serde::Serialize)]
33#[serde(deny_unknown_fields)]
34pub struct PluginCatalog {
35    pub registrations: Vec<crate::Registration>,
36    pub observers: Vec<crate::inventory::observers::Declaration>,
37    pub operations: Vec<crate::operations::Definition>,
38}
39
40#[derive(Clone, Debug, PartialEq, Eq)]
41pub struct Publication {
42    pub stream: String,
43    pub schema: u32,
44    pub sequence: u64,
45    pub published_unix_micros: Option<i64>,
46    pub context: Option<String>,
47    pub payload: Vec<u8>,
48}
49
50/* === Enums === */
51
52#[derive(Clone, Debug, PartialEq, Eq)]
53pub enum Frame {
54    Call { id: u64, method: String, params: Vec<u8> },
55    Cancel { id: u64 },
56    Push { method: String, params: Vec<u8> },
57    Answer { id: u64, result: Vec<u8> },
58    Refusal { id: u64, refusal: harmos_domain::Refusal },
59}
60
61/* === Traits === */
62
63/* === Implementations === */
64
65impl Frame {
66    pub const REGISTER: &'static str = "harmos.register";
67    pub const EXITING: &'static str = "harmos.exiting";
68    pub const LIMIT: usize = 16 * 1024 * 1024;
69    pub const OVERSIZED: i64 = -32_603;
70
71    pub fn request_encoded_len<O: OperationDeclaration>(request: &O) -> usize {
72        Self::request_payload_encoded_len(O::ID, harmos_domain::Message::encoded_len(request))
73    }
74
75    // The largest frame carrying an encoded request of `payload` bytes for
76    // `operation`, as a routed call or as a child call.
77    pub fn request_payload_encoded_len(operation: &str, payload: usize) -> usize {
78        if operation.len() > Self::LIMIT {
79            return usize::MAX;
80        }
81        let payload = binary_payload_growth(payload);
82        let prepare = WireCommand::Prepare { invocation: u64::MAX, operation: operation.to_owned(), payload: Vec::new() };
83        let child = WireChildCall {
84            context: [u8::MAX; 32],
85            generation_context: Some([u8::MAX; 32]),
86            values: Vec::new(),
87            operation: operation.to_owned(),
88            payload: Vec::new(),
89            submit: true,
90        };
91        call_encoded_len(u64::MAX, WIRE_METHOD, prepare.encode().map_or(usize::MAX, |bytes| bytes.len()).saturating_add(payload))
92            .max(call_encoded_len(u64::MAX, CHILD_METHOD, child.encode().map_or(usize::MAX, |bytes| bytes.len()).saturating_add(payload)))
93    }
94
95    pub fn request_fits<O: OperationDeclaration>(request: &O) -> bool {
96        Self::request_encoded_len::<O>(request) <= Self::LIMIT
97    }
98
99    pub fn reply_encoded_len<O: OperationDeclaration>(reply: &O::Response) -> usize {
100        let empty = WireAnswer::Ready { payload: Vec::new() };
101        let payload = empty.encode().map_or(usize::MAX, |bytes| bytes.len()).saturating_add(binary_payload_growth(reply.encoded_len()));
102        answer_encoded_len(u64::MAX, payload)
103    }
104
105    pub fn reply_fits<O: OperationDeclaration>(reply: &O::Response) -> bool {
106        Self::reply_encoded_len::<O>(reply) <= Self::LIMIT
107    }
108
109    pub fn encoded_len(&self) -> usize {
110        match self {
111            Self::Call { id, method, params } => call_encoded_len(*id, method, params.len()),
112            Self::Answer { id, result } => answer_encoded_len(*id, result.len()),
113            Self::Refusal { id, refusal } => refusal_encoded_len(*id, refusal),
114            Self::Cancel { id } => 2 + varint_len(*id) + varint_len(u64::MAX),
115            Self::Push { method, params } => {
116                2usize.saturating_add(binary_bytes_len(method.len())).saturating_add(binary_bytes_len(params.len()))
117            }
118        }
119    }
120
121    pub fn fits(&self) -> bool {
122        self.encoded_len() <= Self::LIMIT
123    }
124
125    #[cfg(feature = "kit")]
126    pub(crate) fn exiting(reason: String) -> Self {
127        Self::Push { method: Self::EXITING.to_owned(), params: reason.encode_to_vec() }
128    }
129
130    pub fn exiting_reason(params: &[u8]) -> Option<String> {
131        String::decode(params).ok()
132    }
133
134    pub async fn diagnostic<R>(input: &mut R) -> std::io::Result<Option<String>>
135    where
136        R: AsyncBufRead + Unpin + Send,
137    {
138        loop {
139            let mut bytes = Vec::new();
140            let read = input.take(Self::LIMIT as u64 + 1).read_until(b'\n', &mut bytes).await?;
141            if read == 0 {
142                return Ok(None);
143            }
144            if bytes.len() > Self::LIMIT {
145                let mut rest = Vec::new();
146                input.read_until(b'\n', &mut rest).await?;
147                continue;
148            }
149
150            let text = String::from_utf8_lossy(&bytes);
151            return Ok(Some(text.trim_end_matches(['\n', '\r']).to_owned()));
152        }
153    }
154}
155
156/* === Public Functions === */
157
158// Postcard Length Of A Publication Without Its Following Rows Field.
159#[cfg(feature = "kit")]
160pub(crate) fn publication_encoded_len(
161    stream: &str, schema: u32, sequence: u64, payload: usize, published_unix_micros: Option<i64>, context: Option<&str>,
162) -> usize {
163    binary_bytes_len(stream.len())
164        .saturating_add(varint_len(u64::from(schema)))
165        .saturating_add(1 + varint_len(sequence))
166        .saturating_add(1 + published_unix_micros.map_or(0, zigzag_len))
167        .saturating_add(1 + context.map_or(0, |value| binary_bytes_len(value.len())))
168        .saturating_add(binary_bytes_len(payload))
169}
170
171// Postcard Length Of One Following Row.
172#[cfg(feature = "kit")]
173pub(crate) fn following_encoded_len(payload: usize, published_delta_micros: Option<i64>) -> usize {
174    binary_bytes_len(payload).saturating_add(1 + published_delta_micros.map_or(0, zigzag_len))
175}
176
177pub(crate) fn varint_len(value: u64) -> usize {
178    ((64 - value.leading_zeros()).max(1) as usize).div_ceil(7)
179}
180
181/* === Private Functions === */
182
183fn binary_payload_growth(bytes: usize) -> usize {
184    bytes.saturating_add(varint_len(bytes as u64).saturating_sub(1))
185}
186
187fn call_encoded_len(id: u64, method: &str, payload_len: usize) -> usize {
188    2usize.saturating_add(varint_len(id)).saturating_add(binary_bytes_len(method.len())).saturating_add(binary_bytes_len(payload_len))
189}
190
191fn answer_encoded_len(id: u64, payload_len: usize) -> usize {
192    2usize.saturating_add(varint_len(id)).saturating_add(binary_bytes_len(payload_len))
193}
194
195fn refusal_encoded_len(id: u64, refusal: &harmos_domain::Refusal) -> usize {
196    let refusal = postcard::experimental::serialized_size(refusal).unwrap_or(usize::MAX);
197    2usize.saturating_add(varint_len(id)).saturating_add(refusal)
198}
199
200#[cfg(feature = "kit")]
201fn zigzag_len(value: i64) -> usize {
202    varint_len(((value as u64) << 1) ^ ((value >> 63) as u64))
203}
204
205fn binary_bytes_len(length: usize) -> usize {
206    varint_len(length as u64).saturating_add(length)
207}
208
209/* === Tests === */
harmos_sidecar/wire.rs

harmos_sidecar/
wire.rs
