diff --git a/runtime/core/Cargo.toml b/runtime/core/Cargo.toml index 530cc42..467cdaf 100644 --- a/runtime/core/Cargo.toml +++ b/runtime/core/Cargo.toml @@ -6,49 +6,34 @@ edition = "2021" [features] -default = ["std", "json_rpc"] +default = ["std"] -# `contract` needs only `serde`'s core traits. `alloc` adds owned generated -# types and `Vec`-backed buffers; `std` adds the transport / async / setup -# layer (`setup`, `package_abi`) and everything it pulls in. +# `contract` + `wire` need 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. alloc = ["serde/alloc"] std = [ "alloc", "serde/std", - "dep:eyre", - "dep:downcast-rs", - "dep:async-trait", - "dep:tokio", "dep:abi_stable", "dep:libloading", - "dep:serde_json", "dep:rmp", "dep:rmp-serde", ] -# Call Systems (all in the `std` layer today) -open_rpc = ["std"] -json_rpc = ["std", "dep:json-rpc-types"] - [dependencies] -# `contract` — no_std, no alloc +# `contract` / `wire` — no_std, no alloc serde = { version = "1.0.190", default-features = false } bytemuck = { version = "1.14.0", features = ["derive"] } # `std` layer -eyre = { version = "0.6.8", optional = true } -downcast-rs = { version = "1.2.0", optional = true } -async-trait = { version = "0.1.74", optional = true } -tokio = { version = "1.33.0", features = ["net", "rt-multi-thread", "sync", "io-util"], default-features = false, optional = true } abi_stable = { version = "0.11.2", optional = true } libloading = { version = "0.8.0", optional = true } -json-rpc-types = { version = "1.3.4", optional = true } rmp = { version = "0.8.12", optional = true } rmp-serde = { version = "1.1.2", optional = true } -serde_json = { version = "1.0.104", optional = true } [dev-dependencies] -tokio = { version = "1.33.0", features = ["full"] } serde = { version = "1.0.190", features = ["derive"] } diff --git a/runtime/core/src/lib.rs b/runtime/core/src/lib.rs index e70cc03..68f44dd 100644 --- a/runtime/core/src/lib.rs +++ b/runtime/core/src/lib.rs @@ -1,5 +1,5 @@ // `no_std`-first: the crate is `#![no_std]` unless the `std` feature is on -// (it is, by default). `--no-default-features` builds just `contract`. +// (it is, by default). `--no-default-features` builds `contract` + `wire`. #![cfg_attr(not(feature = "std"), no_std)] #[cfg(feature = "alloc")] @@ -8,12 +8,19 @@ extern crate alloc; // The `core ↔ target` contract — `no_std`, allocation-free. 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; -// The `std` layer: transport, async, the setup builders, the dylib ABI. +// The frame transport and the provider serve loop. +#[cfg(feature = "alloc")] +pub mod serve; +#[cfg(feature = "alloc")] +pub mod transport; + +// The Comline-package dynamic-library ABI. #[cfg(feature = "std")] pub mod package_abi; -#[cfg(feature = "std")] -pub mod setup; diff --git a/runtime/core/src/serve.rs b/runtime/core/src/serve.rs new file mode 100644 index 0000000..18939d5 --- /dev/null +++ b/runtime/core/src/serve.rs @@ -0,0 +1,56 @@ +//! Provider-side serving: read a request frame, dispatch it, write the +//! response frame. + +use alloc::vec::Vec; + +use crate::contract::{Dispatch, Kind, 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 { + dispatch: D, + format: W, + recv: Vec, + envelope: Vec, + response: Vec, +} + +impl Server { + pub fn new(dispatch: D, format: W) -> Self { + Self { + dispatch, + format, + recv: Vec::new(), + envelope: Vec::new(), + response: Vec::new(), + } + } + + /// Handle one call. `Ok(true)` — a call was served; `Ok(false)` — the + /// transport closed. + pub fn serve_one(&mut self, transport: &mut T) -> Result { + if transport.recv(&mut self.recv).is_err() { + return Ok(false); + } + + let (call_id, request_id, params) = + wire::decode_request(&self.recv).ok_or(RuntimeError::Framing)?; + + self.envelope.clear(); + self.dispatch + .dispatch(Kind::Id(call_id), params, &self.format, &mut self.envelope)?; + + self.response.clear(); + wire::encode_response(request_id, &self.envelope, &mut self.response); + transport.send(&self.response)?; + Ok(true) + } + + /// Serve calls until the transport closes. + pub fn serve(&mut self, transport: &mut T) -> Result<(), RuntimeError> { + while self.serve_one(transport)? {} + Ok(()) + } +} diff --git a/runtime/core/src/setup/abstract_call.rs b/runtime/core/src/setup/abstract_call.rs deleted file mode 100644 index ffbf12e..0000000 --- a/runtime/core/src/setup/abstract_call.rs +++ /dev/null @@ -1,54 +0,0 @@ -// Relative Modules - -// Standard Uses -use std::any::Any; - -// Crate Uses - -// External Uses - - - -pub struct Parameter<'a>(Inner<'a>); // for<'a> Deserialize<'a> -pub type Inner<'a> = &'a (dyn Any + Sync); - -pub struct Message<'a> { - pub parameters: Vec> -} - -impl<'a> Message<'a> { - pub fn new() -> Self { Self { parameters: vec![] } } - pub fn parameter(mut self, parameter: &'a T) -> Self { - self.parameters.push(Parameter(parameter)); self - } -} - -/* -impl<'a> Deserialize<'a> for Message<'a> { - fn deserialize(deserializer: D) -> Result where D: Deserializer<'a> { - todo!() - } -} - - -struct MessageVisitor; -impl Visitor for MessageVisitor { - type Value = (); - - fn expecting(&self, formatter: &mut Formatter) -> std::fmt::Result { - todo!() - } -} -*/ - -#[allow(unused)] -pub struct AbstractCall { - pub(crate) settings: &'static [&'static (&'static str, &'static Setting)], - pub(crate) parameters: P -} - -pub enum Setting { - None, - Num(usize), - Str(&'static str) -} diff --git a/runtime/core/src/setup/call_system/consumer.rs b/runtime/core/src/setup/call_system/consumer.rs deleted file mode 100644 index 8367bbb..0000000 --- a/runtime/core/src/setup/call_system/consumer.rs +++ /dev/null @@ -1,38 +0,0 @@ -// Standard Uses - -// Crate Uses -use crate::setup::CallResult; -use crate::setup::call_system::{CallSystem, Kind, DEFAULT_CALL_TIMEOUT}; -use crate::setup::abstract_call::AbstractCall; - -// External Uses -use serde::{Deserialize, Serialize}; - - -pub trait CallSystemConsumer: CallSystem { - fn send_async_call(&mut self, kind: Kind, message: AbstractCall) - -> impl std::future::Future> + Send - where M: Send + Serialize + for<'de> Deserialize<'de> - ; - - fn send_blocking_call(&mut self, kind: Kind, call: AbstractCall) - -> CallResult - where M: Send + Serialize + for<'de> Deserialize<'de> - ; -} - -pub fn send_call_with_conditions(kind: Kind, call: AbstractCall) -> CallResult - where M: Send -{ - let call_timeout = DEFAULT_CALL_TIMEOUT; - let start_time = std::time::Instant::now(); - - let response: Option = None; - while response.is_none() { - if start_time.elapsed().as_millis() > call_timeout { - return Err(()) - } - } - - todo!() -} diff --git a/runtime/core/src/setup/call_system/meta.rs b/runtime/core/src/setup/call_system/meta.rs deleted file mode 100644 index a4abfc0..0000000 --- a/runtime/core/src/setup/call_system/meta.rs +++ /dev/null @@ -1,26 +0,0 @@ -// Standard Uses - -// Crate Uses -use crate::setup::abstract_call::AbstractCall; - -// External Uses - - -pub trait CallProtocolMeta { - fn calls_names(&self) -> &'static [&'static str]; - fn call_name_from_id(&self, id: u16) -> Option<&'static str> { - self.calls_names().get(id as usize).copied() - } - - //fn arguments() -> &'static [(usize, )]; - - fn make_call(&self, parameters: P) -> AbstractCall

