816 lines
27 KiB
Rust
816 lines
27 KiB
Rust
//! The wire: scalar encodings, strict JSON, the envelope and its framing (bus-v1 section 4).
|
|
//!
|
|
//! A frame is a `u32` little-endian byte count followed by that many bytes of UTF-8 JSON. The
|
|
//! JSON is parsed strictly: duplicate keys at any depth, invalid UTF-8, non-finite numbers,
|
|
//! trailing bytes and unknown envelope fields are all refused. Both ends use this module, so
|
|
//! the router and the client agree byte for byte on what is valid.
|
|
|
|
use std::collections::HashSet;
|
|
use std::fmt;
|
|
|
|
use serde::de::{self, DeserializeSeed, MapAccess, SeqAccess, Visitor};
|
|
use serde_json::{Map, Value};
|
|
use sha2::{Digest as _, Sha256};
|
|
use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
|
|
|
|
use crate::error::BusError;
|
|
|
|
pub const PROTOCOL: &str = "flybus";
|
|
pub const MAJOR: u64 = 1;
|
|
pub const MINOR: u64 = 0;
|
|
/// Largest JSON envelope, in bytes, not counting the four-byte length prefix.
|
|
pub const MAX_ENVELOPE_BYTES: usize = 65_536;
|
|
pub const MAX_ATTACHMENTS: usize = 32;
|
|
/// Release and consume batches carry 1..=64 ids.
|
|
pub const MAX_BATCH: usize = 64;
|
|
/// Queue and credit requests are integers in 1..=65535.
|
|
pub const MAX_CREDIT: u64 = 65_535;
|
|
pub const MAX_NAME_LEN: usize = 192;
|
|
pub const MAX_METHOD_LEN: usize = 128;
|
|
pub const MAX_CONTENT_TYPE_LEN: usize = 127;
|
|
/// The one artifact generation v1 issues; ids and inodes are never reused.
|
|
pub const GENERATION: u64 = 1;
|
|
|
|
/// The operations, delivery and notice shapes this implementation speaks. `contractDigest` is
|
|
/// the SHA-256 of this text, so any change to it changes the digest a client sees in hello.
|
|
pub const CONTRACT: &str = "flybus 1.0
|
|
frame: u32le length, 1..=65536 bytes of strict UTF-8 JSON
|
|
envelope: protocol major minor id replyTo kind op body attachments
|
|
attachment: name ref ownerId
|
|
ref: storeId artifactId generation byteLength contentType digest
|
|
reply: ok value | ok error{code message dispatch}
|
|
bus.hello: clientId clientIncarnation supportedMajors -> routerId connectionId selectedMajor selectedMinor contractDigest limits
|
|
service.register: name maxQueued maxInFlight -> serviceIncarnation
|
|
service.unregister: name serviceIncarnation -> removed
|
|
rpc.call: callId target expectedIncarnation method payload -> accepted serviceIncarnation
|
|
rpc.reply: callId requestDeliveryId outcome -> routed
|
|
rpc.responder.release: callId requestDeliveryId -> released; final attached release emits call.failed dispatched (CALL_GONE, or NO_SERVICE after route loss)
|
|
rpc.cancel: callId -> state(cancelled-before-dispatch|execution-unknown|completed|call-gone)
|
|
topic.declare: name retained(none|latest) -> declared topicIncarnation
|
|
topic.clear: name -> cleared
|
|
topic.delete: name -> deleted
|
|
subscribe: topic mode(latest|bounded) maxQueued maxInFlight replayLatest -> subscriptionId topicIncarnation
|
|
unsubscribe: subscriptionId -> removed
|
|
publish: topic payload -> topicSequence subscribers replaced
|
|
delivery.consumed: deliveryIds -> released
|
|
artifact.allocate: byteLength contentType -> artifactId generation ownerId writeLocation
|
|
artifact.seal: artifactId generation ownerId digest -> ref ownerId
|
|
artifact.open: ref ownerId -> readLocation
|
|
artifact.retain: ref ownerId -> ownerId
|
|
artifact.release: ownerIds -> released
|
|
delivery rpc.request: deliveryId callId caller target serviceIncarnation method payload
|
|
delivery rpc.result: deliveryId callId responder serviceIncarnation outcome
|
|
delivery topic.message: deliveryId subscriptionId topic topicIncarnation topicSequence replaced payload
|
|
notice call.failed: callId code message dispatch
|
|
notice route.removed: name serviceIncarnation reason
|
|
notice subscription.closed: subscriptionId topic topicIncarnation reason
|
|
notice connection.closing: code message
|
|
";
|
|
|
|
/// SHA-256 of [`CONTRACT`], lowercase hex.
|
|
pub fn contract_digest() -> String {
|
|
hex(&Sha256::digest(CONTRACT.as_bytes()))
|
|
}
|
|
|
|
pub(crate) fn hex(bytes: &[u8]) -> String {
|
|
let mut s = String::with_capacity(bytes.len() * 2);
|
|
for b in bytes {
|
|
s.push_str(&format!("{b:02x}"));
|
|
}
|
|
s
|
|
}
|
|
|
|
/// A wire-level validation failure. Always maps to `INVALID_ENVELOPE`.
|
|
#[derive(Clone, Debug, PartialEq, Eq)]
|
|
pub struct WireError(pub String);
|
|
|
|
impl fmt::Display for WireError {
|
|
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
|
|
f.write_str(&self.0)
|
|
}
|
|
}
|
|
|
|
impl std::error::Error for WireError {}
|
|
|
|
impl From<WireError> for BusError {
|
|
fn from(e: WireError) -> BusError {
|
|
BusError::invalid(e.0)
|
|
}
|
|
}
|
|
|
|
fn err<T>(message: impl Into<String>) -> Result<T, WireError> {
|
|
Err(WireError(message.into()))
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------------------------
|
|
// Scalars
|
|
|
|
/// `Id`: `^[a-z0-9][a-z0-9._-]{0,63}$`.
|
|
pub fn is_id(s: &str) -> bool {
|
|
let b = s.as_bytes();
|
|
!b.is_empty()
|
|
&& b.len() <= 64
|
|
&& (b[0].is_ascii_lowercase() || b[0].is_ascii_digit())
|
|
&& b.iter().all(|&c| {
|
|
c.is_ascii_lowercase() || c.is_ascii_digit() || matches!(c, b'.' | b'_' | b'-')
|
|
})
|
|
}
|
|
|
|
/// `U64`: `"0"` or `[1-9][0-9]*`, at most `u64::MAX`.
|
|
pub fn parse_u64(s: &str) -> Option<u64> {
|
|
let b = s.as_bytes();
|
|
if b.is_empty() || b.len() > 20 || (b.len() > 1 && b[0] == b'0') {
|
|
return None;
|
|
}
|
|
let mut n: u64 = 0;
|
|
for &c in b {
|
|
if !c.is_ascii_digit() {
|
|
return None;
|
|
}
|
|
n = n.checked_mul(10)?.checked_add(u64::from(c - b'0'))?;
|
|
}
|
|
Some(n)
|
|
}
|
|
|
|
/// `Digest`: 64 lowercase hexadecimal digits.
|
|
pub fn is_digest(s: &str) -> bool {
|
|
s.len() == 64
|
|
&& s.bytes()
|
|
.all(|c| c.is_ascii_digit() || (b'a'..=b'f').contains(&c))
|
|
}
|
|
|
|
/// Service and topic names: 1..=192 of `[a-z0-9._-]`, no empty dot-separated segment.
|
|
pub fn is_name(s: &str) -> bool {
|
|
!s.is_empty()
|
|
&& s.len() <= MAX_NAME_LEN
|
|
&& s.bytes().all(|c| {
|
|
c.is_ascii_lowercase() || c.is_ascii_digit() || matches!(c, b'.' | b'_' | b'-')
|
|
})
|
|
&& s.split('.').all(|seg| !seg.is_empty())
|
|
}
|
|
|
|
/// RPC method: 1..=128 printable ASCII characters.
|
|
pub fn is_method(s: &str) -> bool {
|
|
!s.is_empty() && s.len() <= MAX_METHOD_LEN && s.bytes().all(|c| (0x20..=0x7e).contains(&c))
|
|
}
|
|
|
|
/// Content type: 1..=127 printable ASCII characters.
|
|
pub fn is_content_type(s: &str) -> bool {
|
|
!s.is_empty()
|
|
&& s.len() <= MAX_CONTENT_TYPE_LEN
|
|
&& s.bytes().all(|c| (0x20..=0x7e).contains(&c))
|
|
}
|
|
|
|
/// Operation names: 1..=64 of `[a-z.]`.
|
|
pub fn is_op(s: &str) -> bool {
|
|
!s.is_empty() && s.len() <= 64 && s.bytes().all(|c| c.is_ascii_lowercase() || c == b'.')
|
|
}
|
|
|
|
/// `<prefix>-<U64>`, the canonical form of every serial-numbered id on the bus.
|
|
pub fn serial_id(prefix: &str, n: u64) -> String {
|
|
format!("{prefix}-{n}")
|
|
}
|
|
|
|
/// Parses `<prefix>-<U64>`; `None` unless canonical.
|
|
pub fn parse_serial_id(prefix: &str, s: &str) -> Option<u64> {
|
|
s.strip_prefix(prefix)?
|
|
.strip_prefix('-')
|
|
.and_then(parse_u64)
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------------------------
|
|
// Strict JSON
|
|
|
|
/// Parses one JSON value, refusing duplicate object keys at any depth, invalid UTF-8,
|
|
/// non-finite numbers and trailing data. Nesting depth is bounded by serde_json's recursion
|
|
/// limit (128).
|
|
pub fn parse_json_strict(bytes: &[u8]) -> Result<Value, WireError> {
|
|
if std::str::from_utf8(bytes).is_err() {
|
|
return err("invalid UTF-8");
|
|
}
|
|
let mut de = serde_json::Deserializer::from_slice(bytes);
|
|
let value = StrictSeed
|
|
.deserialize(&mut de)
|
|
.map_err(|e| WireError(format!("invalid JSON: {e}")))?;
|
|
de.end()
|
|
.map_err(|e| WireError(format!("invalid JSON: {e}")))?;
|
|
Ok(value)
|
|
}
|
|
|
|
struct StrictSeed;
|
|
|
|
impl<'de> DeserializeSeed<'de> for StrictSeed {
|
|
type Value = Value;
|
|
|
|
fn deserialize<D: de::Deserializer<'de>>(self, d: D) -> Result<Value, D::Error> {
|
|
d.deserialize_any(StrictVisitor)
|
|
}
|
|
}
|
|
|
|
struct StrictVisitor;
|
|
|
|
impl<'de> Visitor<'de> for StrictVisitor {
|
|
type Value = Value;
|
|
|
|
fn expecting(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
|
|
f.write_str("a JSON value")
|
|
}
|
|
|
|
fn visit_bool<E>(self, v: bool) -> Result<Value, E> {
|
|
Ok(Value::Bool(v))
|
|
}
|
|
|
|
fn visit_i64<E>(self, v: i64) -> Result<Value, E> {
|
|
Ok(Value::from(v))
|
|
}
|
|
|
|
fn visit_u64<E>(self, v: u64) -> Result<Value, E> {
|
|
Ok(Value::from(v))
|
|
}
|
|
|
|
fn visit_f64<E: de::Error>(self, v: f64) -> Result<Value, E> {
|
|
serde_json::Number::from_f64(v)
|
|
.map(Value::Number)
|
|
.ok_or_else(|| E::custom("non-finite number"))
|
|
}
|
|
|
|
fn visit_str<E>(self, v: &str) -> Result<Value, E> {
|
|
Ok(Value::String(v.to_owned()))
|
|
}
|
|
|
|
fn visit_string<E>(self, v: String) -> Result<Value, E> {
|
|
Ok(Value::String(v))
|
|
}
|
|
|
|
fn visit_unit<E>(self) -> Result<Value, E> {
|
|
Ok(Value::Null)
|
|
}
|
|
|
|
fn visit_seq<A: SeqAccess<'de>>(self, mut seq: A) -> Result<Value, A::Error> {
|
|
let mut out = Vec::new();
|
|
while let Some(v) = seq.next_element_seed(StrictSeed)? {
|
|
out.push(v);
|
|
}
|
|
Ok(Value::Array(out))
|
|
}
|
|
|
|
fn visit_map<A: MapAccess<'de>>(self, mut map: A) -> Result<Value, A::Error> {
|
|
let mut out = Map::new();
|
|
while let Some(key) = map.next_key::<String>()? {
|
|
if out.contains_key(&key) {
|
|
return Err(de::Error::custom(format!("duplicate key {key:?}")));
|
|
}
|
|
let v = map.next_value_seed(StrictSeed)?;
|
|
out.insert(key, v);
|
|
}
|
|
Ok(Value::Object(out))
|
|
}
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------------------------
|
|
// Exact-field object reader
|
|
|
|
/// Reads the fields of one JSON object, then refuses any it did not read.
|
|
pub struct Fields<'a> {
|
|
map: &'a Map<String, Value>,
|
|
seen: HashSet<&'static str>,
|
|
what: &'static str,
|
|
}
|
|
|
|
impl<'a> Fields<'a> {
|
|
pub fn new(v: &'a Value, what: &'static str) -> Result<Fields<'a>, WireError> {
|
|
match v {
|
|
Value::Object(map) => Ok(Fields::of(map, what)),
|
|
_ => err(format!("{what} must be an object")),
|
|
}
|
|
}
|
|
|
|
pub fn of(map: &'a Map<String, Value>, what: &'static str) -> Fields<'a> {
|
|
Fields {
|
|
map,
|
|
seen: HashSet::new(),
|
|
what,
|
|
}
|
|
}
|
|
|
|
pub fn value(&mut self, key: &'static str) -> Result<&'a Value, WireError> {
|
|
self.seen.insert(key);
|
|
match self.map.get(key) {
|
|
Some(v) => Ok(v),
|
|
None => err(format!("{}: missing field {key:?}", self.what)),
|
|
}
|
|
}
|
|
|
|
pub fn string(&mut self, key: &'static str) -> Result<&'a str, WireError> {
|
|
let what = self.what;
|
|
self.value(key)?
|
|
.as_str()
|
|
.ok_or_else(|| WireError(format!("{what}: {key} must be a string")))
|
|
}
|
|
|
|
fn checked(
|
|
&mut self,
|
|
key: &'static str,
|
|
ok: fn(&str) -> bool,
|
|
kind: &str,
|
|
) -> Result<String, WireError> {
|
|
let s = self.string(key)?;
|
|
if ok(s) {
|
|
Ok(s.to_owned())
|
|
} else {
|
|
err(format!("{}: {key} is not a valid {kind}", self.what))
|
|
}
|
|
}
|
|
|
|
pub fn id(&mut self, key: &'static str) -> Result<String, WireError> {
|
|
self.checked(key, is_id, "id")
|
|
}
|
|
|
|
pub fn nullable_id(&mut self, key: &'static str) -> Result<Option<String>, WireError> {
|
|
match self.value(key)? {
|
|
Value::Null => Ok(None),
|
|
_ => self.id(key).map(Some),
|
|
}
|
|
}
|
|
|
|
pub fn name(&mut self, key: &'static str) -> Result<String, WireError> {
|
|
self.checked(key, is_name, "name")
|
|
}
|
|
|
|
pub fn method(&mut self, key: &'static str) -> Result<String, WireError> {
|
|
self.checked(key, is_method, "method")
|
|
}
|
|
|
|
pub fn u64_string(&mut self, key: &'static str) -> Result<u64, WireError> {
|
|
let s = self.string(key)?;
|
|
parse_u64(s).ok_or_else(|| {
|
|
WireError(format!(
|
|
"{}: {key} is not a canonical U64 string",
|
|
self.what
|
|
))
|
|
})
|
|
}
|
|
|
|
/// A JSON integer in `lo..=hi`.
|
|
pub fn int(&mut self, key: &'static str, lo: u64, hi: u64) -> Result<u64, WireError> {
|
|
let what = self.what;
|
|
match self.value(key)?.as_u64() {
|
|
Some(n) if (lo..=hi).contains(&n) => Ok(n),
|
|
_ => err(format!("{what}: {key} must be an integer in {lo}..={hi}")),
|
|
}
|
|
}
|
|
|
|
pub fn boolean(&mut self, key: &'static str) -> Result<bool, WireError> {
|
|
let what = self.what;
|
|
self.value(key)?
|
|
.as_bool()
|
|
.ok_or_else(|| WireError(format!("{what}: {key} must be a boolean")))
|
|
}
|
|
|
|
pub fn object(&mut self, key: &'static str) -> Result<&'a Map<String, Value>, WireError> {
|
|
let what = self.what;
|
|
self.value(key)?
|
|
.as_object()
|
|
.ok_or_else(|| WireError(format!("{what}: {key} must be an object")))
|
|
}
|
|
|
|
pub fn array(
|
|
&mut self,
|
|
key: &'static str,
|
|
lo: usize,
|
|
hi: usize,
|
|
) -> Result<&'a Vec<Value>, WireError> {
|
|
let what = self.what;
|
|
match self.value(key)?.as_array() {
|
|
Some(a) if (lo..=hi).contains(&a.len()) => Ok(a),
|
|
_ => err(format!(
|
|
"{what}: {key} must be an array of {lo}..={hi} items"
|
|
)),
|
|
}
|
|
}
|
|
|
|
/// Refuses fields that were not read.
|
|
pub fn finish(self) -> Result<(), WireError> {
|
|
if let Some(extra) = self.map.keys().find(|k| !self.seen.contains(k.as_str())) {
|
|
return err(format!("{}: unknown field {extra:?}", self.what));
|
|
}
|
|
Ok(())
|
|
}
|
|
}
|
|
|
|
/// An array of ids, each `<prefix>-<U64>`, 1..=64 of them.
|
|
pub fn id_batch(v: &[Value], what: &str) -> Result<Vec<String>, WireError> {
|
|
let mut out = Vec::with_capacity(v.len());
|
|
for item in v {
|
|
match item.as_str() {
|
|
Some(s) if is_id(s) => out.push(s.to_owned()),
|
|
_ => return err(format!("{what}: every entry must be an id")),
|
|
}
|
|
}
|
|
Ok(out)
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------------------------
|
|
// Shared structures
|
|
|
|
/// An immutable artifact's identity. Not an address and not authority to read.
|
|
#[derive(Clone, Debug, PartialEq, Eq, Hash)]
|
|
pub struct ArtifactRef {
|
|
pub store_id: String,
|
|
pub artifact_id: String,
|
|
pub generation: u64,
|
|
pub byte_length: u64,
|
|
pub content_type: String,
|
|
pub digest: Option<String>,
|
|
}
|
|
|
|
impl ArtifactRef {
|
|
pub fn to_json(&self) -> Value {
|
|
let mut m = Map::new();
|
|
m.insert("storeId".into(), self.store_id.clone().into());
|
|
m.insert("artifactId".into(), self.artifact_id.clone().into());
|
|
m.insert("generation".into(), self.generation.to_string().into());
|
|
m.insert("byteLength".into(), self.byte_length.to_string().into());
|
|
m.insert("contentType".into(), self.content_type.clone().into());
|
|
m.insert(
|
|
"digest".into(),
|
|
self.digest.clone().map_or(Value::Null, Value::String),
|
|
);
|
|
Value::Object(m)
|
|
}
|
|
|
|
pub fn from_json(v: &Value) -> Result<ArtifactRef, WireError> {
|
|
let mut f = Fields::new(v, "ref")?;
|
|
let store_id = f.id("storeId")?;
|
|
let artifact_id = f.id("artifactId")?;
|
|
let generation = f.u64_string("generation")?;
|
|
let byte_length = f.u64_string("byteLength")?;
|
|
let content_type = f.string("contentType")?;
|
|
if !is_content_type(content_type) {
|
|
return err("ref: contentType must be 1..=127 printable ASCII characters");
|
|
}
|
|
let digest = match f.value("digest")? {
|
|
Value::Null => None,
|
|
Value::String(s) if is_digest(s) => Some(s.clone()),
|
|
_ => return err("ref: digest must be null or 64 lowercase hex digits"),
|
|
};
|
|
f.finish()?;
|
|
Ok(ArtifactRef {
|
|
store_id,
|
|
artifact_id,
|
|
generation,
|
|
byte_length,
|
|
content_type: content_type.to_owned(),
|
|
digest,
|
|
})
|
|
}
|
|
}
|
|
|
|
/// One attachment entry: an application name, the reference and the sender's owner token.
|
|
#[derive(Clone, Debug, PartialEq, Eq)]
|
|
pub struct Attachment {
|
|
pub name: String,
|
|
pub reference: ArtifactRef,
|
|
pub owner_id: String,
|
|
}
|
|
|
|
impl Attachment {
|
|
pub fn to_json(&self) -> Value {
|
|
let mut m = Map::new();
|
|
m.insert("name".into(), self.name.clone().into());
|
|
m.insert("ref".into(), self.reference.to_json());
|
|
m.insert("ownerId".into(), self.owner_id.clone().into());
|
|
Value::Object(m)
|
|
}
|
|
|
|
pub fn from_json(v: &Value) -> Result<Attachment, WireError> {
|
|
let mut f = Fields::new(v, "attachment")?;
|
|
let name = f.id("name")?;
|
|
let reference = ArtifactRef::from_json(f.value("ref")?)?;
|
|
let owner_id = f.id("ownerId")?;
|
|
f.finish()?;
|
|
Ok(Attachment {
|
|
name,
|
|
reference,
|
|
owner_id,
|
|
})
|
|
}
|
|
}
|
|
|
|
/// A store location grant: a path relative to `<store root>/<storeId>`. SDK-private.
|
|
#[derive(Clone, Debug, PartialEq, Eq)]
|
|
pub struct Location {
|
|
pub store_id: String,
|
|
pub relative_path: String,
|
|
}
|
|
|
|
impl Location {
|
|
pub fn to_json(&self) -> Value {
|
|
let mut m = Map::new();
|
|
m.insert("storeId".into(), self.store_id.clone().into());
|
|
m.insert("relativePath".into(), self.relative_path.clone().into());
|
|
Value::Object(m)
|
|
}
|
|
|
|
pub fn from_json(v: &Value) -> Result<Location, WireError> {
|
|
let mut f = Fields::new(v, "location")?;
|
|
let store_id = f.id("storeId")?;
|
|
let relative_path = f.string("relativePath")?.to_owned();
|
|
f.finish()?;
|
|
Ok(Location {
|
|
store_id,
|
|
relative_path,
|
|
})
|
|
}
|
|
}
|
|
|
|
/// A participant identity as the router reports it on deliveries. It is authenticated only
|
|
/// when the connection was accepted through a launcher-bound transport entry point.
|
|
#[derive(Clone, Debug, PartialEq, Eq, Hash)]
|
|
pub struct Identity {
|
|
pub client_id: String,
|
|
pub client_incarnation: String,
|
|
}
|
|
|
|
impl Identity {
|
|
pub fn to_json(&self) -> Value {
|
|
let mut m = Map::new();
|
|
m.insert("clientId".into(), self.client_id.clone().into());
|
|
m.insert(
|
|
"clientIncarnation".into(),
|
|
self.client_incarnation.clone().into(),
|
|
);
|
|
Value::Object(m)
|
|
}
|
|
|
|
pub fn from_json(v: &Value) -> Result<Identity, WireError> {
|
|
let mut f = Fields::new(v, "identity")?;
|
|
let client_id = f.id("clientId")?;
|
|
let client_incarnation = f.id("clientIncarnation")?;
|
|
f.finish()?;
|
|
Ok(Identity {
|
|
client_id,
|
|
client_incarnation,
|
|
})
|
|
}
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------------------------
|
|
// Envelope
|
|
|
|
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
|
|
pub enum Kind {
|
|
Command,
|
|
Reply,
|
|
Delivery,
|
|
Notice,
|
|
}
|
|
|
|
impl Kind {
|
|
pub fn as_str(self) -> &'static str {
|
|
match self {
|
|
Kind::Command => "command",
|
|
Kind::Reply => "reply",
|
|
Kind::Delivery => "delivery",
|
|
Kind::Notice => "notice",
|
|
}
|
|
}
|
|
|
|
fn parse(s: &str) -> Option<Kind> {
|
|
match s {
|
|
"command" => Some(Kind::Command),
|
|
"reply" => Some(Kind::Reply),
|
|
"delivery" => Some(Kind::Delivery),
|
|
"notice" => Some(Kind::Notice),
|
|
_ => None,
|
|
}
|
|
}
|
|
}
|
|
|
|
#[derive(Clone, Debug, PartialEq)]
|
|
pub struct Envelope {
|
|
pub major: u64,
|
|
pub minor: u64,
|
|
pub id: String,
|
|
pub reply_to: Option<String>,
|
|
pub kind: Kind,
|
|
pub op: String,
|
|
pub body: Map<String, Value>,
|
|
pub attachments: Vec<Attachment>,
|
|
}
|
|
|
|
impl Envelope {
|
|
pub fn new(id: String, kind: Kind, op: &str, body: Map<String, Value>) -> Envelope {
|
|
Envelope {
|
|
major: MAJOR,
|
|
minor: MINOR,
|
|
id,
|
|
reply_to: None,
|
|
kind,
|
|
op: op.to_owned(),
|
|
body,
|
|
attachments: Vec::new(),
|
|
}
|
|
}
|
|
|
|
/// Decodes and validates one frame's bytes.
|
|
pub fn decode(bytes: &[u8]) -> Result<Envelope, WireError> {
|
|
if bytes.is_empty() {
|
|
return err("empty envelope");
|
|
}
|
|
if bytes.len() > MAX_ENVELOPE_BYTES {
|
|
return err(format!(
|
|
"envelope of {} bytes exceeds {MAX_ENVELOPE_BYTES}",
|
|
bytes.len()
|
|
));
|
|
}
|
|
let v = parse_json_strict(bytes)?;
|
|
let mut f = Fields::new(&v, "envelope")?;
|
|
if f.string("protocol")? != PROTOCOL {
|
|
return err("envelope: protocol must be \"flybus\"");
|
|
}
|
|
let major = f.int("major", 0, MAX_CREDIT)?;
|
|
let minor = f.int("minor", 0, MAX_CREDIT)?;
|
|
let id = f.id("id")?;
|
|
let reply_to = f.nullable_id("replyTo")?;
|
|
let kind = Kind::parse(f.string("kind")?)
|
|
.ok_or_else(|| WireError("envelope: unknown kind".into()))?;
|
|
let op = f.string("op")?;
|
|
if !is_op(op) {
|
|
return err("envelope: op must be 1..=64 of [a-z.]");
|
|
}
|
|
let op = op.to_owned();
|
|
let body = f.object("body")?.clone();
|
|
let raw = f.array("attachments", 0, MAX_ATTACHMENTS)?;
|
|
let mut attachments = Vec::with_capacity(raw.len());
|
|
let mut names = HashSet::new();
|
|
for a in raw {
|
|
let a = Attachment::from_json(a)?;
|
|
if !names.insert(a.name.clone()) {
|
|
return err(format!("envelope: duplicate attachment name {:?}", a.name));
|
|
}
|
|
attachments.push(a);
|
|
}
|
|
f.finish()?;
|
|
Ok(Envelope {
|
|
major,
|
|
minor,
|
|
id,
|
|
reply_to,
|
|
kind,
|
|
op,
|
|
body,
|
|
attachments,
|
|
})
|
|
}
|
|
|
|
pub fn to_value(&self) -> Value {
|
|
let mut m = Map::new();
|
|
m.insert("protocol".into(), PROTOCOL.into());
|
|
m.insert("major".into(), self.major.into());
|
|
m.insert("minor".into(), self.minor.into());
|
|
m.insert("id".into(), self.id.clone().into());
|
|
m.insert(
|
|
"replyTo".into(),
|
|
self.reply_to.clone().map_or(Value::Null, Value::String),
|
|
);
|
|
m.insert("kind".into(), self.kind.as_str().into());
|
|
m.insert("op".into(), self.op.clone().into());
|
|
m.insert("body".into(), Value::Object(self.body.clone()));
|
|
m.insert(
|
|
"attachments".into(),
|
|
Value::Array(self.attachments.iter().map(Attachment::to_json).collect()),
|
|
);
|
|
Value::Object(m)
|
|
}
|
|
|
|
/// Serializes, refusing anything over [`MAX_ENVELOPE_BYTES`].
|
|
pub fn encode(&self) -> Result<Vec<u8>, WireError> {
|
|
let bytes = serde_json::to_vec(&self.to_value()).map_err(|e| WireError(e.to_string()))?;
|
|
if bytes.len() > MAX_ENVELOPE_BYTES {
|
|
return err(format!(
|
|
"envelope of {} bytes exceeds {MAX_ENVELOPE_BYTES}",
|
|
bytes.len()
|
|
));
|
|
}
|
|
Ok(bytes)
|
|
}
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------------------------
|
|
// Framing
|
|
|
|
#[derive(Debug)]
|
|
pub enum FrameError {
|
|
Io(std::io::Error),
|
|
/// A zero length prefix.
|
|
Empty,
|
|
/// The length prefix exceeds the limit; nothing was allocated for it.
|
|
TooLarge(u32),
|
|
/// End of stream inside a frame.
|
|
Truncated,
|
|
}
|
|
|
|
impl fmt::Display for FrameError {
|
|
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
|
|
match self {
|
|
FrameError::Io(e) => write!(f, "I/O error: {e}"),
|
|
FrameError::Empty => f.write_str("zero-length frame"),
|
|
FrameError::TooLarge(n) => write!(f, "frame of {n} bytes exceeds {MAX_ENVELOPE_BYTES}"),
|
|
FrameError::Truncated => f.write_str("stream ended inside a frame"),
|
|
}
|
|
}
|
|
}
|
|
|
|
impl std::error::Error for FrameError {}
|
|
|
|
/// Reads one frame. `Ok(None)` is a clean end of stream at a frame boundary. The length is
|
|
/// checked against [`MAX_ENVELOPE_BYTES`] before any buffer is allocated.
|
|
pub async fn read_frame<R: AsyncRead + Unpin>(r: &mut R) -> Result<Option<Vec<u8>>, FrameError> {
|
|
let mut len = [0u8; 4];
|
|
let mut got = 0;
|
|
while got < 4 {
|
|
let n = r.read(&mut len[got..]).await.map_err(FrameError::Io)?;
|
|
if n == 0 {
|
|
return if got == 0 {
|
|
Ok(None)
|
|
} else {
|
|
Err(FrameError::Truncated)
|
|
};
|
|
}
|
|
got += n;
|
|
}
|
|
let len = u32::from_le_bytes(len);
|
|
if len == 0 {
|
|
return Err(FrameError::Empty);
|
|
}
|
|
if len as usize > MAX_ENVELOPE_BYTES {
|
|
return Err(FrameError::TooLarge(len));
|
|
}
|
|
let mut buf = vec![0u8; len as usize];
|
|
r.read_exact(&mut buf).await.map_err(|e| match e.kind() {
|
|
std::io::ErrorKind::UnexpectedEof => FrameError::Truncated,
|
|
_ => FrameError::Io(e),
|
|
})?;
|
|
Ok(Some(buf))
|
|
}
|
|
|
|
/// Writes one frame: the length prefix and the bytes, as one buffer.
|
|
pub async fn write_frame<W: AsyncWrite + Unpin>(w: &mut W, bytes: &[u8]) -> std::io::Result<()> {
|
|
debug_assert!(!bytes.is_empty() && bytes.len() <= MAX_ENVELOPE_BYTES);
|
|
let mut buf = Vec::with_capacity(bytes.len() + 4);
|
|
buf.extend_from_slice(&(bytes.len() as u32).to_le_bytes());
|
|
buf.extend_from_slice(bytes);
|
|
w.write_all(&buf).await?;
|
|
w.flush().await
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
|
|
#[test]
|
|
fn scalars() {
|
|
assert!(is_id("a") && is_id("msg-7") && is_id("0.a_b-c"));
|
|
assert!(!is_id("") && !is_id("-a") && !is_id("A") && !is_id(&"a".repeat(65)));
|
|
assert!(is_id(&"a".repeat(64)));
|
|
assert_eq!(parse_u64("0"), Some(0));
|
|
assert_eq!(parse_u64("18446744073709551615"), Some(u64::MAX));
|
|
for bad in ["", "01", "18446744073709551616", "-1", "1.0", "+1", " 1"] {
|
|
assert_eq!(parse_u64(bad), None, "{bad:?}");
|
|
}
|
|
assert!(is_name("session.demo.snapshots") && is_name(&"a".repeat(192)));
|
|
for bad in ["", ".a", "a.", "a..b", "A", "a/b", &"a".repeat(193)] {
|
|
assert!(!is_name(bad), "{bad:?}");
|
|
}
|
|
assert!(is_digest(&"0f".repeat(32)) && !is_digest(&"0F".repeat(32)) && !is_digest("00"));
|
|
assert_eq!(parse_serial_id("msg", "msg-12"), Some(12));
|
|
assert_eq!(parse_serial_id("msg", "msg-012"), None);
|
|
assert_eq!(parse_serial_id("msg", "msgx-1"), None);
|
|
assert!(is_method("Counter.Increment") && !is_method("a\u{7f}") && !is_method(""));
|
|
}
|
|
|
|
#[test]
|
|
fn strict_json() {
|
|
assert!(parse_json_strict(br#"{"a":{"b":[{"c":1,"d":2}]}}"#).is_ok());
|
|
for bad in [
|
|
&br#"{"a":1,"a":2}"#[..],
|
|
br#"{"a":{"b":[{"c":1,"c":2}]}}"#,
|
|
br#"[{"x":1},{"y":{"z":1,"z":1}}]"#,
|
|
br#"{"a":NaN}"#,
|
|
br#"{"a":Infinity}"#,
|
|
br#"{"a":1e400}"#,
|
|
br#"{"a":"\ud800"}"#,
|
|
br#"{"a":1} x"#,
|
|
b"{\"a\":\"\xff\"}",
|
|
] {
|
|
assert!(
|
|
parse_json_strict(bad).is_err(),
|
|
"{:?}",
|
|
String::from_utf8_lossy(bad)
|
|
);
|
|
}
|
|
let deep = format!("{}{}", "[".repeat(200), "]".repeat(200));
|
|
assert!(parse_json_strict(deep.as_bytes()).is_err());
|
|
}
|
|
}
|