aimdb-core 2.0.0

Type-safe async data pipelines — one Rust codebase from MCU to cloud
Documentation
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
//! Connector-session substrate — the shared machinery for transport-based
//! remote access.
//!
//! Three layers: transport ([`Connection`]/[`Listener`]/[`Dialer`]), codec
//! ([`EnvelopeCodec`]), and dispatch ([`Dispatch`]/[`Session`]), over a
//! role-neutral [`Inbound`]/[`Outbound`] message set shared by the reactive
//! server engine (`serve`/`run_session`) and the proactive client engine
//! (`run_client`/`pump_client`). Data-plane connectors use `pump_sink`/
//! `pump_source` over the [`Source`] / [`Connector`](crate::transport::Connector)
//! capabilities.
//!
//! All contracts are `dyn`-safe and compile on `std` and `no_std + alloc`.

use alloc::{boxed::Box, string::String, sync::Arc, vec::Vec};
use core::future::Future;
use core::pin::Pin;

use futures_core::Stream;

// The engines are runtime-neutral (`futures` channels + the adapter's `RuntimeOps`
// clock, no `tokio`/`embassy-*`), so they cross-compile to `no_std + alloc`.
#[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;

// Concrete AimX protocol substrate. The transport lives in a separate connector
// crate (`aimdb-uds-connector`) — core keeps the protocol plus the generic
// [`SessionClientConnector`] / [`SessionServerConnector`] spine.
#[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};

// ===========================================================================
// Shared aliases
// ===========================================================================

/// Boxed, `Send` future — the object-safe async return shape used throughout.
pub type BoxFut<'a, T> = Pin<Box<dyn Future<Output = T> + Send + 'a>>;

/// Boxed, `Send` stream — the reply shape of a subscription ([`Session::subscribe`]).
pub type BoxStream<'a, T> = Pin<Box<dyn Stream<Item = T> + Send + 'a>>;

/// A serialized record value, carried opaquely through the codec.
///
/// `Arc<[u8]>` so fan-out is a refcount bump. Bytes stay opaque until a handler
/// inspects them.
pub type Payload = Arc<[u8]>;

/// One update delivered on a subscription stream (server [`Session::subscribe`]
/// and client [`ClientHandle::subscribe`] alike).
///
/// `topic` names the concrete record that fired: `Some` on wildcard
/// subscriptions, which fan in many records under one subscription id, and on
/// any transport that tags every event; `None` on an exact-topic subscription.
#[derive(Clone, Debug)]
pub struct SubUpdate {
    /// Concrete record topic that fired, when the producer side tags it.
    pub topic: Option<Arc<str>>,
    /// The serialized record value.
    pub data: Payload,
    /// Count of skipped records
    pub skipped: u64,
    /// Set on the last update of the late-join snapshot burst, carrying its total `skipped`.
    ///
    /// Only set when the burst produced a final update to carry it — an empty
    /// burst, an exact-topic subscribe, or a tail that failed to encode end
    /// without one — so its absence says nothing.
    pub snapshot_end: bool,
}

impl SubUpdate {
    /// An untagged update (exact-topic subscription).
    pub fn new(data: Payload) -> Self {
        Self {
            topic: None,
            data,
            skipped: 0,
            snapshot_end: false,
        }
    }

    /// A topic-tagged update (wildcard / bus subscription).
    pub fn tagged(topic: Arc<str>, data: Payload) -> Self {
        Self {
            topic: Some(topic),
            data,
            skipped: 0,
            snapshot_end: false,
        }
    }

    /// Mark that `n` updates were lost on this subscription immediately before
    /// this one (`0` leaves the update lossless).
    pub fn with_skipped(mut self, n: u64) -> Self {
        self.skipped = n;
        self
    }

    /// Mark this update as the last of the late-join snapshot burst (see
    /// [`snapshot_end`](Self::snapshot_end)).
    pub fn with_snapshot_end(mut self) -> Self {
        self.snapshot_end = true;
        self
    }
}

