diff --git a/runtime/core/Cargo.toml b/runtime/core/Cargo.toml index 49a28ae..371cebb 100644 --- a/runtime/core/Cargo.toml +++ b/runtime/core/Cargo.toml @@ -8,7 +8,7 @@ license = "MPL-2.0" [features] default = ["std"] -# `contract` + `wire` need only `serde`'s core traits and no allocation. +# `contract` needs only `serde`'s core traits and no allocation. # `alloc` adds owned generated types, the frame `transport`, and the `serve` # loop. `std` adds the MessagePack `format` and the in-memory transport, plus # the dylib ABI. @@ -20,11 +20,12 @@ std = [ "dep:libloading", "dep:rmp", "dep:rmp-serde", + "dep:serde_json", ] [dependencies] -# `contract` / `wire` — no_std, no alloc +# `contract` — no_std, no alloc serde = { version = "1.0.190", default-features = false } bytemuck = { version = "1.14.0", features = ["derive"] } @@ -33,6 +34,7 @@ abi_stable = { version = "0.11.2", optional = true } libloading = { version = "0.8.0", optional = true } rmp = { version = "0.8.12", optional = true } rmp-serde = { version = "1.1.2", optional = true } +serde_json = { version = "1.0.108", optional = true, features = ["raw_value"] } [dev-dependencies] diff --git a/runtime/core/src/client.rs b/runtime/core/src/client.rs index 2f6ab8e..45ccd4d 100644 --- a/runtime/core/src/client.rs +++ b/runtime/core/src/client.rs @@ -1,17 +1,15 @@ //! Consumer-side calling: the mirror of [`Server`](crate::serve::Server). //! -//! A [`Client`] owns a [`Transport`] and a [`WireFormat`], hands out request -//! ids, and reuses its send / receive buffers across calls (§4.6 — no per-call -//! allocation on the frame path). One call is: encode the params straight into -//! the frame, send, block for the response, hand back the [`Envelope`] -//! borrowing the receive buffer. +//! A [`Client`] owns a [`Transport`], a [`WireFormat`] and a [`Framing`], hands +//! out request ids, and reuses its send / receive buffers across calls (§4.6 — +//! no per-call allocation on the frame path). One call is: hand the params to +//! the framing (which serialises them with the format), send, block for the +//! response, hand back the [`Envelope`] borrowing the receive buffer. //! //! The generated `Client` stub wraps this: it knows each function's -//! ordinal and its params / result / error types, so it calls [`Client::call`] -//! with a `u16` and turns the returned `(Envelope, &W)` into `Result>` — `Ok` payload decoded as `R`, an `Err` ordinal mapped through -//! its generated error table. This type stays protocol-agnostic; it does no -//! decoding, which is why `call` also hands back the format. +//! ordinal *and* name (it passes a [`Call`]), its params / result / error +//! types, and turns the returned `(Envelope, &W)` into `Result>`. use alloc::vec::Vec; @@ -19,93 +17,109 @@ use core::time::Duration; use serde::Serialize; -use crate::contract::{Envelope, Handshake, RuntimeError, WireFormat}; +use crate::contract::{ + Call, DatagramFraming, Envelope, Framing, Handshake, RuntimeError, WireFormat, +}; use crate::transport::Transport; -use crate::wire; -/// A calling endpoint bound to one transport. -pub struct Client { +/// A calling endpoint bound to one transport. Generic over the [`Framing`]; +/// defaults to the Comline datagram framing. +pub struct Client { transport: T, format: W, + framing: F, next_id: u64, request: Vec, response: Vec, } -impl Client { +impl Client { /// Bind a client to a transport **without a handshake** — "misaligned /// mode". Use when the peer can't take part (a legacy server); a /// wire-format or schema mismatch then surfaces later as a decode / framing /// error instead of up front. [`connect`](Self::connect) is the checked /// path. pub fn new(transport: T, format: W) -> Self { + Self::with_framing(transport, format, DatagramFraming) + } + + /// Bind + run the connection [`Handshake`] with the datagram framing. + pub fn connect(transport: T, format: W, local: Handshake) -> Result { + Self::connect_with_framing(transport, format, DatagramFraming, local) + } +} + +impl Client { + pub fn with_framing(transport: T, format: W, framing: F) -> Self { Self { transport, format, + framing, next_id: 0, request: Vec::new(), response: Vec::new(), } } - /// Bind a client and run the connection [`Handshake`]: send `local`, read - /// the peer's, and refuse (`RuntimeError::Handshake`) if they disagree on - /// schema hash, wire format, or framing. - pub fn connect(transport: T, format: W, local: Handshake) -> Result { - let mut client = Self::new(transport, format); - client.exchange_handshake(local)?; + /// [`with_framing`](Self::with_framing) + run the connection [`Handshake`]: + /// send `local`, read the peer's, refuse (`RuntimeError::Handshake`) on a + /// schema / wire-format / framing mismatch. + pub fn connect_with_framing( + transport: T, + format: W, + framing: F, + local: Handshake, + ) -> Result { + let mut client = Self::with_framing(transport, format, framing); + client.request.clear(); + local.encode(&mut client.request); + client.transport.send(&client.request)?; + client.response.clear(); + client.transport.recv(&mut client.response)?; + let peer = Handshake::decode(&client.response).ok_or(RuntimeError::Handshake)?; + local.check(&peer)?; Ok(client) } - fn exchange_handshake(&mut self, local: Handshake) -> Result<(), RuntimeError> { - self.request.clear(); - local.encode(&mut self.request); - self.transport.send(&self.request)?; - - self.response.clear(); - self.transport.recv(&mut self.response)?; - let peer = Handshake::decode(&self.response).ok_or(RuntimeError::Handshake)?; - local.check(&peer) - } - - /// Make call `call_id` with `params`, block for the response, and return - /// its [`Envelope`] (borrowing this client's receive buffer) together with - /// the format to decode it: `Ok(payload)` for the stub to read as `R`, or + /// Make `call` with `params`, block for the response, and return its + /// [`Envelope`] (borrowing this client's receive buffer) with the format + /// to decode it: `Ok(payload)` for the stub to read as `R`, or /// `Err { id, body }` to map through the generated error table. /// - /// The format rides along because both views come out of the same - /// `&mut self` borrow — the stub can't reach back into the client for it - /// while holding the envelope. + /// `call` is `impl Into` — a bare `u16` for datagram-only callers, + /// or `Call::new(id, name)` from a generated stub (a name-oriented framing + /// needs the name). /// /// Borrows `self` mutably for as long as the result is held: the previous - /// response must be decoded (or dropped) before the next call. A pipelined - /// / multiplexed client is a later, additive layer. - pub fn call

(&mut self, call_id: u16, params: &P) -> Result<(Envelope<'_>, &W), RuntimeError> + /// response must be decoded (or dropped) before the next call. + pub fn call(&mut self, call: C, params: &P) -> Result<(Envelope<'_>, &W), RuntimeError> where + C: Into, P: Serialize + ?Sized, { - self.request_response(call_id, params, None) + self.request_response(call.into(), params, None) } /// [`call`](Self::call), but give up after `timeout` waiting for the /// response — `Err(RuntimeError::Timeout)`. What a generated stub emits for /// a `@timeout_ms` function annotation. Honoured only by transports that /// override [`Transport::recv_timeout`] (the `std` ones do); others block. - pub fn call_with_timeout

