use alloc::{boxed::Box, string::String, sync::Arc, vec::Vec};
use core::future::Future;
use core::pin::Pin;
use futures_core::Stream;
#[cfg(feature = "connector-session")]
mod client;
#[cfg(feature = "connector-session")]
mod connector;
#[cfg(feature = "connector-session")]
mod endpoint;
#[cfg(feature = "connector-session")]
mod io;
#[cfg(feature = "connector-session")]
mod pump;
#[cfg(feature = "connector-session")]
mod server;
#[cfg(all(feature = "connector-session", feature = "remote"))]
pub mod aimx;
mod topic_match;
pub use topic_match::{is_wildcard, pattern_contains, topic_matches};
#[cfg(feature = "connector-session")]
pub use client::{pump_client, run_client, ClientConfig, ClientHandle};
#[cfg(feature = "connector-session")]
pub use connector::{SessionClientConnector, SessionServerConnector};
#[cfg(feature = "connector-session")]
pub use endpoint::{split_host_port, split_host_port_opt, EndpointError};
#[cfg(feature = "connector-session")]
pub use io::{
ByteRead, ByteStream, ByteWrite, Datagram, DatagramBinder, Delay, FrameFault, FramedConnection,
Framer, FramerFactory, FramingDialer, FramingListener, IoError, OneShot, OneShotDialer,
OneShotListener, StreamDialer, StreamListener,
};
#[cfg(feature = "connector-session")]
pub use pump::{pump_sink, pump_source};
#[cfg(feature = "connector-session")]
pub use server::{run_session, serve, SessionConfig};
pub type BoxFut<'a, T> = Pin<Box<dyn Future<Output = T> + Send + 'a>>;
pub type BoxStream<'a, T> = Pin<Box<dyn Stream<Item = T> + Send + 'a>>;
pub type Payload = Arc<[u8]>;
#[derive(Clone, Debug)]
pub struct SubUpdate {
pub topic: Option<Arc<str>>,
pub data: Payload,
pub skipped: u64,
pub snapshot_end: bool,
}
impl SubUpdate {
pub fn new(data: Payload) -> Self {
Self {
topic: None,
data,
skipped: 0,
snapshot_end: false,
}
}
pub fn tagged(topic: Arc<str>, data: Payload) -> Self {
Self {
topic: Some(topic),
data,
skipped: 0,
snapshot_end: false,
}
}
pub fn with_skipped(mut self, n: u64) -> Self {
self.skipped = n;
self
}
pub fn with_snapshot_end(mut self) -> Self {
self.snapshot_end = true;
self
}
}
pub type TransportResult<T> = Result<T, TransportError>;
#[derive(Clone, Default)]
#[non_exhaustive]
pub struct PeerInfo {
pub peer_addr: Option<String>,
pub ext: Option<Arc<dyn core::any::Any + Send + Sync>>,
}
impl PeerInfo {
pub fn with_ext(mut self, ext: Arc<dyn core::any::Any + Send + Sync>) -> Self {
self.ext = Some(ext);
self
}
pub fn ext_as<T: core::any::Any + Send + Sync>(&self) -> Option<Arc<T>> {
self.ext.clone()?.downcast::<T>().ok()
}
}
impl core::fmt::Debug for PeerInfo {
fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
f.debug_struct("PeerInfo")
.field("peer_addr", &self.peer_addr)
.field("ext", &self.ext.as_ref().map(|_| "<opaque>"))
.finish()
}
}
#[derive(Clone, Default)]
#[non_exhaustive]
pub struct SessionCtx {
pub ext: Option<Arc<dyn core::any::Any + Send + Sync>>,
}
impl SessionCtx {
pub fn with_ext(ext: Arc<dyn core::any::Any + Send + Sync>) -> Self {
Self { ext: Some(ext) }
}
pub fn ext_as<T: core::any::Any + Send + Sync>(&self) -> Option<Arc<T>> {
self.ext.clone()?.downcast::<T>().ok()
}
}
impl core::fmt::Debug for SessionCtx {
fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
f.debug_struct("SessionCtx")
.field("ext", &self.ext.as_ref().map(|_| "<opaque>"))
.finish()
}
}
#[derive(Debug, Clone)]
pub struct SessionLimits {
pub max_connections: usize,
pub max_subs_per_connection: usize,
}
impl Default for SessionLimits {
fn default() -> Self {
Self {
max_connections: 16,
max_subs_per_connection: 32,
}
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
#[non_exhaustive]
pub enum TransportError {
Closed,
Io,
Framing,
Busy,
}
#[derive(Debug, Clone, PartialEq, Eq)]
#[non_exhaustive]
pub enum CodecError {
Malformed,
}
#[derive(Debug, Clone, PartialEq, Eq)]
#[non_exhaustive]
pub enum RpcError {
NotFound,
Denied,
Internal,
VersionMismatch,
}
#[derive(Debug, Clone, PartialEq, Eq)]
#[non_exhaustive]
pub enum AuthError {
Unauthorized,
}
pub enum Inbound {
Request {
id: u64,
method: String,
params: Payload,
},
Subscribe {
id: u64,
topic: String,
},
Unsubscribe {
sub: String,
},
Write {
topic: String,
payload: Payload,
},
Ping,
}
pub enum Outbound<'a> {
Reply {
id: u64,
result: Result<Payload, RpcError>,
},
Event {
sub: &'a str,
seq: u64,
topic: Option<&'a str>,
data: Payload,
},
Snapshot {
sub: &'a str,
seq: u64,
last: bool,
topic: &'a str,
data: Payload,
},
Subscribed {
sub: &'a str,
},
Pong,
}
pub trait Connection: Send {
fn recv(&mut self) -> BoxFut<'_, TransportResult<Option<Vec<u8>>>>;
fn send<'a>(&'a mut self, frame: &'a [u8]) -> BoxFut<'a, TransportResult<()>>;
fn peer(&self) -> &PeerInfo;
}
pub trait Listener: Send {
fn accept(&mut self) -> BoxFut<'_, TransportResult<Box<dyn Connection>>>;
}
pub trait Dialer: Send {
fn connect(&self) -> BoxFut<'_, TransportResult<Box<dyn Connection>>>;
}
impl Dialer for Box<dyn Dialer> {
fn connect(&self) -> BoxFut<'_, TransportResult<Box<dyn Connection>>> {
(**self).connect()
}
}
pub trait Dispatch: Send + Sync {
fn authenticate<'a>(
&'a self,
peer: &'a PeerInfo,
first: Option<&'a [u8]>,
) -> BoxFut<'a, Result<SessionCtx, AuthError>>;
fn open(&self, ctx: &SessionCtx) -> Box<dyn Session>;
}
pub trait Session: Send {
fn call<'a>(
&'a mut self,
method: &'a str,
params: Payload,
) -> BoxFut<'a, Result<Payload, RpcError>>;
fn subscribe<'a>(
&'a mut self,
topic: &'a str,
) -> BoxFut<'a, Result<BoxStream<'static, SubUpdate>, RpcError>> {
let _ = topic;
Box::pin(async { Err(RpcError::NotFound) })
}
fn snapshots(&mut self, topic: &str) -> Vec<(String, Payload)> {
let _ = topic;
Vec::new()
}
fn write<'a>(
&'a mut self,
topic: &'a str,
payload: Payload,
) -> BoxFut<'a, Result<(), RpcError>>;
}
pub trait EnvelopeCodec: Send + Sync {
fn decode(&self, frame: &[u8]) -> Result<Inbound, CodecError>;
fn encode(&self, msg: Outbound<'_>, out: &mut Vec<u8>) -> Result<(), CodecError>;
fn encode_inbound(&self, msg: Inbound, out: &mut Vec<u8>) -> Result<(), CodecError>;
fn decode_outbound<'a>(&self, frame: &'a [u8]) -> Result<Outbound<'a>, CodecError>;
}
pub trait Source: Send {
fn next(&mut self) -> BoxFut<'_, Option<(String, Payload)>>;
}
#[allow(dead_code, clippy::too_many_arguments)]
fn _assert_object_safe(
_connection: &dyn Connection,
_listener: &dyn Listener,
_dialer: &dyn Dialer,
_dispatch: &dyn Dispatch,
_session: &dyn Session,
_codec: &dyn EnvelopeCodec,
_source: &dyn Source,
) {
}
#[cfg(test)]
mod tests {
use super::*;
struct MockConnection;
impl Connection for MockConnection {
fn recv(&mut self) -> BoxFut<'_, TransportResult<Option<Vec<u8>>>> {
unimplemented!()
}
fn send<'a>(&'a mut self, _frame: &'a [u8]) -> BoxFut<'a, TransportResult<()>> {
unimplemented!()
}
fn peer(&self) -> &PeerInfo {
unimplemented!()
}
}
struct MockListener;
impl Listener for MockListener {
fn accept(&mut self) -> BoxFut<'_, TransportResult<Box<dyn Connection>>> {
unimplemented!()
}
}
struct MockDialer;
impl Dialer for MockDialer {
fn connect(&self) -> BoxFut<'_, TransportResult<Box<dyn Connection>>> {
unimplemented!()
}
}
struct MockDispatch;
impl Dispatch for MockDispatch {
fn authenticate<'a>(
&'a self,
_peer: &'a PeerInfo,
_first: Option<&'a [u8]>,
) -> BoxFut<'a, Result<SessionCtx, AuthError>> {
unimplemented!()
}
fn open(&self, _ctx: &SessionCtx) -> Box<dyn Session> {
unimplemented!()
}
}
struct MockSession;
impl Session for MockSession {
fn call<'a>(
&'a mut self,
_method: &'a str,
_params: Payload,
) -> BoxFut<'a, Result<Payload, RpcError>> {
unimplemented!()
}
fn subscribe<'a>(
&'a mut self,
_topic: &'a str,
) -> BoxFut<'a, Result<BoxStream<'static, SubUpdate>, RpcError>> {
unimplemented!()
}
fn write<'a>(
&'a mut self,
_topic: &'a str,
_payload: Payload,
) -> BoxFut<'a, Result<(), RpcError>> {
unimplemented!()
}
}
struct MockCodec;
impl EnvelopeCodec for MockCodec {
fn decode(&self, _frame: &[u8]) -> Result<Inbound, CodecError> {
unimplemented!()
}
fn encode(&self, _msg: Outbound<'_>, _out: &mut Vec<u8>) -> Result<(), CodecError> {
unimplemented!()
}
fn encode_inbound(&self, _msg: Inbound, _out: &mut Vec<u8>) -> Result<(), CodecError> {
unimplemented!()
}
fn decode_outbound<'a>(&self, _frame: &'a [u8]) -> Result<Outbound<'a>, CodecError> {
unimplemented!()
}
}
struct MockSource;
impl Source for MockSource {
fn next(&mut self) -> BoxFut<'_, Option<(String, Payload)>> {
unimplemented!()
}
}
#[test]
fn traits_are_object_safe() {
let _connection: Box<dyn Connection> = Box::new(MockConnection);
let _listener: Box<dyn Listener> = Box::new(MockListener);
let _dialer: Box<dyn Dialer> = Box::new(MockDialer);
let _dispatch: Box<dyn Dispatch> = Box::new(MockDispatch);
let _session: Box<dyn Session> = Box::new(MockSession);
let _codec: Box<dyn EnvelopeCodec> = Box::new(MockCodec);
let _source: Box<dyn Source> = Box::new(MockSource);
}
#[test]
fn boxed_dialer_is_a_dialer() {
fn takes_dialer<D: Dialer + 'static>(_d: D) {}
let boxed: Box<dyn Dialer> = Box::new(MockDialer);
takes_dialer(boxed);
}
}