/// Result of a transport-layer operation.
pub type TransportResult<T> = Result<T, TransportError>;

// ===========================================================================
// Supporting types (stubs — sufficient for the signatures to compile)
// ===========================================================================

/// Remote-peer metadata carried by a [`Connection`].
///
/// A neutral [`peer_addr`](Self::peer_addr) plus a type-erased
/// [`ext`](Self::ext) slot a connector fills with its own resolved identity
/// (e.g. WS attaches `ClientInfo` at the HTTP upgrade), keeping core
/// connector-agnostic. Downcast `ext` with [`ext_as`](Self::ext_as).
#[derive(Clone, Default)]
#[non_exhaustive]
pub struct PeerInfo {
    /// Remote address, if the transport exposes one.
    pub peer_addr: Option<String>,
    /// Connector-resolved identity, type-erased so core need not know the
    /// connector's auth types. Downcast with [`ext_as`](Self::ext_as).
    pub ext: Option<Arc<dyn core::any::Any + Send + Sync>>,
}

impl PeerInfo {
    /// Attach a connector-resolved identity (consumed by [`Dispatch::authenticate`]).
    pub fn with_ext(mut self, ext: Arc<dyn core::any::Any + Send + Sync>) -> Self {
        self.ext = Some(ext);
        self
    }

    /// Downcast the [`ext`](Self::ext) identity to a concrete connector type.
    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()
    }
}

/// The authenticated session context produced by [`Dispatch::authenticate`] and
/// threaded into [`Dispatch::open`].
///
/// Carries the resolved principal as a type-erased [`ext`](Self::ext) for
/// per-operation authorization in the [`Session`]; downcast with
/// [`ext_as`](Self::ext_as). Connectors that don't authenticate leave it `None`.
#[derive(Clone, Default)]
#[non_exhaustive]
pub struct SessionCtx {
    /// The resolved principal, type-erased. Downcast with [`ext_as`](Self::ext_as).
    pub ext: Option<Arc<dyn core::any::Any + Send + Sync>>,
}

impl SessionCtx {
    /// Build a context carrying a connector-resolved principal.
    pub fn with_ext(ext: Arc<dyn core::any::Any + Send + Sync>) -> Self {
        Self { ext: Some(ext) }
    }

    /// Downcast the [`ext`](Self::ext) principal to a concrete connector type.
    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()
    }
}

/// Per-session resource bounds consumed by the engines.
#[derive(Debug, Clone)]
pub struct SessionLimits {
    /// Maximum concurrently served connections.
    pub max_connections: usize,
    /// Maximum live subscriptions per connection.
    pub max_subs_per_connection: usize,
}

impl Default for SessionLimits {
    fn default() -> Self {
        Self {
            max_connections: 16,
            max_subs_per_connection: 32,
        }
    }
}

/// Transport-layer failure (Layer 1).
#[derive(Debug, Clone, PartialEq, Eq)]
#[non_exhaustive]
pub enum TransportError {
    /// The connection was closed or reset by the peer.
    Closed,
    /// An underlying I/O operation failed.
    Io,
    /// The byte stream could not be framed, and the framer cannot resynchronize.
    Framing,
    /// The transport's one endpoint resource is already in use — a second dial
    /// on a single-socket transport while the first connection is live.
    Busy,
}

/// Envelope-codec failure — a frame could not be decoded/encoded.
#[derive(Debug, Clone, PartialEq, Eq)]
#[non_exhaustive]
pub enum CodecError {
    /// The frame was not valid for this envelope format.
    Malformed,
}

/// Dispatch-layer (application) failure for `call` / `subscribe` / `write`.
#[derive(Debug, Clone, PartialEq, Eq)]
#[non_exhaustive]
pub enum RpcError {
    /// No such method or topic.
    NotFound,
    /// The caller lacks permission for this operation.
    Denied,
    /// The handler failed.
    Internal,
    /// The peer's declared protocol version is incompatible with this server's
    /// `PROTOCOL_VERSION`. Raised at the handshake, before `hello` completes.
    VersionMismatch,
}

