stream/
queue.rs

1use async_trait::async_trait;
2
3use crate::Result;
4
5/// Publishes items onto a lane. An adapter's producer is cheap to obtain, so
6/// the ingest engine holds one per lane.
7#[async_trait]
8pub trait Producer<T: Send>: std::fmt::Debug + Send + Sync {
9    /// Publish an item. May wait (backpressure) until the lane has room.
10    async fn publish(&self, item: T) -> Result<()>;
11}
12
13/// Pulls items from a lane for processing. Exactly one consumer is active per
14/// lane at a time; a restarted engine obtains a fresh one over the same lane
15/// and is redelivered whatever the previous one left unacknowledged.
16#[async_trait]
17pub trait Consumer<T: Send>: std::fmt::Debug + Send {
18    /// The next delivery, or `None` once the lane is closed and drained.
19    async fn recv(&mut self) -> Result<Option<Delivery<T>>>;
20
21    /// Whether nothing is waiting behind the last delivery — the sink engine's
22    /// *caught-up* hint, read at a batch boundary and forwarded to the sink's
23    /// flush so a destination with a cost to making writes visible can pay it
24    /// only when idle. A point-in-time snapshot; an adapter that cannot tell
25    /// returns `false`.
26    fn is_empty(&self) -> bool;
27}
28
29/// A received item paired with the handle that confirms or returns it.
30///
31/// Processing is complete only when the [`AckHandle`] taken from
32/// [`into_parts`](Self::into_parts) is acked; until then the adapter may
33/// redeliver after a restart. Dropping a delivery without acking leaves it
34/// unconfirmed (it is redelivered to the next consumer of the lane).
35#[derive(Debug)]
36pub struct Delivery<T> {
37    item: T,
38    handle: Box<dyn AckHandle>,
39}
40
41impl<T> Delivery<T> {
42    /// Build a delivery from an item and its adapter ack handle.
43    pub fn new(item: T, handle: Box<dyn AckHandle>) -> Self {
44        Self { item, handle }
45    }
46
47    /// Peek at the item without consuming the delivery.
48    pub fn item(&self) -> &T {
49        &self.item
50    }
51
52    /// Split into the item and its ack handle, so the item can be processed and
53    /// the handle acked once the work is durably done.
54    pub fn into_parts(self) -> (T, Box<dyn AckHandle>) {
55        (self.item, self.handle)
56    }
57
58    /// Return the item to the lane for redelivery (processing failed).
59    pub async fn nack(self) -> Result<()> {
60        self.handle.nack().await
61    }
62}
63
64/// Confirms or returns a single delivery. Adapter-specific: an in-memory
65/// bookkeeping step for the channel, a server acknowledgement for a broker.
66/// Acking a lane batch is also what moves that lane's watermark.
67#[async_trait]
68pub trait AckHandle: std::fmt::Debug + Send {
69    /// Confirm the delivery; it will not be redelivered.
70    async fn ack(self: Box<Self>) -> Result<()>;
71
72    /// Return the delivery for redelivery.
73    async fn nack(self: Box<Self>) -> Result<()>;
74}
stream/queue.rs

stream/
queue.rs