{ - let call = AbstractCall { - settings: &[], - parameters, - }; - - call - } -} - diff --git a/runtime/core/src/setup/call_system/mod.rs b/runtime/core/src/setup/call_system/mod.rs deleted file mode 100644 index 5548ee6..0000000 --- a/runtime/core/src/setup/call_system/mod.rs +++ /dev/null @@ -1,66 +0,0 @@ -// Relative Modules -pub mod meta; -pub mod systems; -pub mod consumer; -pub mod provider; - -// Standard Uses -use std::sync::{Arc, RwLock}; - -// Crate Uses -use crate::setup::call_system::provider::CallSystemProvider; -use crate::setup::transport::consumer::CommunicationConsumer; -use crate::setup::transport::provider::CommunicationProvider; -use crate::setup::call_system::consumer::CallSystemConsumer; - -// External Uses -use downcast_rs::{DowncastSync, impl_downcast}; -use tokio::sync::watch; - - -pub const DEFAULT_CALL_TIMEOUT: u128 = 800; - -pub trait CallSystem: Send + Sync + DowncastSync { - fn add_event_listener(&mut self, callback: EventType); - - fn receive_data(&mut self, data: &[u8]); - fn send_data(&mut self, data: &[u8]); -} -impl_downcast!(sync CallSystem); - -pub enum Origin { - Consumer(Arc>), - Provider(Arc>) -} - - - -pub enum EventType { - ReceivedBytes(watch::Receiver>>), - SentBytes(watch::Receiver>>), -} - -pub enum Event<'data, Incoming, Outgoing> { - ReceivedBytes(&'data [u8]), - SentBytes(&'data [u8]), - - ReceivedMessage(&'data Incoming), - SentMessage(&'data Outgoing), -} - -pub trait Callback: Send + Sync { - fn on_received_data(&mut self, data: &[u8]); - fn on_sent_data(&mut self, data: &[u8]); -} - - -/// Kind of call name -pub enum Kind { - Id(u16), - Named(String) -} - -pub trait CallSystemBuilder { - fn new_consumer() -> impl CallSystemConsumer; - fn new_provider() -> impl CallSystemProvider; -} \ No newline at end of file diff --git a/runtime/core/src/setup/call_system/provider.rs b/runtime/core/src/setup/call_system/provider.rs deleted file mode 100644 index 4d46c6a..0000000 --- a/runtime/core/src/setup/call_system/provider.rs +++ /dev/null @@ -1,23 +0,0 @@ -// Standard Uses - -use serde::{Deserialize, Serialize}; - -// Crate Uses -use crate::setup::CallResult; -use crate::setup::call_system::CallSystem; -use crate::setup::call_system::Kind; -use crate::setup::abstract_call::AbstractCall; - -// External Uses - - -pub trait CallSystemProvider: CallSystem { - fn receive_async_call(&mut self, kind: Kind, message: AbstractCall) - -> CallResult - where M: Send + Serialize + for<'de> Deserialize<'de> - ; -} - -pub fn send_call_with_conditions() -> CallResult { - todo!() -} diff --git a/runtime/core/src/setup/call_system/systems/json_rpc.rs b/runtime/core/src/setup/call_system/systems/json_rpc.rs deleted file mode 100644 index 98a2d7e..0000000 --- a/runtime/core/src/setup/call_system/systems/json_rpc.rs +++ /dev/null @@ -1,195 +0,0 @@ -// Standard Uses -use std::sync::{Arc, RwLock}; - -// Crate Uses -use crate::setup::CallResult; -use crate::setup::call_system; -use crate::setup::call_system::{Callback, CallSystem, CallSystemProvider, EventType, Origin}; -use crate::setup::call_system::consumer::CallSystemConsumer; -use crate::setup::call_system::Kind; -use crate::setup::abstract_call::AbstractCall; - -// External Uses -use eyre::Result; -use json_rpc_types::Id; -use serde::de::DeserializeOwned; -use serde::{Deserialize, Serialize}; -use tokio::sync::watch; - - -// type Request = json_rpc_types::Request; -type Request