/// Authentication failure raised by [`Dispatch::authenticate`].
#[derive(Debug, Clone, PartialEq, Eq)]
#[non_exhaustive]
pub enum AuthError {
    /// Credentials were missing or rejected.
    Unauthorized,
}

// ===========================================================================
// Logical message set — role-neutral: the server's `Inbound` is the client's
// outbound and vice versa.
// ===========================================================================

/// A logical request arriving over a [`Connection`] (what the server receives).
pub enum Inbound {
    /// An RPC call expecting a single [`Outbound::Reply`].
    Request {
        /// Correlation id, echoed in the reply.
        id: u64,
        /// Method name (e.g. `"record.set"`, `"query"`).
        method: String,
        /// Unparsed method parameters.
        params: Payload,
    },
    /// Open a subscription producing many [`Outbound::Event`]s.
    Subscribe {
        /// Correlation id for the subscription handshake.
        id: u64,
        /// Topic to subscribe to.
        topic: String,
    },
    /// Close a previously opened subscription.
    Unsubscribe {
        /// Subscription id to cancel.
        sub: String,
    },
    /// A fire-and-forget write (no reply).
    Write {
        /// Destination topic.
        topic: String,
        /// Unparsed record value.
        payload: Payload,
    },
    /// Keepalive.
    Ping,
}

/// A logical message sent back over a [`Connection`] (what the server emits).
pub enum Outbound<'a> {
    /// Reply to an [`Inbound::Request`].
    Reply {
        /// Correlation id of the originating request.
        id: u64,
        /// The result, or an [`RpcError`].
        result: Result<Payload, RpcError>,
    },
    /// A subscription update.
    Event {
        /// Subscription id this event belongs to.
        sub: &'a str,
        /// Monotonic sequence number.
        seq: u64,
        /// Concrete record topic that fired, when the subscription tags it
        /// (wildcard subscriptions fan in many records — see [`SubUpdate`]).
        topic: Option<&'a str>,
        /// Unparsed record value.
        data: Payload,
    },
    /// An initial snapshot emitted when a subscription opens (late-join).
    Snapshot {
        /// Subscription id the snapshot belongs to, so the client demux routes
        /// it like an [`Event`](Outbound::Event) (a wildcard subscription's
        /// snapshots carry topics that don't string-match the pattern key).
        sub: &'a str,
        /// Monotonic sequence number, in the **same space** as this
        /// subscription's [`Event`](Outbound::Event)s: the burst is numbered
        /// `1..=N` and the first event continues at `N + 1`, so a lost snapshot
        /// surfaces in the next update's [`SubUpdate::skipped`].
        seq: u64,
        /// Set on the last snapshot of the burst, so the loss total lands on an
        /// update the subscriber is guaranteed to see. The client engine
        /// reserves a sink slot for this frame.
        last: bool,
        /// Topic the snapshot is for.
        topic: &'a str,
        /// Unparsed record value.
        data: Payload,
    },
    /// An explicit acknowledgement that a subscription opened. Emitted by
    /// [`run_session`] only when [`SessionConfig::acks_subscribe`] is set.
    /// The `sub` is the subscription's routing id — the same value that tags its
    /// [`Event`](Outbound::Event)s.
    Subscribed {
        /// Subscription id that was opened.
        sub: &'a str,
    },
    /// Keepalive response.
    Pong,
}

// ===========================================================================
// Layer 1 — transport. Framing lives in the transport: `recv` returns one
// logical frame. `Dialer` is the client-side dual of `Listener`.
// ===========================================================================

/// A framed, bidirectional pipe — role-neutral (yielded by either
/// [`Listener::accept`] or [`Dialer::connect`]).
pub trait Connection: Send {
    /// Receive one logical frame. `Ok(None)` signals the peer closed.
    fn recv(&mut self) -> BoxFut<'_, TransportResult<Option<Vec<u8>>>>;

