From e2dedcd53d8695ac022c4e4bcaf6ea3e49f3a66d Mon Sep 17 00:00:00 2001 From: Kinflou Date: Wed, 2 Sep 2026 01:34:43 +0800 Subject: [PATCH] feat: one-way calls -- Client::notify + Server skips the empty reply (7f) For `_return: None` schema functions: fire-and-forget, no response frame. - Client::notify

(call_id, &P) -> Result<(), RuntimeError> -- frames and sends a request, returns without a recv. Request ids stay monotonic. - Server::serve_one -- after dispatch, if the (generated) dispatcher wrote no Envelope, there is nothing to reply: skip encode_response + send. Any real envelope is >= 1 tag byte, so "empty" is unambiguous. No wire or contract change. - InMemory::try_recv -- non-blocking receive, for single-threaded pumping and for asserting a one-way call drew no reply. tests/oneway_roundtrip.rs: a Log { record(line: str); } stand-in -- client notifies twice, the server pumps both and replies to neither, try_recv confirms the client's side is silent. --- README.md | 1 + runtime/README.md | 3 - runtime/core/src/client.rs | 20 +++++++ runtime/core/src/serve.rs | 8 +++ runtime/core/src/transport.rs | 18 ++++++ runtime/core/tests/oneway_roundtrip.rs | 83 ++++++++++++++++++++++++++ 6 files changed, 130 insertions(+), 3 deletions(-) create mode 100644 README.md create mode 100644 runtime/core/tests/oneway_roundtrip.rs diff --git a/README.md b/README.md new file mode 100644 index 0000000..8d1c8b6 --- /dev/null +++ b/README.md @@ -0,0 +1 @@ + diff --git a/runtime/README.md b/runtime/README.md index ea2b339..7f55993 100644 --- a/runtime/README.md +++ b/runtime/README.md @@ -27,7 +27,4 @@ CBOR ## Consideration of Features -https://discord.com/channels/619623572318453784/737119153282089109/1194612976985055243 https://capnproto.org/news/2013-12-12-capnproto-0.4-time-travel.html - -Suggestion by Cat diff --git a/runtime/core/src/client.rs b/runtime/core/src/client.rs index c15ecfc..04c4b84 100644 --- a/runtime/core/src/client.rs +++ b/runtime/core/src/client.rs @@ -79,6 +79,26 @@ impl Client { 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> + where + 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.transport.send(&self.request) + } + /// The underlying transport, e.g. to close it or read its peer address. pub fn transport_mut(&mut self) -> &mut T { &mut self.transport diff --git a/runtime/core/src/serve.rs b/runtime/core/src/serve.rs index 18939d5..f4a03d6 100644 --- a/runtime/core/src/serve.rs +++ b/runtime/core/src/serve.rs @@ -42,6 +42,14 @@ impl Server { self.dispatch .dispatch(Kind::Id(call_id), params, &self.format, &mut self.envelope)?; + // 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.response.clear(); wire::encode_response(request_id, &self.envelope, &mut self.response); transport.send(&self.response)?; diff --git a/runtime/core/src/transport.rs b/runtime/core/src/transport.rs index 46d36eb..b67b4fb 100644 --- a/runtime/core/src/transport.rs +++ b/runtime/core/src/transport.rs @@ -41,6 +41,24 @@ mod in_memory { (InMemory { tx: a_tx, rx: b_rx }, InMemory { tx: b_tx, rx: a_rx }) } + impl InMemory { + /// Non-blocking receive: `Ok(true)` if a frame was read into `buf`, + /// `Ok(false)` if the peer has sent nothing (yet). For single-threaded + /// pumping, and for asserting a one-way call drew no reply. + pub fn try_recv(&mut self, buf: &mut Vec) -> Result { + use std::sync::mpsc::TryRecvError; + match self.rx.try_recv() { + Ok(frame) => { + buf.clear(); + buf.extend_from_slice(&frame); + Ok(true) + } + Err(TryRecvError::Empty) => Ok(false), + Err(TryRecvError::Disconnected) => Err(RuntimeError::Transport), + } + } + } + impl Transport for InMemory { fn send(&mut self, frame: &[u8]) -> Result<(), RuntimeError> { self.tx diff --git a/runtime/core/tests/oneway_roundtrip.rs b/runtime/core/tests/oneway_roundtrip.rs new file mode 100644 index 0000000..df4ce76 --- /dev/null +++ b/runtime/core/tests/oneway_roundtrip.rs @@ -0,0 +1,83 @@ +//! A one-way call (`_return: None`): the client `notify`s, the provider's +//! dispatcher runs the handler but writes no `Envelope`, and the `Server` +//! sends nothing back. Hand-written stand-in for what `comline-rust` emits +//! for a no-return `function`. +#![cfg(feature = "std")] + +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::format::MsgPack; +use comline_runtime::serve::Server; +use comline_runtime::transport::duplex; +use serde::{Deserialize, Serialize}; + +// protocol Log { function record(line: str); } // no `->` : one-way + +#[derive(Serialize, Deserialize)] +struct RecordParams<'a> { + #[serde(borrow)] + line: &'a str, +} + +const CALLS: &[&str] = &["record"]; + +trait Log { + fn record(&self, line: &str); +} + +struct LogDispatcher(T); + +impl Dispatch for LogDispatcher { + fn dispatch( + &self, + call: Kind, + params: &[u8], + fmt: &W, + _out: &mut dyn BufMut, // one-way: nothing is written here + ) -> Result<(), RuntimeError> { + match call.resolve(CALLS).ok_or(RuntimeError::UnknownCall)? { + 0 => { + let p: RecordParams = fmt.decode(params)?; + self.0.record(p.line); + Ok(()) + } + _ => Err(RuntimeError::UnknownCall), + } + } +} + +struct Recorder(Rc>>); +impl Log for Recorder { + fn record(&self, line: &str) { + self.0.borrow_mut().push(line.to_string()); + } +} + +#[test] +fn a_one_way_call_reaches_the_handler_and_draws_no_reply() { + let (client_side, mut provider_side) = duplex(); + let log = Rc::new(RefCell::new(Vec::new())); + let mut server = Server::new(LogDispatcher(Recorder(log.clone())), MsgPack); + let mut client = Client::new(client_side, MsgPack); + + client + .notify(0, &RecordParams { line: "first" }) + .unwrap(); + client + .notify(0, &RecordParams { line: "second" }) + .unwrap(); + + // Two frames queued; the server pumps both, replying to neither. + assert!(server.serve_one(&mut provider_side).unwrap()); + assert!(server.serve_one(&mut provider_side).unwrap()); + assert_eq!(&*log.borrow(), &["first".to_string(), "second".to_string()]); + + let mut buf = Vec::new(); + assert!( + !client.transport_mut().try_recv(&mut buf).unwrap(), + "a one-way call must not produce a response frame", + ); +}