aimdb_core/buffer/
reader.rs

1//! Consumer-facing reader handles.
2//!
3//! The [`BufferReader`] / [`JsonBufferReader`] SPIs are object-safe and
4//! poll-based so adapters can implement them without a per-message
5//! `Pin<Box<dyn Future>>` allocation. These handles restore the ergonomic
6//! `async fn recv().await` surface for consumers by wrapping the erased
7//! reader's `poll_*` method in [`core::future::poll_fn`] — which is `core`-only
8//! (no_std-clean), allocation-free, and `unsafe`-free.
9//!
10//! The wrapped future is `Send` because the boxed reader is `Send`.
11
12use alloc::boxed::Box;
13#[cfg(feature = "remote")]
14use alloc::vec::Vec;
15use core::future::poll_fn;
16
17use crate::buffer::BufferReader;
18use crate::DbError;
19
20#[cfg(feature = "remote")]
21use crate::buffer::JsonBufferReader;
22
23/// Owned, ergonomic handle over an erased [`BufferReader`].
24///
25/// Returned by `Consumer::subscribe`. This is the "boxed lane": one indirect
26/// call per `recv`, zero AimDB-added heap allocations per message. (The generic
27/// monomorphized `Reader<T, B>` fast lane remains dormant.)
28pub struct Reader<T: Clone + Send> {
29    inner: Box<dyn BufferReader<T> + Send>,
30}
31
32impl<T: Clone + Send> Reader<T> {
33    /// Wrap an erased reader in an ergonomic handle.
34    pub fn new(inner: Box<dyn BufferReader<T> + Send>) -> Self {
35        Self { inner }
36    }
37
38    /// Receive the next value, awaiting until one is available.
39    ///
40    /// Allocation-free: wraps the erased reader's
41    /// [`poll_recv`](BufferReader::poll_recv) via `core::future::poll_fn`.
42    ///
43    /// # Behavior by Buffer Type
44    /// - **SPMC Ring**: Returns next value, or `Lagged(n)` if fell behind
45    /// - **SingleLatest**: Waits for value change, returns most recent
46    /// - **Mailbox**: Waits for slot value, takes and clears it
47    pub async fn recv(&mut self) -> Result<T, DbError> {
48        poll_fn(|cx| self.inner.poll_recv(cx)).await
49    }
50
51    /// Non-blocking receive — returns immediately.
52    ///
53    /// Returns `Err(DbError::BufferEmpty)` if no pending values.
54    pub fn try_recv(&mut self) -> Result<T, DbError> {
55        self.inner.try_recv()
56    }
57}
58
59/// Owned, ergonomic handle over an erased [`JsonBufferReader`].
60///
61/// Returned by `subscribe_json`. Awaiting `recv_json` or
62/// `recv_json_bytes` wraps the corresponding poll method via
63/// `core::future::poll_fn` — no extra per-message future box.
64#[cfg(feature = "remote")]
65pub struct JsonReader {
66    inner: Box<dyn JsonBufferReader + Send>,
67}
68
69#[cfg(feature = "remote")]
70impl JsonReader {
71    /// Wrap an erased JSON reader in an ergonomic handle.
72    pub fn new(inner: Box<dyn JsonBufferReader + Send>) -> Self {
73        Self { inner }
74    }
75
76    /// Receive the next value as JSON, awaiting until one is available.
77    pub async fn recv_json(&mut self) -> Result<serde_json::Value, DbError> {
78        poll_fn(|cx| self.inner.poll_recv_json(cx)).await
79    }
80
81    /// Receive the next value as owned JSON bytes.
82    pub async fn recv_json_bytes(&mut self) -> Result<Vec<u8>, DbError> {
83        poll_fn(|cx| self.inner.poll_recv_json_bytes(cx)).await
84    }
85
86    /// Non-blocking receive as JSON — returns immediately.
87    ///
88    /// Returns `Err(DbError::BufferEmpty)` if no pending values.
89    pub fn try_recv_json(&mut self) -> Result<serde_json::Value, DbError> {
90        self.inner.try_recv_json()
91    }
92
93    /// Non-blocking receive as owned JSON bytes — returns immediately.
94    ///
95    /// Returns `Err(DbError::BufferEmpty)` if no pending values.
96    pub fn try_recv_json_bytes(&mut self) -> Result<Vec<u8>, DbError> {
97        self.inner.try_recv_json_bytes()
98    }
99}
aimdb_core/buffer/reader.rs

aimdb_core/buffer/
reader.rs