    /// Send one logical frame.
    fn send<'a>(&'a mut self, frame: &'a [u8]) -> BoxFut<'a, TransportResult<()>>;

    /// Peer metadata (remote addr, headers, pre-resolved auth).
    fn peer(&self) -> &PeerInfo;
}

/// The accepting (server) side — produces [`Connection`]s we did not initiate.
pub trait Listener: Send {
    /// Accept the next inbound connection.
    fn accept(&mut self) -> BoxFut<'_, TransportResult<Box<dyn Connection>>>;
}

/// The initiating (client) side — the dual of [`Listener`]; dials out and
/// produces the same [`Connection`].
pub trait Dialer: Send {
    /// Open a connection to the configured remote.
    fn connect(&self) -> BoxFut<'_, TransportResult<Box<dyn Connection>>>;
}

/// A boxed dialer is itself a [`Dialer`], so a runtime-selected
/// `Box<dyn Dialer>` — from a `scheme://` URL resolver, say — can be passed
/// where [`run_client`]`<D: Dialer>` is expected. The `Send` supertrait makes
/// the box `Send + 'static`.
impl Dialer for Box<dyn Dialer> {
    fn connect(&self) -> BoxFut<'_, TransportResult<Box<dyn Connection>>> {
        (**self).connect()
    }
}

// ===========================================================================
// Layer 3 — dispatch. RPC and streaming unify in one per-connection role with
// three reply cardinalities: `call` (one) / `subscribe` (many) / `write` (none).
// Split across two traits: the shared, immutable [`Dispatch`] (one
// `Arc<dyn Dispatch>` per server) and the per-connection, `&mut`-threaded
// [`Session`] that can hold mutable state without a lock.
// ===========================================================================

/// The shared application dispatch: authenticate a connection, then open a
/// per-connection [`Session`]. One `Arc<dyn Dispatch>` is shared across every
/// accepted connection, so it stays `Send + Sync` and behind `&self`.
pub trait Dispatch: Send + Sync {
    /// Resolve a [`SessionCtx`] from peer metadata and/or the first frame
    /// (a pre-resolved identity in [`PeerInfo`], or an in-band Hello in `first`).
    fn authenticate<'a>(
        &'a self,
        peer: &'a PeerInfo,
        first: Option<&'a [u8]>,
    ) -> BoxFut<'a, Result<SessionCtx, AuthError>>;

    /// Open the per-connection [`Session`] once, after [`authenticate`](Self::authenticate). It owns
    /// the connection's mutable dispatch state that the shared `Arc<Self>` cannot
    /// hold behind `&self`; the engine threads `&mut` into it.
    fn open(&self, ctx: &SessionCtx) -> Box<dyn Session>;
}