= json_rpc_types::Request

; -type Response = json_rpc_types::Response; - - -#[allow(dead_code)] -pub struct JsonRPCv2 { - transporter: call_system::Origin, - events_sender: watch::Sender>>, - events_receiver: watch::Receiver>>, - event_listeners: Vec, -} - -impl JsonRPCv2 { - pub fn new(origin: call_system::Origin) -> Self { - let (sx, rx) = watch::channel(None); - - Self { - transporter: origin, - events_sender: sx, - events_receiver: rx, - event_listeners: vec![], - } - } - pub fn into_threaded(self) -> Arc> { Arc::new(RwLock::new(self)) } -} - -#[allow(unused_variables)] -impl CallSystem for JsonRPCv2 { - fn add_event_listener(&mut self, callback: EventType) { - self.event_listeners.push(callback); - } - - fn receive_data(&mut self, data: &[u8]) { - todo!() - } - fn send_data(&mut self, data: &[u8]) { - todo!() - } -} - - -impl CallSystemProvider for JsonRPCv2 { - fn receive_async_call(&mut self, kind: Kind, message: AbstractCall) - -> CallResult - where M: Send + Serialize + for<'de> Deserialize<'de> - { - // let incoming: Request = serde_json::from_slice(data).unwrap(); - - todo!() - } -} - - -impl CallSystemConsumer for JsonRPCv2 { - async fn send_async_call(&mut self, kind: Kind, call: AbstractCall) - -> CallResult - where M: Send + Serialize + for<'de> Deserialize<'de> - { - // TODO: Request Id might be just a call instance number - let id = Some(Id::Num(0)); - - let method = match kind { - Kind::Id(id) => id.to_string().as_str().try_into().unwrap(), - Kind::Named(name) => name.as_str().try_into().unwrap(), - }; - - let request = Request { - jsonrpc: json_rpc_types::Version::V2, - id: id.clone(), method, - //params: map_call_request_to_json(call).map_err(|_| ())?, - params: Some(call.parameters), - }; - let request = serde_json::to_vec(&request).unwrap(); - - match &self.transporter { - Origin::Consumer(transporter) => { - let mut transporter = transporter.write().unwrap(); - transporter.send_data(&request).unwrap(); // TODO: This needs to be properly raised up - }, - _ => panic!( - "A Call System Provider implementation can only hold a\ - Consumer transporter and not Provider transporter" - ) - } - - - let mut response: Option> = None; - - for listener in self.event_listeners.iter_mut() { - match listener { - EventType::ReceivedBytes(cb) => { - response = Some(receive_event_by_id(cb, id.clone().unwrap()).await.unwrap()); - }, - _ => () - } - } - let inner = response.unwrap().payload.map_err(|_| ())?; - - Ok(inner) - } - - fn send_blocking_call(&mut self, kind: Kind, call: AbstractCall) - -> CallResult - where M: Send + Serialize + for<'de> Deserialize<'de> - { - tokio::task::block_in_place(move || { - tokio::runtime::Handle::current().block_on(async move { - self.send_async_call(kind, call).await - }) - }) - } -} - - -impl Callback for JsonRPCv2 { - fn on_received_data(&mut self, data: &[u8]) { - //self.events_sender.send(data).unwrap(); - - /* - for listener in &self.events_sender { - match listener { - EventType::ReceivedBytes(cb) => { - // TODO: Data reference - // self.events_sender.send(data); - }, - _ => panic!("How are we here, this is a runtime developer mistake") - } - } - */ - } - - #[allow(unused)] - fn on_sent_data(&mut self, data: &[u8]) { - todo!() - } -} - -async fn receive_event_by_id( - receiver: &mut watch::Receiver>>, id: Id -) -> Result> { - let data = receiver.wait_for(|v| { - let v = v.as_ref().unwrap(); - - match id { - Id::Num(n) => { - let incoming_id = u64::from_le_bytes(v[0..8].try_into().unwrap()); - incoming_id == n - }, - Id::Str(s) => { - let incoming_id = std::str::from_utf8(&v[0..36]).map_err(|_| ()).unwrap(); - incoming_id == s - } - } - }).await?; - - let request: Response = serde_json::from_slice(&data.to_owned().unwrap()).unwrap(); - - Ok(request) -} - -/* -fn map_call_request_to_json(message: M) -> Result> { - // TODO: If there is not parameters, assigning as None is much better - let parameters = Some(serde_json::Value::Array(vec![])); - - /* - for parameter in message.parameters { - - } - */ - - Ok(parameters) -} -*/ - diff --git a/runtime/core/src/setup/call_system/systems/mod.rs b/runtime/core/src/setup/call_system/systems/mod.rs deleted file mode 100644 index 455341b..0000000 --- a/runtime/core/src/setup/call_system/systems/mod.rs +++ /dev/null @@ -1,7 +0,0 @@ -// Relative Modules -//#[cfg(feature = "open_rpc")] -//pub mod open_rpc; - -#[cfg(feature = "json_rpc")] -pub mod json_rpc; - diff --git a/runtime/core/src/setup/call_system/systems/xml_rpc.rs b/runtime/core/src/setup/call_system/systems/xml_rpc.rs deleted file mode 100644 index 01ee861..0000000 --- a/runtime/core/src/setup/call_system/systems/xml_rpc.rs +++ /dev/null @@ -1,6 +0,0 @@ -// Standard Uses - -// Crate Uses - -// External Uses - diff --git a/runtime/core/src/setup/mod.rs b/runtime/core/src/setup/mod.rs deleted file mode 100644 index 72c3a0c..0000000 --- a/runtime/core/src/setup/mod.rs +++ /dev/null @@ -1,10 +0,0 @@ -// Relative Modules -pub mod transport; -pub mod call_system; -pub mod abstract_call; - - -// TODO: A Error type will be necessary here, for things like API Call state and response status, information, etc -pub type CallResult = Result; - - diff --git a/runtime/core/src/setup/transport/consumer.rs b/runtime/core/src/setup/transport/consumer.rs deleted file mode 100644 index f3908bf..0000000 --- a/runtime/core/src/setup/transport/consumer.rs +++ /dev/null @@ -1,110 +0,0 @@ -// Standard Uses -use std::sync::{Arc, RwLock}; - -// Crate Uses -use crate::setup::call_system::consumer::CallSystemConsumer; -use crate::setup::call_system::Origin; - -// External Uses -use async_trait::async_trait; -use downcast_rs::{DowncastSync, impl_downcast}; - - -#[async_trait] -pub trait CommunicationConsumer: Send + Sync { - async fn connect_to_provider(&self); - - fn send_data(&mut self, data: &[u8]) -> eyre::Result<()>; - async fn send_data_async(&mut self, data: &[u8]) -> eyre::Result<()>; -} - -pub struct ConsumerSetup where T: CommunicationConsumer, CC: CallSystemConsumer { - pub transport_method: Arc>, - pub call_system: Option>>, - pub capabilities: Vec> -} - -impl ConsumerSetup where T: CommunicationConsumer + 'static, CC: CallSystemConsumer { - pub fn with_transport(transporter: T) -> Self { - Self { - transport_method: Arc::new(RwLock::new(transporter)), - call_system: None, - capabilities: vec![], - } - } - - pub fn with_call_system(mut self, call_system: CFn) -> Self - where CFn: FnOnce(Origin) -> CC - { - self.call_system = Some(Arc::new(RwLock::new(call_system( - Origin::Consumer(self.transport_method.clone()), - )))); - self - } - - pub fn with_capability(mut self, capability: Cfn) -> Self - where - C: ConsumerCapability, - Cfn: FnOnce(Arc>) -> C - { - self.capabilities.push(Box::new(capability(self.call_system.as_ref().unwrap().clone()))); - self - } - - pub fn into_threaded(self) -> Arc> { Arc::new(RwLock::new(self)) } - - pub fn add_default_capability(mut self, capability: Cfn) -> Self - where - C: ConsumerCapability, - Cfn: FnOnce(Arc>) -> C - { - self.capabilities.push(Box::new(capability(self.call_system.as_ref().unwrap().clone()))); - self - } - - // TODO: Unsure if these are necessary right now, their signatures are also incorrect - // they should have the same parameters ad the `ẁith_capability` method - /* - pub fn add_capability< - C: ConsumerCapability, - Cfn: Fn(&ConsumerSetup) -> C - >( - &mut self, capability_fn: Cfn - ) { - self.capabilities.push(Box::new(capability_fn(&*self))); - } - - pub fn add_capabilities(&mut self, mut capabilities: Vec>) { - self.capabilities.append(&mut capabilities); - } - */ - - pub fn capability(&self) -> Option<&C> { - for capability in self.capabilities.iter() { - if let Some(cap) = capability.downcast_ref::() { - return Some(cap); - } - } - - None - } - - pub fn capability_mut(&mut self) -> Option<&mut C> { - for capability in &mut self.capabilities { - if let Some(cap) = capability.downcast_mut::() { - return Some(&mut *cap); - } - } - - None - } -} - - -pub type SharedConsumerSetup = Arc>>; - -pub trait ConsumerCapability: DowncastSync { - //fn setup(&self) -> Arc>; -} -impl_downcast!(sync ConsumerCapability); - diff --git a/runtime/core/src/setup/transport/methods/mod.rs b/runtime/core/src/setup/transport/methods/mod.rs deleted file mode 100644 index 19c9805..0000000 --- a/runtime/core/src/setup/transport/methods/mod.rs +++ /dev/null @@ -1,2 +0,0 @@ -// Relative Modules -pub mod tcp; diff --git a/runtime/core/src/setup/transport/methods/tcp.rs b/runtime/core/src/setup/transport/methods/tcp.rs deleted file mode 100644 index 15276bc..0000000 --- a/runtime/core/src/setup/transport/methods/tcp.rs +++ /dev/null @@ -1,178 +0,0 @@ -// Standard Uses - -// Crate Uses - -// External Uses - - -pub mod provider { - // Standard Uses - use std::net::SocketAddr; - use std::sync::{Arc, RwLock}; - - // Crate Uses - use crate::setup::transport::provider::CommunicationProvider; - use crate::setup::call_system::Callback; - - // External Uses - use eyre::Result; - use tokio::net::{TcpListener, TcpStream}; - use tokio::io::AsyncReadExt; - use async_trait::async_trait; - - - pub struct TcpProvider { - listener: TcpListener, - pub connection_count: usize, - data_received_callback: Vec>>, - } - - impl TcpProvider { - // const INCOMING_DATA_MIN_LEN: usize = 1; - - pub async fn with_address( - address: &str, - // data_received_callback: fn(&[u8]) - ) -> Result { - Ok(Self { - listener: TcpListener::bind(address).await?, - connection_count: 0, - data_received_callback: vec![], - }) - } - - pub fn and_callback(mut self, callback: Arc>) -> Self - { - self.data_received_callback.push(callback); - self - } - - pub fn into_threaded(self) -> Arc> { Arc::new(RwLock::new(self)) } - - pub async fn listen_connections(&mut self, /*call_system: &mut dyn CallSystem*/) { - loop { self.listen_incoming_connection(/*call_system*/).await } - } - - pub async fn listen_incoming_connection( - &self, - // call_system: &mut dyn CallSystem - ) { - let (stream, address) = self.listener.accept().await.unwrap(); - stream.set_nodelay(true).unwrap(); - - // TODO: Introduce interior mutability for count, and/or atomics - //self.connection_count += 1; - self.listen_stream(stream, address, /*call_system*/).await; - //self.connection_count -= 1; - } - - pub async fn listen_stream( - &self, mut stream: TcpStream, address: SocketAddr, - //call_system: &mut dyn CallSystem - ) { - let mut buf = [0; 1024]; - - loop { - let length = match stream.read(&mut buf).await { - Ok(0) => return, // Stream closed - Ok(n) => n, - Err(e) => { - // TODO: Shouldn't be a panic and needs to do proper error setup - panic!("Couldn't read on stream: {e}"); - } - }; - - /* - if length < Self::INCOMING_DATA_MIN_LEN { - panic!( - "Incoming packet is not big enough, got {} but expected at least {} bytes", - length, Self::INCOMING_DATA_MIN_LEN - ) - } - */ - let data = &buf[..length]; - - println!( - "[Provider] {} - Incoming data ({} bytes, first 10 bytes: {:?}", - address, length, &data - ); - - for callback in &self.data_received_callback { - callback.write().unwrap().on_received_data(&data) - } - } - } - } - - #[async_trait] - impl CommunicationProvider for TcpProvider { - fn add_received_data_callback(&mut self, callback: Arc>) { - self.data_received_callback.push(callback) - } - - async fn listen_for_connections(&mut self, /*call_system: &mut dyn CallSystem*/) { - self.listen_connections(/*call_system*/).await; - } - } -} - -pub mod consumer { - // Standard Uses - use std::io::Write; - use std::net::TcpStream; - use std::sync::{Arc, RwLock}; - - // Crate Uses - use crate::setup::transport::consumer::CommunicationConsumer; - - // External Uses - use eyre::Result; - use async_trait::async_trait; - - - pub struct TcpConsumer { - stream: TcpStream, - data_received_callback: Vec, - } - impl TcpConsumer { - pub fn into_threaded(self) -> Arc> { Arc::new(RwLock::new(self)) } - } - - impl TcpConsumer { - pub fn with_address(address: &str) -> Result { - Ok(Self { - stream: TcpStream::connect(address)?, - data_received_callback: vec![] - }) - } - pub fn and_callback(mut self, callback: fn(&[u8])) -> Self { - self.data_received_callback.push(callback); - self - } - } - - #[async_trait] - impl CommunicationConsumer for TcpConsumer { - async fn connect_to_provider(&self) { - todo!() - } - - #[allow(unused)] - fn send_data(&mut self, data: &[u8]) -> Result<()> { - println!( - "Sending data ({} bytes, first 10: {:?}): {:?}", - data.len(), &data[..10], String::from_utf8_lossy(&data[..10]) - ); - - self.stream.write_all(data)?; - self.stream.flush(); - - Ok(()) - } - - #[allow(unused)] - async fn send_data_async(&mut self, data: &[u8]) -> Result<()> { - todo!() - } - } -} diff --git a/runtime/core/src/setup/transport/mod.rs b/runtime/core/src/setup/transport/mod.rs deleted file mode 100644 index 319816c..0000000 --- a/runtime/core/src/setup/transport/mod.rs +++ /dev/null @@ -1,29 +0,0 @@ -// Relative Modules -pub mod provider; -pub mod consumer; -pub mod methods; - -// External Uses - - -/* -pub trait MessageReceiver: Send + DowncastSync { - #[allow(unused)] - fn receive_data(&self, data: &[u8]) { - todo!() - } -} -impl_downcast!(sync MessageReceiver); - -#[allow(unused)] -pub trait MessageSender: Send + DowncastSync { - fn send_data(&self, data: &[u8]) { - todo!() - } -} -impl_downcast!(sync MessageSender); - -pub trait MessageSenderBuilder: MessageSender { - fn new() -> Self; -} -*/ diff --git a/runtime/core/src/setup/transport/provider.rs b/runtime/core/src/setup/transport/provider.rs deleted file mode 100644 index 5c95f7a..0000000 --- a/runtime/core/src/setup/transport/provider.rs +++ /dev/null @@ -1,69 +0,0 @@ -// Standard Uses -use std::sync::{Arc, RwLock}; - -// Crate Uses -use crate::setup::call_system::{Callback, Origin}; -use crate::setup::call_system::provider::CallSystemProvider; - -// External Uses -use async_trait::async_trait; -use downcast_rs::{DowncastSync, impl_downcast}; - - -#[async_trait] -pub trait CommunicationProvider: DowncastSync { - fn add_received_data_callback(&mut self, callback: Arc>); - async fn listen_for_connections(&mut self); -} -impl_downcast!(sync CommunicationProvider); - - -pub struct ProviderSetup { - pub transporter: Arc>, - pub call_system: Option>>, - pub capabilities: Vec> -} - -impl ProviderSetup where T: CommunicationProvider, CS: CallSystemProvider { - pub fn with_transporter(transporter: T) -> Self { - Self { - transporter: Arc::new(RwLock::new(transporter)), - call_system: None, - capabilities: vec![], - } - } - - pub fn with_call_system(mut self, call_system: Cfn) -> Self - where - CS: CallSystemProvider + Callback, - Cfn: FnOnce(Origin) -> CS - { - let call_system = Arc::new(RwLock::new( - call_system(Origin::Provider(self.transporter.clone())) - )); - self.transporter.write().unwrap().add_received_data_callback(call_system.clone()); - - self.call_system = Some(call_system); - self - } - - pub fn with_capability(mut self, capability: Cfn) -> Self - where - C: ProviderCapability + 'static, - Cfn: FnOnce(Arc>) -> C - { - self.capabilities.push(Box::new(capability( - self.call_system.as_ref().unwrap().clone() - ))); - self - } - - pub fn add_capabilities(&mut self, mut capabilities: Vec>) { - self.capabilities.append(&mut capabilities); - } - - pub fn into_threaded(self) -> Arc { Arc::new(self) } -} - -pub trait ProviderCapability {} - diff --git a/runtime/core/src/transport.rs b/runtime/core/src/transport.rs new file mode 100644 index 0000000..3b59cf8 --- /dev/null +++ b/runtime/core/src/transport.rs @@ -0,0 +1,57 @@ +//! The byte-frame transport. Message-oriented — `send` / `recv` move whole +//! request / response frames (see [`wire`](crate::wire)). Sync, to pair with +//! the sync [`Dispatch`](crate::contract::Dispatch) and [`Server`](crate::serve::Server). + +use alloc::vec::Vec; + +use crate::contract::RuntimeError; + +/// A duplex frame channel. One end's `send` is the other end's `recv`. +pub trait Transport { + /// Send one frame. + fn send(&mut self, frame: &[u8]) -> Result<(), RuntimeError>; + + /// Receive the next frame into `buf` (the caller clears and reuses it). + /// `Err(RuntimeError::Transport)` once the peer is gone. + fn recv(&mut self, buf: &mut Vec) -> Result<(), RuntimeError>; +} + +#[cfg(feature = "std")] +mod in_memory { + use std::sync::mpsc::{channel, Receiver, Sender}; + + use super::{Transport, Vec}; + use crate::contract::RuntimeError; + + /// An in-process [`Transport`] — for tests, examples, and same-binary + /// consumer/provider setups. + pub struct InMemory { + tx: Sender>, + rx: Receiver>, + } + + /// A crossed pair: what `a` sends, `b` receives, and vice versa. + pub fn duplex() -> (InMemory, InMemory) { + let (a_tx, a_rx) = channel(); + let (b_tx, b_rx) = channel(); + (InMemory { tx: a_tx, rx: b_rx }, InMemory { tx: b_tx, rx: a_rx }) + } + + impl Transport for InMemory { + fn send(&mut self, frame: &[u8]) -> Result<(), RuntimeError> { + self.tx + .send(frame.to_vec()) + .map_err(|_| RuntimeError::Transport) + } + + fn recv(&mut self, buf: &mut Vec) -> Result<(), RuntimeError> { + let frame = self.rx.recv().map_err(|_| RuntimeError::Transport)?; + buf.clear(); + buf.extend_from_slice(&frame); + Ok(()) + } + } +} + +#[cfg(feature = "std")] +pub use in_memory::{duplex, InMemory}; diff --git a/runtime/core/src/wire.rs b/runtime/core/src/wire.rs new file mode 100644 index 0000000..9a5f09f --- /dev/null +++ b/runtime/core/src/wire.rs @@ -0,0 +1,69 @@ +//! 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. +pub fn encode_request(call_id: u16, request_id: u64, params: &[u8], out: &mut dyn BufMut) { + out.put_u16_le(call_id); + out.put_u64_le(request_id); + out.put_slice(params); +} + +/// `(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); + } +} diff --git a/runtime/core/tests/mod.rs b/runtime/core/tests/mod.rs deleted file mode 100644 index 4a0cb93..0000000 --- a/runtime/core/tests/mod.rs +++ /dev/null @@ -1,3 +0,0 @@ -// The `setups` suite drives the `std`-only `setup/` layer. -#[cfg(feature = "std")] -pub mod setups; diff --git a/runtime/core/tests/serve_roundtrip.rs b/runtime/core/tests/serve_roundtrip.rs new file mode 100644 index 0000000..9aa836d --- /dev/null +++ b/runtime/core/tests/serve_roundtrip.rs @@ -0,0 +1,94 @@ +//! 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. +#![cfg(feature = "std")] + +use std::thread; + +use comline_runtime::contract::{BufMut, Dispatch, Envelope, Kind, 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; } + +#[derive(Serialize, Deserialize)] +struct HelloParams<'a> { + #[serde(borrow)] + name: &'a str, +} + +const CALLS: &[&str] = &["hello"]; + +trait Greet { + fn hello(&self, name: &str) -> String; +} + +struct GreetDispatcher(T); + +impl Dispatch for GreetDispatcher { + fn dispatch( + &self, + call: Kind, + params: &[u8], + fmt: &W, + out: &mut dyn BufMut, + ) -> 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 mut body = Vec::new(); + fmt.encode(&reply, &mut body)?; + Envelope::encode_ok(&body, out); + Ok(()) + } + _ => Err(RuntimeError::UnknownCall), + } + } +} + +struct Impl; +impl Greet for Impl { + fn hello(&self, name: &str) -> String { + format!("hi, {name}") + } +} + +#[test] +fn a_call_round_trips_over_the_transport() { + let (mut client, provider) = duplex(); + + let server = thread::spawn(move || { + let mut provider = provider; + Server::new(GreetDispatcher(Impl), MsgPack) + .serve(&mut provider) + .unwrap(); + }); + + // 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); + 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(); + assert_eq!(request_id, 1); + + let reply: String = match Envelope::decode(envelope).unwrap() { + Envelope::Ok(payload) => MsgPack.decode(payload).unwrap(), + Envelope::Err { .. } => panic!("unexpected error frame"), + }; + assert_eq!(reply, "hi, world"); + + drop(client); // closes the transport → `serve` returns + server.join().unwrap(); +} diff --git a/runtime/core/tests/setups/jrpc_tcp/client.rs b/runtime/core/tests/setups/jrpc_tcp/client.rs deleted file mode 100644 index 314e90c..0000000 --- a/runtime/core/tests/setups/jrpc_tcp/client.rs +++ /dev/null @@ -1,62 +0,0 @@ -// Standard Uses - -// Crate Uses -use crate::setups::jrpc_tcp::generated::{ - schemas::consumer::GreetConsumerProtocol, - consumer::GreetConsumer -}; - -// External Uses -use comline_runtime::setup::{ - transport::{ - consumer::ConsumerSetup, - methods::tcp::consumer::TcpConsumer - }, - call_system::{ - consumer::CallSystemConsumer, - meta::CallProtocolMeta, Kind, - systems::json_rpc::JsonRPCv2 - }, - CallResult -}; - - -impl GreetConsumerProtocol for GreetConsumer { - fn greet(&self, name: &str) -> CallResult { - let call_name = Kind::Named(self.call_name_from_id(0).unwrap().to_owned()); - let call = self.make_call(name.to_owned()); - - let mut caller = self.caller.write().unwrap(); - - let result = caller.send_blocking_call(call_name, call)?; - Ok(result) - } -} - -pub(crate) async fn main() { - println!("Running Client"); - - let (address, port) = ("127.0.0.1", "2620"); - let full_address = &*(address.to_owned() + ":" + port); - - let transporter = TcpConsumer::with_address(full_address).unwrap(); - let mut setup = ConsumerSetup::with_transport(transporter) - .with_call_system(JsonRPCv2::new) - .with_capability(GreetConsumer::new); - - greet_with_name(&mut setup); -} - - -fn greet_with_name(setup: &mut ConsumerSetup) { - //let mut setup_write = setup.write().unwrap(); - let greeter = setup.capability_mut::>().unwrap(); - let name = "Client"; - - println!("[Client] Sending a greet request with name '{}'", name); - let response = greeter.greet(name).unwrap(); - println!("[Client] Received a greet response saying: '{}'", response); - - assert_eq!("Hello Client", response); -} - diff --git a/runtime/core/tests/setups/jrpc_tcp/generated.rs b/runtime/core/tests/setups/jrpc_tcp/generated.rs deleted file mode 100644 index a2f54df..0000000 --- a/runtime/core/tests/setups/jrpc_tcp/generated.rs +++ /dev/null @@ -1,104 +0,0 @@ -// These structures are just mimics of what Comline would generate - -pub mod schemas { - - // Internal Uses - use comline_runtime::setup::CallResult; - use comline_runtime::setup::{ - call_system::meta::CallProtocolMeta, - transport::{ - provider::ProviderCapability, - consumer::ConsumerCapability - } - }; - - // External Uses - - - pub trait GreetProtocol: CallProtocolMeta {} - - pub mod provider { - use super::*; - - pub trait GreetProviderProtocol: GreetProtocol + ProviderCapability { - fn greet(&self, name: &str) -> CallResult; - } - } - - - pub mod consumer { - use super::*; - - pub trait GreetConsumerProtocol: GreetProtocol + ConsumerCapability { - fn greet(&self, name: &str) -> CallResult; - } - } -} - -pub mod provider { - // Standard Uses - use std::sync::{Arc, RwLock}; - - // Crate Uses - use super::schemas::GreetProtocol; - - // Internal Uses - use comline_runtime::setup::{ - call_system::meta::CallProtocolMeta, - transport::{ - provider::ProviderCapability - } - }; - - pub struct GreetProvider { - #[allow(dead_code)] - pub(crate) caller: Arc>, - } - impl GreetProvider { - pub fn new(caller: Arc>) -> Self { Self { caller } } - } - - impl GreetProtocol for GreetProvider {} - impl ProviderCapability for GreetProvider {} - impl CallProtocolMeta for GreetProvider { - //const CALL_NAMES: &'static [&'static str] = &[]; - - fn calls_names(&self) -> &'static [&'static str] { - todo!() - } - } -} - -pub mod consumer { - // Standard Uses - use std::sync::{Arc, RwLock}; - - // Crate Uses - use super::schemas::GreetProtocol; - - // External Uses - use comline_runtime::setup::{ - transport::{ - consumer::ConsumerCapability - }, - call_system::meta::CallProtocolMeta, - }; - use comline_runtime::setup::call_system::consumer::CallSystemConsumer; - - pub struct GreetConsumer { - #[allow(unused_variables)] - pub(crate) caller: Arc>, - } - impl GreetConsumer { - pub fn new(caller: Arc>) -> Self { Self { caller } } - } - - impl GreetProtocol for GreetConsumer {} - impl ConsumerCapability for GreetConsumer {} - impl CallProtocolMeta for GreetConsumer { - fn calls_names(&self) -> &'static [&'static str] { - &["greet"] - } - } -} - diff --git a/runtime/core/tests/setups/jrpc_tcp/mod.rs b/runtime/core/tests/setups/jrpc_tcp/mod.rs deleted file mode 100644 index 875d2c8..0000000 --- a/runtime/core/tests/setups/jrpc_tcp/mod.rs +++ /dev/null @@ -1,26 +0,0 @@ -// Relative Modules -mod generated; -mod server; -mod client; - - -// The `setup/` call-system + transport layer is still stubbed (`todo!()`). -// Re-enable when `Dispatch` / `WireFormat` are wired in — rollout step 7b+. -#[ignore = "setup/ layer is stubbed"] -#[tokio::test(flavor = "multi_thread", worker_threads = 1)] -async fn send_name_from_client_and_receive_hello_from_server() { - // This entry point is just an example of simulation, you would do differently - // if not simulating both parts - - tokio::task::LocalSet::new().run_until(async move { - // Lets spawn a handle for the server, pretending its a different process - let server_thread = tokio::task::spawn_local(server::main()); - - // And for the client we just run in our existing thread - let client_thread = tokio::task::spawn(client::main()); - - server_thread.await.unwrap(); - client_thread.await.unwrap(); - }).await; -} - diff --git a/runtime/core/tests/setups/jrpc_tcp/server.rs b/runtime/core/tests/setups/jrpc_tcp/server.rs deleted file mode 100644 index 7c9f042..0000000 --- a/runtime/core/tests/setups/jrpc_tcp/server.rs +++ /dev/null @@ -1,46 +0,0 @@ -// Standard Uses - -// Crate Uses -use crate::setups::jrpc_tcp::generated::{ - schemas::provider::GreetProviderProtocol, - provider::GreetProvider -}; - -// External Uses -use comline_runtime::setup::CallResult; -use comline_runtime::setup::{ - transport::{methods::tcp::provider::TcpProvider, provider::ProviderSetup}, - call_system::systems::json_rpc::JsonRPCv2, -}; - - -impl GreetProviderProtocol for GreetProvider { - fn greet(&self, name: &str) -> CallResult { - println!("[Server] Received a greet request with name '{}'", name); - - Ok("Hello ".to_owned() + name) - } -} - - -pub(crate) async fn main() { - println!("Running Server"); - - let (address, port) = ("127.0.0.1", "2620"); - let full_address = &*(address.to_owned() + ":" + port); - - let transporter = TcpProvider::with_address(full_address).await.unwrap(); - let mut setup = ProviderSetup::with_transporter(transporter) - .with_call_system::(JsonRPCv2::new) - .with_capability(GreetProvider::new) - ; - - respond_to_incoming_hellos(&mut setup).await; -} - -async fn respond_to_incoming_hellos(setup: &mut ProviderSetup) { - setup.transporter.read().unwrap() - .listen_incoming_connection() - .await; -} - diff --git a/runtime/core/tests/setups/mod.rs b/runtime/core/tests/setups/mod.rs deleted file mode 100644 index 1a38821..0000000 --- a/runtime/core/tests/setups/mod.rs +++ /dev/null @@ -1,5 +0,0 @@ -//#![feature(json_rpc)] - -// Relative modules -mod jrpc_tcp; -