( + pub fn call_with_timeout( &mut self, - call_id: u16, + call: C, params: &P, timeout: Duration, ) -> Result<(Envelope<'_>, &W), RuntimeError> where + C: Into, P: Serialize + ?Sized, { - self.request_response(call_id, params, Some(timeout)) + self.request_response(call.into(), params, Some(timeout)) } fn request_response

( &mut self, - call_id: u16, + call: Call, params: &P, timeout: Option, ) -> Result<(Envelope<'_>, &W), RuntimeError> @@ -116,8 +130,8 @@ impl Client { self.next_id = self.next_id.wrapping_add(1); self.request.clear(); - wire::encode_request_header(call_id, request_id, &mut self.request); - self.format.encode(params, &mut self.request)?; + self.framing + .encode_request(call, request_id, params, &self.format, &mut self.request)?; self.transport.send(&self.request)?; self.response.clear(); @@ -130,34 +144,38 @@ impl Client { } } - let (echoed, envelope) = - wire::decode_response(&self.response).ok_or(RuntimeError::Framing)?; + let (echoed, envelope) = self + .framing + .decode_response(&self.response) + .ok_or(RuntimeError::Framing)?; if echoed != request_id { // One outstanding call at a time, so a mismatched id is a stale or // corrupt frame, not reordering. return Err(RuntimeError::Framing); } - let envelope = Envelope::decode(envelope).ok_or(RuntimeError::Framing)?; Ok((envelope, &self.format)) } - /// Fire a **one-way** call: frame `call_id` + `params`, send, return. No - /// response is awaited — for `_return: None` schema functions, whose - /// generated dispatcher writes no [`Envelope`] and whose peer [`Server`] - /// therefore sends nothing back. `Ok(())` means the frame left the - /// transport, never a remote outcome. - pub fn notify

(&mut self, call_id: u16, params: &P) -> Result<(), RuntimeError> + /// Fire a **one-way** call: frame it, send, return. No response is awaited + /// — for `_return: None` schema functions, whose generated dispatcher + /// writes no [`Envelope`] and whose peer [`Server`] therefore sends nothing + /// back. `Ok(())` means the frame left the transport. + pub fn notify(&mut self, call: C, params: &P) -> Result<(), RuntimeError> where + C: Into, P: Serialize + ?Sized, { - // Keep request ids monotonic across mixed call / notify use, even - // though nothing reads this one back. let request_id = self.next_id; self.next_id = self.next_id.wrapping_add(1); self.request.clear(); - wire::encode_request_header(call_id, request_id, &mut self.request); - self.format.encode(params, &mut self.request)?; + self.framing.encode_request( + call.into(), + request_id, + params, + &self.format, + &mut self.request, + )?; self.transport.send(&self.request) } diff --git a/runtime/core/src/contract/dispatch.rs b/runtime/core/src/contract/dispatch.rs index 1385f5d..864ae27 100644 --- a/runtime/core/src/contract/dispatch.rs +++ b/runtime/core/src/contract/dispatch.rs @@ -28,23 +28,69 @@ impl Kind { } } +/// What a dispatched call produced. Framing-agnostic: the dispatcher calls +/// [`ok`](Reply::ok) or [`err`](Reply::err) exactly once, or neither for a +/// one-way call. The framing then wraps the accumulated body. +pub struct Reply<'a> { + body: &'a mut dyn BufMut, + outcome: Outcome, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Outcome { + /// One-way — nothing written, no response frame. + None, + /// A success payload was written. + Ok, + /// A schema error `body` was written, keyed by ordinal `id`. + Err(u16), +} + +impl<'a> Reply<'a> { + pub fn new(body: &'a mut dyn BufMut) -> Self { + Self { + body, + outcome: Outcome::None, + } + } + + /// Record a success: `payload` is the serialised return value. + pub fn ok(&mut self, payload: &[u8]) { + self.body.put_slice(payload); + self.outcome = Outcome::Ok; + } + + /// Record a raised schema error: `body` is the serialised error struct, + /// `id` its schema-global ordinal. + pub fn err(&mut self, id: u16, body: &[u8]) { + self.body.put_slice(body); + self.outcome = Outcome::Err(id); + } + + pub fn outcome(&self) -> Outcome { + self.outcome + } +} + /// The provider side of a protocol: given an inbound call and its encoded -/// params, run the user's handler and write the response [`Envelope`] into -/// `out`. +/// params, run the user's handler and record the outcome on `reply`. /// /// Sync — no boxed futures on the `no_std` path. An async server layer wraps /// this behind the `std` feature. The generated `Dispatcher` implements -/// it; the call system holds one by value / `&D` (generic — no vtable, no -/// `dyn`) and calls it with the format it was configured with. -/// -/// [`Envelope`]: crate::contract::Envelope +/// it; [`Server`](crate::serve::Server) holds one and routes inbound frames +/// to it. pub trait Dispatch { + /// The protocol's function names, in declaration order — so a name-oriented + /// framing's method name can be resolved to an ordinal. The generated + /// dispatcher returns its `_CALLS` constant. + fn calls(&self) -> &'static [&'static str]; + fn dispatch( &self, call: Kind, params: &[u8], format: &W, - out: &mut dyn BufMut, + reply: &mut Reply, ) -> Result<(), RuntimeError>; } diff --git a/runtime/core/src/contract/framing.rs b/runtime/core/src/contract/framing.rs new file mode 100644 index 0000000..714ec9b --- /dev/null +++ b/runtime/core/src/contract/framing.rs @@ -0,0 +1,178 @@ +use serde::Serialize; + +use crate::contract::{BufMut, Envelope, RuntimeError, WireFormat}; + +/// How a call is turned into bytes on the wire and back — the axis orthogonal +/// to [`WireFormat`] (which serialises the *parts*). The Comline datagram +/// framing ([`DatagramFraming`]) is one; JSON-RPC +/// (`comline_runtime::framing::JsonRpcFraming`) is another. +/// +/// [`Client`](crate::client::Client) and [`Server`](crate::serve::Server) are +/// generic over one, chosen at setup; the pair must agree (the connection +/// [`Handshake`](crate::contract::Handshake) carries `name()`, hashed). +pub trait Framing: Default { + /// A stable name, folded into the handshake. + fn name(&self) -> &'static str; + + /// Frame a request: the call, its id, and its params (serialised with + /// `fmt`, or wrapped by the framing — JSON-RPC nests them). + fn encode_request( + &self, + call: Call, + request_id: u64, + params: &P, + fmt: &W, + out: &mut dyn BufMut, + ) -> Result<(), RuntimeError> + where + W: WireFormat, + P: Serialize + ?Sized; + + /// Parse a request frame. The `params` slice borrows `frame` and is + /// independently decodable with the peer's `WireFormat`. + fn decode_request<'f>(&self, frame: &'f [u8]) -> Option>; + + /// Frame an `ok` response around an already-serialised `payload`. + fn encode_response_ok(&self, request_id: u64, payload: &[u8], out: &mut dyn BufMut); + + /// Frame an `err` response around an already-serialised error `body`, + /// keyed by its schema-global ordinal. + fn encode_response_err(&self, request_id: u64, id: u16, body: &[u8], out: &mut dyn BufMut); + + /// Parse a response frame into `(request_id, envelope)`; the envelope + /// borrows `frame`. + fn decode_response<'f>(&self, frame: &'f [u8]) -> Option<(u64, Envelope<'f>)>; +} + +/// A call to make, carrying *both* addresses so the framing picks: the +/// append-only ordinal for [`DatagramFraming`], the name for a name-oriented +/// framing. Generated stubs emit both; `From` covers hand-written +/// datagram-only callers (`name` is then `""`). +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct Call { + pub id: u16, + pub name: &'static str, +} + +impl Call { + pub const fn new(id: u16, name: &'static str) -> Self { + Self { id, name } + } +} + +impl From for Call { + fn from(id: u16) -> Self { + Self { id, name: "" } + } +} + +/// A decoded request. `call` is whichever address the framing put on the wire; +/// [`Server`](crate::serve::Server) resolves it to an index via +/// [`Dispatch::calls`](crate::contract::Dispatch::calls). +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct Request<'f> { + pub call: RequestCall<'f>, + pub request_id: u64, + pub params: &'f [u8], +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum RequestCall<'f> { + Id(u16), + Name(&'f str), +} + +// ── the Comline datagram framing ─────────────────────────────────────────── + +/// `no_std`, allocation-free. Request `[call_id u16 LE][request_id u64 LE] +/// [params…]`; response `[request_id u64 LE][envelope…]` where the envelope is +/// the tag-byte [`Envelope`] form. +#[derive(Debug, Default, Clone, Copy)] +pub struct DatagramFraming; + +impl Framing for DatagramFraming { + fn name(&self) -> &'static str { + crate::contract::FRAMING_DATAGRAM + } + + fn encode_request( + &self, + call: Call, + request_id: u64, + params: &P, + fmt: &W, + out: &mut dyn BufMut, + ) -> Result<(), RuntimeError> + where + W: WireFormat, + P: Serialize + ?Sized, + { + out.put_u16_le(call.id); + out.put_u64_le(request_id); + fmt.encode(params, out) + } + + fn decode_request<'f>(&self, frame: &'f [u8]) -> Option> { + let (head, params) = frame.split_at_checked(10)?; + let id = u16::from_le_bytes([head[0], head[1]]); + let request_id = u64::from_le_bytes(head[2..10].try_into().ok()?); + Some(Request { + call: RequestCall::Id(id), + request_id, + params, + }) + } + + fn encode_response_ok(&self, request_id: u64, payload: &[u8], out: &mut dyn BufMut) { + out.put_u64_le(request_id); + Envelope::encode_ok(payload, out); + } + + fn encode_response_err(&self, request_id: u64, id: u16, body: &[u8], out: &mut dyn BufMut) { + out.put_u64_le(request_id); + Envelope::encode_err(id, body, out); + } + + fn decode_response<'f>(&self, frame: &'f [u8]) -> Option<(u64, Envelope<'f>)> { + let (head, rest) = frame.split_at_checked(8)?; + let request_id = u64::from_le_bytes(head.try_into().ok()?); + Some((request_id, Envelope::decode(rest)?)) + } +} + +#[cfg(all(test, feature = "alloc"))] +mod tests { + use super::*; + use alloc::vec::Vec; + + #[test] + fn datagram_request_round_trips() { + // `encode_request`'s WireFormat leg is covered by the integration + // tests; here just the header + params split. + let mut frame = Vec::new(); + frame.extend_from_slice(&3u16.to_le_bytes()); + frame.extend_from_slice(&42u64.to_le_bytes()); + frame.extend_from_slice(b"args"); + + let req = DatagramFraming.decode_request(&frame).unwrap(); + assert_eq!(req.call, RequestCall::Id(3)); + assert_eq!(req.request_id, 42); + assert_eq!(req.params, b"args"); + assert_eq!(DatagramFraming.decode_request(&[0, 0, 0]), None); + } + + #[test] + fn datagram_response_round_trips() { + let f = DatagramFraming; + let mut ok = Vec::new(); + f.encode_response_ok(7, b"payload", &mut ok); + assert_eq!(f.decode_response(&ok), Some((7, Envelope::Ok(b"payload")))); + + let mut err = Vec::new(); + f.encode_response_err(7, 2, b"fields", &mut err); + assert_eq!( + f.decode_response(&err), + Some((7, Envelope::Err { id: 2, body: b"fields" })) + ); + } +} diff --git a/runtime/core/src/contract/mod.rs b/runtime/core/src/contract/mod.rs index 5192eb5..7d7e834 100644 --- a/runtime/core/src/contract/mod.rs +++ b/runtime/core/src/contract/mod.rs @@ -11,13 +11,15 @@ mod call; mod dispatch; mod envelope; mod error; +mod framing; mod handshake; mod wire; pub use buf::{BufMut, SliceBuf}; pub use call::CallError; -pub use dispatch::{Dispatch, Kind}; +pub use dispatch::{Dispatch, Kind, Outcome, Reply}; pub use envelope::Envelope; pub use error::RuntimeError; +pub use framing::{Call, DatagramFraming, Framing, Request, RequestCall}; pub use handshake::{name_hash, Handshake, FRAMING_DATAGRAM}; pub use wire::WireFormat; diff --git a/runtime/core/src/format/json.rs b/runtime/core/src/format/json.rs new file mode 100644 index 0000000..a9d3348 --- /dev/null +++ b/runtime/core/src/format/json.rs @@ -0,0 +1,56 @@ +use serde::{Deserialize, Serialize}; + +use super::BufWriter; +use crate::contract::{BufMut, RuntimeError, WireFormat}; + +/// [`WireFormat`] over JSON (`serde_json`). Verbose next to MessagePack, but +/// the pairing for [`JsonRpcFraming`](crate::framing::JsonRpcFraming) and handy +/// for debugging on the wire. +#[derive(Debug, Default, Clone, Copy)] +pub struct Json; + +impl WireFormat for Json { + fn name(&self) -> &'static str { + "json" + } + + fn encode( + &self, + value: &T, + out: &mut dyn BufMut, + ) -> Result<(), RuntimeError> { + serde_json::to_writer(BufWriter(out), value).map_err(|_| RuntimeError::Serialization) + } + + fn decode<'de, T: Deserialize<'de>>(&self, bytes: &'de [u8]) -> Result { + serde_json::from_slice(bytes).map_err(|_| RuntimeError::Serialization) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[derive(Serialize, Deserialize, PartialEq, Debug)] + struct Msg<'a> { + n: u32, + #[serde(borrow)] + s: &'a str, + } + + #[test] + fn round_trips_borrowed() { + let mut buf = Vec::new(); + Json.encode(&Msg { n: 7, s: "hi" }, &mut buf).unwrap(); + assert_eq!(buf, br#"{"n":7,"s":"hi"}"#); + + let back: Msg = Json.decode(&buf).unwrap(); + assert_eq!(back, Msg { n: 7, s: "hi" }); + } + + #[test] + fn garbage_is_a_serialization_error() { + let err = Json.decode::(b"not json").unwrap_err(); + assert_eq!(err, RuntimeError::Serialization); + } +} diff --git a/runtime/core/src/format/mod.rs b/runtime/core/src/format/mod.rs index 2473108..dad93b6 100644 --- a/runtime/core/src/format/mod.rs +++ b/runtime/core/src/format/mod.rs @@ -1,8 +1,29 @@ //! [`WireFormat`](crate::contract::WireFormat) implementations. //! -//! `std`-gated for now — `rmp-serde` needs `std::io`. A `no_std` MessagePack -//! path (hand-rolled on the `no_std` `rmp` crate) comes later. +//! `std`-gated for now — `rmp-serde` / `serde_json` want `std::io`. A `no_std` +//! path comes later. +use std::io::Write; + +use crate::contract::BufMut; + +mod json; mod msgpack; +pub use json::Json; pub use msgpack::MsgPack; + +/// `std::io::Write` over a [`BufMut`], so an `encode` serialises straight into +/// the caller's buffer with no intermediate `Vec`. +pub(crate) struct BufWriter<'a>(pub &'a mut dyn BufMut); + +impl Write for BufWriter<'_> { + fn write(&mut self, bytes: &[u8]) -> std::io::Result { + self.0.put_slice(bytes); + Ok(bytes.len()) + } + + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } +} diff --git a/runtime/core/src/format/msgpack.rs b/runtime/core/src/format/msgpack.rs index 7f612f3..d817ec6 100644 --- a/runtime/core/src/format/msgpack.rs +++ b/runtime/core/src/format/msgpack.rs @@ -1,7 +1,6 @@ -use std::io::Write; - use serde::{Deserialize, Serialize}; +use super::BufWriter; use crate::contract::{BufMut, RuntimeError, WireFormat}; /// [`WireFormat`] over MessagePack (`rmp-serde`): compact, `serde`-native, @@ -10,21 +9,6 @@ use crate::contract::{BufMut, RuntimeError, WireFormat}; #[derive(Debug, Default, Clone, Copy)] pub struct MsgPack; -/// `std::io::Write` over a `BufMut`, so `encode` serialises straight into the -/// caller's buffer with no intermediate `Vec`. -struct BufWriter<'a>(&'a mut dyn BufMut); - -impl Write for BufWriter<'_> { - fn write(&mut self, bytes: &[u8]) -> std::io::Result { - self.0.put_slice(bytes); - Ok(bytes.len()) - } - - fn flush(&mut self) -> std::io::Result<()> { - Ok(()) - } -} - impl WireFormat for MsgPack { fn name(&self) -> &'static str { "msgpack" diff --git a/runtime/core/src/framing.rs b/runtime/core/src/framing.rs new file mode 100644 index 0000000..65bc20c --- /dev/null +++ b/runtime/core/src/framing.rs @@ -0,0 +1,171 @@ +//! Alternative [`Framing`](crate::contract::Framing) implementations. +//! +//! The Comline datagram framing is `DatagramFraming` in `contract`; this module +//! holds ones that need `std` / extra deps. + +use serde::{Deserialize, Serialize}; +use serde_json::value::RawValue; + +use crate::contract::{ + BufMut, Call, Envelope, Framing, Request, RequestCall, RuntimeError, WireFormat, +}; + +/// [JSON-RPC 2.0](https://www.jsonrpc.org/specification) framing — a +/// **name-oriented**, human-readable frame. Pair it with +/// [`Json`](crate::format::Json). +/// +/// Request: `{"jsonrpc":"2.0","method":,"params":,"id":}` +/// Response: `{"jsonrpc":"2.0","result":,"id":}` or +/// `{"jsonrpc":"2.0","error":{"code":,"message":...,"data":},"id":}` +/// +/// A raised schema error maps to a JSON-RPC `error` object whose `code` is the +/// schema-global ordinal and whose `data` is the serialised error struct. +#[derive(Debug, Default, Clone, Copy)] +pub struct JsonRpcFraming; + +fn u64_bytes(n: u64) -> impl AsRef<[u8]> { + // small, no itoa dep + let s = n.to_string(); + s.into_bytes() +} + +impl Framing for JsonRpcFraming { + fn name(&self) -> &'static str { + "jsonrpc-2.0" + } + + fn encode_request( + &self, + call: Call, + request_id: u64, + params: &P, + fmt: &W, + out: &mut dyn BufMut, + ) -> Result<(), RuntimeError> + where + W: WireFormat, + P: Serialize + ?Sized, + { + // method names are generated identifiers — no JSON escaping needed. + out.put_slice(br#"{"jsonrpc":"2.0","method":""#); + out.put_slice(call.name.as_bytes()); + out.put_slice(br#"","params":"#); + fmt.encode(params, out)?; + out.put_slice(br#","id":"#); + out.put_slice(u64_bytes(request_id).as_ref()); + out.put_slice(b"}"); + Ok(()) + } + + fn decode_request<'f>(&self, frame: &'f [u8]) -> Option> { + #[derive(Deserialize)] + struct ReqIn<'a> { + method: &'a str, + #[serde(borrow, default)] + params: Option<&'a RawValue>, + #[serde(default)] + id: Option, + } + let r: ReqIn = serde_json::from_slice(frame).ok()?; + Some(Request { + call: RequestCall::Name(r.method), + request_id: r.id.unwrap_or(0), + params: r.params.map(|p| p.get().as_bytes()).unwrap_or(b"null"), + }) + } + + fn encode_response_ok(&self, request_id: u64, payload: &[u8], out: &mut dyn BufMut) { + out.put_slice(br#"{"jsonrpc":"2.0","result":"#); + out.put_slice(if payload.is_empty() { b"null" } else { payload }); + out.put_slice(br#","id":"#); + out.put_slice(u64_bytes(request_id).as_ref()); + out.put_slice(b"}"); + } + + fn encode_response_err(&self, request_id: u64, id: u16, body: &[u8], out: &mut dyn BufMut) { + out.put_slice(br#"{"jsonrpc":"2.0","error":{"code":"#); + out.put_slice(u64_bytes(u64::from(id)).as_ref()); + out.put_slice(br#","message":"application error","data":"#); + out.put_slice(if body.is_empty() { b"null" } else { body }); + out.put_slice(br#"},"id":"#); + out.put_slice(u64_bytes(request_id).as_ref()); + out.put_slice(b"}"); + } + + fn decode_response<'f>(&self, frame: &'f [u8]) -> Option<(u64, Envelope<'f>)> { + #[derive(Deserialize)] + struct ErrIn<'a> { + code: i64, + #[serde(borrow, default)] + data: Option<&'a RawValue>, + } + #[derive(Deserialize)] + struct RespIn<'a> { + #[serde(borrow, default)] + result: Option<&'a RawValue>, + #[serde(borrow, default)] + error: Option>, + id: u64, + } + let r: RespIn = serde_json::from_slice(frame).ok()?; + if let Some(e) = r.error { + Some(( + r.id, + Envelope::Err { + id: e.code as u16, + body: e.data.map(|d| d.get().as_bytes()).unwrap_or(b"null"), + }, + )) + } else { + Some(( + r.id, + Envelope::Ok(r.result.map(|p| p.get().as_bytes()).unwrap_or(b"null")), + )) + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::format::Json; + + #[test] + fn request_round_trips() { + let f = JsonRpcFraming; + let mut frame = Vec::new(); + f.encode_request(Call::new(0, "greet"), 1, &(7u32, "x"), &Json, &mut frame) + .unwrap(); + assert_eq!( + std::str::from_utf8(&frame).unwrap(), + r#"{"jsonrpc":"2.0","method":"greet","params":[7,"x"],"id":1}"# + ); + + let req = f.decode_request(&frame).unwrap(); + assert_eq!(req.call, RequestCall::Name("greet")); + assert_eq!(req.request_id, 1); + assert_eq!(req.params, br#"[7,"x"]"#); + } + + #[test] + fn ok_response_round_trips() { + let f = JsonRpcFraming; + let mut frame = Vec::new(); + f.encode_response_ok(9, br#"{"body":"hi"}"#, &mut frame); + assert_eq!( + f.decode_response(&frame), + Some((9, Envelope::Ok(br#"{"body":"hi"}"#))) + ); + } + + #[test] + fn err_response_carries_the_ordinal() { + let f = JsonRpcFraming; + let mut frame = Vec::new(); + f.encode_response_err(9, 3, br#"{"why":"no"}"#, &mut frame); + assert_eq!( + f.decode_response(&frame), + Some((9, Envelope::Err { id: 3, body: br#"{"why":"no"}"# })) + ); + } +} diff --git a/runtime/core/src/lib.rs b/runtime/core/src/lib.rs index d1706e1..99f1425 100644 --- a/runtime/core/src/lib.rs +++ b/runtime/core/src/lib.rs @@ -9,16 +9,19 @@ #[cfg(feature = "alloc")] extern crate alloc; -// The `core ↔ target` contract — `no_std`, allocation-free. +// The `core ↔ target` contract — `no_std`, allocation-free. The Comline +// datagram `Framing` lives here (`contract::DatagramFraming`). pub mod contract; -// Request / response framing — `no_std`, allocation-free. -pub mod wire; - // `WireFormat` implementations (`std`-gated for now — see the module). #[cfg(feature = "std")] pub mod format; +// Alternative `Framing` implementations that need `std` / extra deps +// (JSON-RPC). The datagram framing is in `contract`. +#[cfg(feature = "std")] +pub mod framing; + // The frame transport, the provider serve loop, and the consumer call side. #[cfg(feature = "alloc")] pub mod client; diff --git a/runtime/core/src/serve.rs b/runtime/core/src/serve.rs index 0d2e1e1..53ff89e 100644 --- a/runtime/core/src/serve.rs +++ b/runtime/core/src/serve.rs @@ -3,27 +3,38 @@ use alloc::vec::Vec; -use crate::contract::{Dispatch, Handshake, Kind, RuntimeError, WireFormat}; +use crate::contract::{ + DatagramFraming, Dispatch, Framing, Handshake, Kind, Outcome, Reply, RequestCall, RuntimeError, + WireFormat, +}; use crate::transport::Transport; -use crate::wire; /// Serves one protocol implementation over a [`Transport`], reusing its buffers -/// across calls (§4.6 — no per-call allocation on the frame path). -pub struct Server { +/// across calls (§4.6 — no per-call allocation on the frame path). Generic over +/// the [`Framing`]; defaults to the Comline datagram framing. +pub struct Server { dispatch: D, format: W, + framing: F, recv: Vec, - envelope: Vec, + body: Vec, response: Vec, } -impl Server { +impl Server { pub fn new(dispatch: D, format: W) -> Self { + Self::with_framing(dispatch, format, DatagramFraming) + } +} + +impl Server { + pub fn with_framing(dispatch: D, format: W, framing: F) -> Self { Self { dispatch, format, + framing, recv: Vec::new(), - envelope: Vec::new(), + body: Vec::new(), response: Vec::new(), } } @@ -35,23 +46,43 @@ impl Server { return Ok(false); } - let (call_id, request_id, params) = - wire::decode_request(&self.recv).ok_or(RuntimeError::Framing)?; + let req = self + .framing + .decode_request(&self.recv) + .ok_or(RuntimeError::Framing)?; + let request_id = req.request_id; - self.envelope.clear(); - self.dispatch - .dispatch(Kind::Id(call_id), params, &self.format, &mut self.envelope)?; + // Whatever address the framing carried, resolve it to an ordinal. + let idx = match req.call { + RequestCall::Id(id) => id, + RequestCall::Name(name) => self + .dispatch + .calls() + .iter() + .position(|c| *c == name) + .ok_or(RuntimeError::UnknownCall)? as u16, + }; - // A one-way call (`_return: None`): the generated dispatcher ran the - // handler and wrote no [`Envelope`] — there is nothing to reply. - // Any real envelope is at least one tag byte, so "empty" is - // unambiguous. - if self.envelope.is_empty() { - return Ok(true); - } + self.body.clear(); + let outcome = { + let mut reply = Reply::new(&mut self.body); + self.dispatch + .dispatch(Kind::Id(idx), req.params, &self.format, &mut reply)?; + reply.outcome() + }; self.response.clear(); - wire::encode_response(request_id, &self.envelope, &mut self.response); + match outcome { + // A one-way call (`_return: None`): nothing to reply. + Outcome::None => return Ok(true), + Outcome::Ok => self + .framing + .encode_response_ok(request_id, &self.body, &mut self.response), + Outcome::Err(id) => { + self.framing + .encode_response_err(request_id, id, &self.body, &mut self.response) + } + } transport.send(&self.response)?; Ok(true) } diff --git a/runtime/core/src/wire.rs b/runtime/core/src/wire.rs deleted file mode 100644 index 334923e..0000000 --- a/runtime/core/src/wire.rs +++ /dev/null @@ -1,88 +0,0 @@ -//! Compact request / response framing. -//! -//! - **Request** — `[call_id: u16 LE] [request_id: u64 LE] [params …]` -//! - **Response** — `[request_id: u64 LE] [envelope …]`, where the envelope is -//! the tag-byte form from [`Envelope`](crate::contract::Envelope). -//! -//! Datagram-oriented: one frame per message. A length-prefixed stream framing -//! for byte-stream transports (TCP) layers on top of this. - -use crate::contract::BufMut; - -const REQUEST_HEADER: usize = 2 + 8; -const RESPONSE_HEADER: usize = 8; - -/// Write a request frame around already-encoded params. -pub fn encode_request(call_id: u16, request_id: u64, params: &[u8], out: &mut dyn BufMut) { - encode_request_header(call_id, request_id, out); - out.put_slice(params); -} - -/// Write just the request header. A client follows it by serializing the params -/// straight into the same buffer — no params scratch buffer, one contiguous -/// frame. -pub fn encode_request_header(call_id: u16, request_id: u64, out: &mut dyn BufMut) { - out.put_u16_le(call_id); - out.put_u64_le(request_id); -} - -/// `(call_id, request_id, params)` — `params` borrows `frame`. `None` if the -/// frame is shorter than the header. -pub fn decode_request(frame: &[u8]) -> Option<(u16, u64, &[u8])> { - let (head, params) = frame.split_at_checked(REQUEST_HEADER)?; - let call_id = u16::from_le_bytes([head[0], head[1]]); - let request_id = u64::from_le_bytes(head[2..10].try_into().ok()?); - Some((call_id, request_id, params)) -} - -/// Write a response frame around an already-encoded envelope. -pub fn encode_response(request_id: u64, envelope: &[u8], out: &mut dyn BufMut) { - out.put_u64_le(request_id); - out.put_slice(envelope); -} - -/// `(request_id, envelope_bytes)` — the envelope borrows `frame`; parse it with -/// [`Envelope::decode`](crate::contract::Envelope::decode). `None` if truncated. -pub fn decode_response(frame: &[u8]) -> Option<(u64, &[u8])> { - let (head, envelope) = frame.split_at_checked(RESPONSE_HEADER)?; - let request_id = u64::from_le_bytes(head.try_into().ok()?); - Some((request_id, envelope)) -} - -#[cfg(all(test, feature = "alloc"))] -mod tests { - use super::*; - use alloc::vec::Vec; - - #[test] - fn request_round_trips() { - let mut frame = Vec::new(); - encode_request(3, 0x0102_0304_0506_0708, b"args", &mut frame); - assert_eq!(decode_request(&frame), Some((3, 0x0102_0304_0506_0708, &b"args"[..]))); - } - - #[test] - fn response_round_trips() { - let mut frame = Vec::new(); - encode_response(42, b"\x00payload", &mut frame); - assert_eq!(decode_response(&frame), Some((42, &b"\x00payload"[..]))); - } - - #[test] - fn truncated_frames_are_rejected() { - assert_eq!(decode_request(&[0, 0, 0]), None); - assert_eq!(decode_response(&[0, 0, 0]), None); - } - - #[test] - fn header_then_params_equals_whole_frame() { - let mut split = Vec::new(); - encode_request_header(7, 99, &mut split); - split.put_slice(b"body"); - - let mut whole = Vec::new(); - encode_request(7, 99, b"body", &mut whole); - - assert_eq!(split, whole); - } -} diff --git a/runtime/core/tests/client_roundtrip.rs b/runtime/core/tests/client_roundtrip.rs index ac8b121..77601e3 100644 --- a/runtime/core/tests/client_roundtrip.rs +++ b/runtime/core/tests/client_roundtrip.rs @@ -9,7 +9,7 @@ use std::thread; use comline_runtime::client::Client; use comline_runtime::contract::{ - BufMut, CallError, Dispatch, Envelope, Kind, RuntimeError, WireFormat, + CallError, Dispatch, Envelope, Kind, Reply, RuntimeError, WireFormat, }; use comline_runtime::format::MsgPack; use comline_runtime::serve::Server; @@ -49,26 +49,30 @@ trait Greet { struct GreetDispatcher(T); impl Dispatch for GreetDispatcher { + fn calls(&self) -> &'static [&'static str] { + CALLS + } + fn dispatch( &self, call: Kind, params: &[u8], fmt: &W, - out: &mut dyn BufMut, + reply: &mut Reply, ) -> Result<(), RuntimeError> { match call.resolve(CALLS).ok_or(RuntimeError::UnknownCall)? { 0 => { let p: HelloParams = fmt.decode(params)?; match self.0.hello(p.name) { - Ok(reply) => { + Ok(r) => { let mut body = Vec::new(); - fmt.encode(&reply, &mut body)?; - Envelope::encode_ok(&body, out); + fmt.encode(&r, &mut body)?; + reply.ok(&body); } Err(GreetHelloError::Rude(e)) => { let mut body = Vec::new(); fmt.encode(&e, &mut body)?; - Envelope::encode_err(ERR_RUDE, &body, out); + reply.err(ERR_RUDE, &body); } } Ok(()) diff --git a/runtime/core/tests/dispatch_roundtrip.rs b/runtime/core/tests/dispatch_roundtrip.rs index 93e58c3..9c6c01c 100644 --- a/runtime/core/tests/dispatch_roundtrip.rs +++ b/runtime/core/tests/dispatch_roundtrip.rs @@ -1,12 +1,11 @@ -//! End to end, no network: a hand-written stand-in for what `comline-rust` will -//! generate — a client stub and a `Dispatch` impl — driven directly with -//! `MsgPack`. Proves the `contract` surface (`Kind`, `WireFormat`, `Dispatch`, -//! `Envelope`, `BufMut`, `CallError`) fits together before any codegen or -//! `setup/` rework. +//! End to end, no network: a hand-written stand-in for what `comline-rust` +//! generates — a client stub and a `Dispatch` impl — driven directly with +//! `MsgPack`, no framing. Proves the `contract` surface (`Kind`, `WireFormat`, +//! `Dispatch`, `Reply`, `Outcome`, `CallError`) fits together. #![cfg(feature = "std")] use comline_runtime::contract::{ - BufMut, CallError, Dispatch, Envelope, Kind, RuntimeError, WireFormat, + CallError, Dispatch, Kind, Outcome, Reply, RuntimeError, WireFormat, }; use comline_runtime::format::MsgPack; use serde::{Deserialize, Serialize}; @@ -52,38 +51,40 @@ trait Echo { struct EchoDispatcher(T); impl Dispatch for EchoDispatcher { + fn calls(&self) -> &'static [&'static str] { + CALLS + } + fn dispatch( &self, call: Kind, params: &[u8], fmt: &W, - out: &mut dyn BufMut, + reply: &mut Reply, ) -> Result<(), RuntimeError> { - // A real dispatcher reuses one scratch buffer; a fresh `Vec` per arm - // keeps the shape readable here. match call.resolve(CALLS).ok_or(RuntimeError::UnknownCall)? { 0 => { let p: SayParams = fmt.decode(params)?; match self.0.say(p.msg) { - Ok(reply) => { + Ok(r) => { let mut body = Vec::new(); - fmt.encode(&reply, &mut body)?; - Envelope::encode_ok(&body, out); + fmt.encode(&r, &mut body)?; + reply.ok(&body); } Err(SayError::TooLong(e)) => { let mut body = Vec::new(); fmt.encode(&e, &mut body)?; - Envelope::encode_err(ERR_TOO_LONG, &body, out); + reply.err(ERR_TOO_LONG, &body); } } Ok(()) } 1 => { let p: BumpParams = fmt.decode(params)?; - let reply = self.0.bump(p.n).unwrap(); + let r = self.0.bump(p.n).unwrap(); let mut body = Vec::new(); - fmt.encode(&reply, &mut body)?; - Envelope::encode_ok(&body, out); + fmt.encode(&r, &mut body)?; + reply.ok(&body); Ok(()) } _ => Err(RuntimeError::UnknownCall), @@ -91,10 +92,10 @@ impl Dispatch for EchoDispatcher { } } -// ── consumer: the generated client stub ──────────────────────────────────── +// ── consumer: the generated client stub (no framing — talks to the dispatcher) ─ struct EchoClient<'d, D> { - dispatcher: &'d D, // stands in for a call system + transport + dispatcher: &'d D, fmt: MsgPack, } @@ -103,20 +104,19 @@ impl EchoClient<'_, D> { let mut params = Vec::new(); self.fmt.encode(&SayParams { msg }, &mut params)?; - let mut frame = Vec::new(); + let mut body = Vec::new(); + let mut reply = Reply::new(&mut body); self.dispatcher - .dispatch(Kind::Id(0), ¶ms, &self.fmt, &mut frame)?; - - match Envelope::decode(&frame).ok_or(RuntimeError::Framing)? { - Envelope::Ok(payload) => self.fmt.decode(payload).map_err(CallError::Runtime), - Envelope::Err { - id: ERR_TOO_LONG, - body, - } => { - let e: TooLong = self.fmt.decode(body)?; + .dispatch(Kind::Id(0), ¶ms, &self.fmt, &mut reply)?; + + match reply.outcome() { + Outcome::Ok => self.fmt.decode(&body).map_err(CallError::Runtime), + Outcome::Err(ERR_TOO_LONG) => { + let e: TooLong = self.fmt.decode(&body)?; Err(CallError::App(SayError::TooLong(e))) } - Envelope::Err { id, .. } => Err(CallError::Runtime(RuntimeError::Remote { id })), + Outcome::Err(id) => Err(CallError::Runtime(RuntimeError::Remote { id })), + Outcome::None => Err(CallError::Runtime(RuntimeError::Framing)), } } @@ -124,22 +124,24 @@ impl EchoClient<'_, D> { let mut params = Vec::new(); self.fmt.encode(&BumpParams { n }, &mut params)?; - let mut frame = Vec::new(); + let mut body = Vec::new(); + let mut reply = Reply::new(&mut body); self.dispatcher - .dispatch(Kind::Id(1), ¶ms, &self.fmt, &mut frame)?; + .dispatch(Kind::Id(1), ¶ms, &self.fmt, &mut reply)?; - match Envelope::decode(&frame).ok_or(RuntimeError::Framing)? { - Envelope::Ok(payload) => self.fmt.decode(payload).map_err(CallError::Runtime), - Envelope::Err { id, .. } => Err(CallError::Runtime(RuntimeError::Remote { id })), + match reply.outcome() { + Outcome::Ok => self.fmt.decode(&body).map_err(CallError::Runtime), + Outcome::Err(id) => Err(CallError::Runtime(RuntimeError::Remote { id })), + Outcome::None => Err(CallError::Runtime(RuntimeError::Framing)), } } } // ── the service and the assertions ──────────────────────────────────────── -struct Server; +struct Service; -impl Echo for Server { +impl Echo for Service { fn say(&self, msg: &str) -> Result { if msg.len() > 8 { return Err(SayError::TooLong(TooLong { limit: 8 })); @@ -152,17 +154,14 @@ impl Echo for Server { } } -fn client() -> (EchoDispatcher, MsgPack) { - (EchoDispatcher(Server), MsgPack) +fn client() -> (EchoDispatcher, MsgPack) { + (EchoDispatcher(Service), MsgPack) } #[test] fn ok_path_round_trips() { let (d, fmt) = client(); - let c = EchoClient { - dispatcher: &d, - fmt, - }; + let c = EchoClient { dispatcher: &d, fmt }; assert_eq!(c.say("hi").unwrap(), "echo: hi"); assert_eq!(c.bump(41).unwrap(), 42); } @@ -170,10 +169,7 @@ fn ok_path_round_trips() { #[test] fn a_raised_error_reaches_the_client_typed() { let (d, fmt) = client(); - let c = EchoClient { - dispatcher: &d, - fmt, - }; + let c = EchoClient { dispatcher: &d, fmt }; let err = c.say("this is far too long").unwrap_err(); assert_eq!(err, CallError::App(SayError::TooLong(TooLong { limit: 8 }))); } @@ -181,9 +177,10 @@ fn a_raised_error_reaches_the_client_typed() { #[test] fn an_unknown_call_ordinal_is_a_runtime_error() { let (d, _) = client(); - let mut out = Vec::new(); + let mut body = Vec::new(); + let mut reply = Reply::new(&mut body); let err = d - .dispatch(Kind::Id(9), &[], &MsgPack, &mut out) + .dispatch(Kind::Id(9), &[], &MsgPack, &mut reply) .unwrap_err(); assert_eq!(err, RuntimeError::UnknownCall); } diff --git a/runtime/core/tests/handshake_roundtrip.rs b/runtime/core/tests/handshake_roundtrip.rs index 0fc5df6..ad966e5 100644 --- a/runtime/core/tests/handshake_roundtrip.rs +++ b/runtime/core/tests/handshake_roundtrip.rs @@ -7,7 +7,7 @@ use std::thread; use comline_runtime::client::Client; use comline_runtime::contract::{ - BufMut, Dispatch, Envelope, Handshake, Kind, RuntimeError, WireFormat, FRAMING_DATAGRAM, + Dispatch, Envelope, Handshake, Kind, Reply, RuntimeError, WireFormat, FRAMING_DATAGRAM, }; use comline_runtime::format::MsgPack; use comline_runtime::serve::Server; @@ -30,19 +30,23 @@ trait Echo { struct EchoDispatcher(T); impl Dispatch for EchoDispatcher { + fn calls(&self) -> &'static [&'static str] { + CALLS + } + fn dispatch( &self, call: Kind, params: &[u8], fmt: &W, - out: &mut dyn BufMut, + reply: &mut Reply, ) -> Result<(), RuntimeError> { match call.resolve(CALLS).ok_or(RuntimeError::UnknownCall)? { 0 => { let p: BumpParams = fmt.decode(params)?; let mut body = Vec::new(); fmt.encode(&self.0.bump(p.n), &mut body)?; - Envelope::encode_ok(&body, out); + reply.ok(&body); Ok(()) } _ => Err(RuntimeError::UnknownCall), diff --git a/runtime/core/tests/jsonrpc_roundtrip.rs b/runtime/core/tests/jsonrpc_roundtrip.rs new file mode 100644 index 0000000..d6ac556 --- /dev/null +++ b/runtime/core/tests/jsonrpc_roundtrip.rs @@ -0,0 +1,155 @@ +//! The framing axis is pluggable: the exact same `Client` / `Server` / dispatch +//! code, over `JsonRpcFraming` + `Json` instead of the datagram default. A call, +//! a raised typed error, and a peek at the actual bytes (they're JSON-RPC 2.0). +#![cfg(feature = "std")] + +use std::thread; + +use comline_runtime::client::Client; +use comline_runtime::contract::{ + Call, CallError, Dispatch, Envelope, Kind, Reply, RuntimeError, WireFormat, +}; +use comline_runtime::format::Json; +use comline_runtime::framing::JsonRpcFraming; +use comline_runtime::serve::Server; +use comline_runtime::transport::{duplex, Transport}; +use serde::{Deserialize, Serialize}; + +// protocol Greet { function hello(name: str) -> str ! Rude; } + +#[derive(Serialize, Deserialize)] +struct HelloParams<'a> { + #[serde(borrow)] + name: &'a str, +} + +#[derive(Serialize, Deserialize, PartialEq, Eq, Debug)] +struct Rude { + reason: String, +} + +const CALLS: &[&str] = &["hello"]; +const ERR_RUDE: u16 = 0; + +#[derive(Debug, PartialEq, Eq)] +enum GreetHelloError { + Rude(Rude), +} + +trait Greet { + fn hello(&self, name: &str) -> Result; +} + +struct GreetDispatcher(T); + +impl Dispatch for GreetDispatcher { + fn calls(&self) -> &'static [&'static str] { + CALLS + } + + fn dispatch( + &self, + call: Kind, + params: &[u8], + fmt: &W, + reply: &mut Reply, + ) -> Result<(), RuntimeError> { + match call.resolve(CALLS).ok_or(RuntimeError::UnknownCall)? { + 0 => { + let p: HelloParams = fmt.decode(params)?; + match self.0.hello(p.name) { + Ok(r) => { + let mut body = Vec::new(); + fmt.encode(&r, &mut body)?; + reply.ok(&body); + } + Err(GreetHelloError::Rude(e)) => { + let mut body = Vec::new(); + fmt.encode(&e, &mut body)?; + reply.err(ERR_RUDE, &body); + } + } + Ok(()) + } + _ => Err(RuntimeError::UnknownCall), + } + } +} + +struct Polite; +impl Greet for Polite { + fn hello(&self, name: &str) -> Result { + if name.is_empty() { + return Err(GreetHelloError::Rude(Rude { + reason: "no name".into(), + })); + } + Ok(format!("hi, {name}")) + } +} + +/// Stand-in for the generated `GreetClient` — note it passes `Call::new(id, +/// name)`; the name is what a name-oriented framing needs. +struct GreetClient(Client); + +impl GreetClient +where + T: Transport, + W: WireFormat, + F: comline_runtime::contract::Framing, +{ + fn hello(&mut self, name: &str) -> Result> { + let (reply, fmt) = self.0.call(Call::new(0, "hello"), &HelloParams { name })?; + match reply { + Envelope::Ok(payload) => fmt.decode(payload).map_err(CallError::Runtime), + Envelope::Err { id: ERR_RUDE, body } => { + let e: Rude = fmt.decode(body)?; + Err(CallError::App(GreetHelloError::Rude(e))) + } + Envelope::Err { id, .. } => Err(CallError::Runtime(RuntimeError::Remote { id })), + } + } +} + +#[test] +fn a_call_and_an_error_round_trip_over_json_rpc() { + let (client_side, provider_side) = duplex(); + + let provider = thread::spawn(move || { + let mut provider_side = provider_side; + Server::with_framing(GreetDispatcher(Polite), Json, JsonRpcFraming) + .serve(&mut provider_side) + .unwrap(); + }); + + let mut client = GreetClient(Client::with_framing(client_side, Json, JsonRpcFraming)); + + assert_eq!(client.hello("world").unwrap(), "hi, world"); + assert_eq!( + client.hello("").unwrap_err(), + CallError::App(GreetHelloError::Rude(Rude { + reason: "no name".into(), + })), + ); + + drop(client); + provider.join().unwrap(); +} + +#[test] +fn the_bytes_on_the_wire_are_json_rpc() { + let (a, mut b) = duplex(); + let mut client = Client::with_framing(a, Json, JsonRpcFraming); + + // fire a request; don't wait for a reply (nobody serving `b` yet) + client + .notify(Call::new(0, "hello"), &HelloParams { name: "x" }) + .unwrap(); + + let mut frame = Vec::new(); + b.recv(&mut frame).unwrap(); + assert_eq!( + std::str::from_utf8(&frame).unwrap(), + r#"{"jsonrpc":"2.0","method":"hello","params":{"name":"x"},"id":0}"# + ); +} diff --git a/runtime/core/tests/oneway_roundtrip.rs b/runtime/core/tests/oneway_roundtrip.rs index df4ce76..5f761d4 100644 --- a/runtime/core/tests/oneway_roundtrip.rs +++ b/runtime/core/tests/oneway_roundtrip.rs @@ -8,7 +8,7 @@ use std::cell::RefCell; use std::rc::Rc; use comline_runtime::client::Client; -use comline_runtime::contract::{BufMut, Dispatch, Kind, RuntimeError, WireFormat}; +use comline_runtime::contract::{Dispatch, Kind, Reply, RuntimeError, WireFormat}; use comline_runtime::format::MsgPack; use comline_runtime::serve::Server; use comline_runtime::transport::duplex; @@ -31,12 +31,16 @@ trait Log { struct LogDispatcher(T); impl Dispatch for LogDispatcher { + fn calls(&self) -> &'static [&'static str] { + CALLS + } + fn dispatch( &self, call: Kind, params: &[u8], fmt: &W, - _out: &mut dyn BufMut, // one-way: nothing is written here + _reply: &mut Reply, // one-way: nothing is recorded ) -> Result<(), RuntimeError> { match call.resolve(CALLS).ok_or(RuntimeError::UnknownCall)? { 0 => { diff --git a/runtime/core/tests/serve_roundtrip.rs b/runtime/core/tests/serve_roundtrip.rs index 9aa836d..5e75e45 100644 --- a/runtime/core/tests/serve_roundtrip.rs +++ b/runtime/core/tests/serve_roundtrip.rs @@ -1,15 +1,17 @@ -//! A call over a real (in-process) transport: client frames a request, the -//! `Server` on another thread reads it, dispatches, and frames the response. -//! Exercises `wire` + `transport::InMemory` + `serve::Server` together. +//! A call over a real (in-process) transport: the client hand-frames a request +//! with `DatagramFraming`, the `Server` on another thread reads it, dispatches, +//! and frames the response. Exercises `Framing` + `transport::InMemory` + +//! `serve::Server` together. #![cfg(feature = "std")] use std::thread; -use comline_runtime::contract::{BufMut, Dispatch, Envelope, Kind, RuntimeError, WireFormat}; +use comline_runtime::contract::{ + Call, DatagramFraming, Dispatch, Envelope, Framing, Kind, Reply, RuntimeError, WireFormat, +}; use comline_runtime::format::MsgPack; use comline_runtime::serve::Server; use comline_runtime::transport::{duplex, Transport}; -use comline_runtime::wire; use serde::{Deserialize, Serialize}; // protocol Greet { function hello(name: str) -> str; } @@ -29,20 +31,24 @@ trait Greet { struct GreetDispatcher(T); impl Dispatch for GreetDispatcher { + fn calls(&self) -> &'static [&'static str] { + CALLS + } + fn dispatch( &self, call: Kind, params: &[u8], fmt: &W, - out: &mut dyn BufMut, + reply: &mut Reply, ) -> Result<(), RuntimeError> { match call.resolve(CALLS).ok_or(RuntimeError::UnknownCall)? { 0 => { let p: HelloParams = fmt.decode(params)?; - let reply = self.0.hello(p.name); + let r = self.0.hello(p.name); let mut body = Vec::new(); - fmt.encode(&reply, &mut body)?; - Envelope::encode_ok(&body, out); + fmt.encode(&r, &mut body)?; + reply.ok(&body); Ok(()) } _ => Err(RuntimeError::UnknownCall), @@ -69,21 +75,25 @@ fn a_call_round_trips_over_the_transport() { }); // client: frame `hello("world")` as request #1 - let mut params = Vec::new(); - MsgPack - .encode(&HelloParams { name: "world" }, &mut params) - .unwrap(); let mut request = Vec::new(); - wire::encode_request(0, 1, ¶ms, &mut request); + DatagramFraming + .encode_request( + Call::from(0), + 1, + &HelloParams { name: "world" }, + &MsgPack, + &mut request, + ) + .unwrap(); client.send(&request).unwrap(); // client: read the response let mut frame = Vec::new(); client.recv(&mut frame).unwrap(); - let (request_id, envelope) = wire::decode_response(&frame).unwrap(); + let (request_id, envelope) = DatagramFraming.decode_response(&frame).unwrap(); assert_eq!(request_id, 1); - let reply: String = match Envelope::decode(envelope).unwrap() { + let reply: String = match envelope { Envelope::Ok(payload) => MsgPack.decode(payload).unwrap(), Envelope::Err { .. } => panic!("unexpected error frame"), }; diff --git a/runtime/core/tests/timeout_roundtrip.rs b/runtime/core/tests/timeout_roundtrip.rs index ac34e4e..d9b6889 100644 --- a/runtime/core/tests/timeout_roundtrip.rs +++ b/runtime/core/tests/timeout_roundtrip.rs @@ -8,7 +8,7 @@ use std::thread; use std::time::Duration; use comline_runtime::client::Client; -use comline_runtime::contract::{BufMut, Dispatch, Envelope, Kind, RuntimeError, WireFormat}; +use comline_runtime::contract::{Dispatch, Envelope, Kind, Reply, RuntimeError, WireFormat}; use comline_runtime::format::MsgPack; use comline_runtime::serve::Server; use comline_runtime::transport::duplex; @@ -30,19 +30,23 @@ trait Echo { struct EchoDispatcher(T); impl Dispatch for EchoDispatcher { + fn calls(&self) -> &'static [&'static str] { + CALLS + } + fn dispatch( &self, call: Kind, params: &[u8], fmt: &W, - out: &mut dyn BufMut, + reply: &mut Reply, ) -> Result<(), RuntimeError> { match call.resolve(CALLS).ok_or(RuntimeError::UnknownCall)? { 0 => { let p: PingParams = fmt.decode(params)?; let mut body = Vec::new(); fmt.encode(&self.0.ping(p.n), &mut body)?; - Envelope::encode_ok(&body, out); + reply.ok(&body); Ok(()) } _ => Err(RuntimeError::UnknownCall),