/// The per-connection session: serves calls, subscriptions, and writes for one
/// accepted [`Connection`]. The engine owns the `Box<dyn Session>` and threads
/// `&mut self` into each method, so it can hold per-connection state without a
/// lock; the shared, immutable role stays on [`Dispatch`].
pub trait Session: Send {
    /// One-shot RPC: one request → one reply.
    fn call<'a>(
        &'a mut self,
        method: &'a str,
        params: Payload,
    ) -> BoxFut<'a, Result<Payload, RpcError>>;

    /// Open a subscription yielding many [`SubUpdate`]s. The stream is `'static`
    /// (it captures cloned handles) so it outlives the `&mut` borrow and lives in
    /// the engine. Async so a connector can await per-operation authorization.
    /// Defaulted to [`RpcError::NotFound`] for dispatches with no streaming.
    fn subscribe<'a>(
        &'a mut self,
        topic: &'a str,
    ) -> BoxFut<'a, Result<BoxStream<'static, SubUpdate>, RpcError>> {
        let _ = topic;
        Box::pin(async { Err(RpcError::NotFound) })
    }

    /// Late-join snapshots: one `(topic, value)` per record covered by `topic`
    /// (several for a wildcard subscription), each emitted as an
    /// [`Outbound::Snapshot`] right after a successful
    /// [`subscribe`](Session::subscribe) and before the first event. Defaulted
    /// to empty (no snapshots).
    ///
    /// The engine numbers the burst `1..=N` and starts the event stream at
    /// `N + 1`, so a snapshot dropped in transit shows up as
    /// [`SubUpdate::skipped`] on the next delivered update.
    ///
    /// The burst's last snapshot is flagged and reaches the subscriber as the
    /// one update with [`SubUpdate::snapshot_end`] set, carrying the burst's
    /// whole loss; the client engine reserves a sink slot so it survives a slow
    /// consumer.
    ///
    /// **Do not block on that flag.** It is absent when `topic` matches no
    /// records, and when the final snapshot fails to encode or its frame is
    /// rejected — the flag rides that frame. Loss accounting survives all of
    /// these; only the end-of-burst signal is lost.
    fn snapshots(&mut self, topic: &str) -> Vec<(String, Payload)> {
        let _ = topic;
        Vec::new()
    }

    /// Fire-and-forget write: no reply. Routes through the producer/arbiter path,
    /// so single-writer-per-key stays intact.
    fn write<'a>(
        &'a mut self,
        topic: &'a str,
        payload: Payload,
    ) -> BoxFut<'a, Result<(), RpcError>>;
}

/// The protocol-envelope codec: frame bytes ↔ the logical message set. Layered
/// above (and nesting) the record-value `JsonCodec`, so the wire format stays
/// pluggable; `decode`/`encode` keep the record value as an opaque [`Payload`]
/// sliced from / spliced into the frame.
///
/// Symmetric, so one codec object serves both engines: `decode`/`encode` are the
/// server direction (read requests / write replies) and
/// [`encode_inbound`](EnvelopeCodec::encode_inbound) /
/// [`decode_outbound`](EnvelopeCodec::decode_outbound) the client direction.
pub trait EnvelopeCodec: Send + Sync {
    /// Decode one frame into a logical [`Inbound`] message (server reads a request).
    fn decode(&self, frame: &[u8]) -> Result<Inbound, CodecError>;

    /// Encode a logical [`Outbound`] message, appending its bytes to `out`
    /// (server writes a reply/event).
    fn encode(&self, msg: Outbound<'_>, out: &mut Vec<u8>) -> Result<(), CodecError>;

    /// Encode a logical [`Inbound`] message, appending its bytes to `out`
    /// (client writes a request). The dual of [`decode`](EnvelopeCodec::decode).
    fn encode_inbound(&self, msg: Inbound, out: &mut Vec<u8>) -> Result<(), CodecError>;

    /// Decode one frame into a logical [`Outbound`] message (client reads a
    /// reply/event). The dual of [`encode`](EnvelopeCodec::encode); the result
    /// borrows the frame (`Outbound`'s `sub`/`topic` are `&str` slices into it).
    fn decode_outbound<'a>(&self, frame: &'a [u8]) -> Result<Outbound<'a>, CodecError>;
}

// ===========================================================================
// Data-plane capabilities — connectionless (an external library owns any
// session). The outbound `Sink` is the canonical
// [`Connector`](crate::transport::Connector); the inbound `Source` is below.
// ===========================================================================

/// External → AimDB data-plane: a stream of inbound frames, drained by
/// `pump_source`.
pub trait Source: Send {
    /// Yield the next `(topic, payload)`, or `None` when the source is done.
    fn next(&mut self) -> BoxFut<'_, Option<(String, Payload)>>;
}

// ===========================================================================
// Taking each trait as `&dyn Trait` forces the dyn-compatibility check on all
// targets, not just under `cargo test`.
// ===========================================================================

#[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!()
        }
    }

    /// Every trait is `dyn`-usable.
    #[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);
    }

    /// `Box<dyn Dialer>` satisfies the `Dialer` bound, so a runtime-selected
    /// dialer can be passed where `D: Dialer` is expected.
    #[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);
    }
}