cortexkit_bus_trait/
stream.rs

1use std::time::Duration;
2
3use async_trait::async_trait;
4
5use crate::{BusResult, ContentDigest, Headers, Message};
6
7/// The server's decision about a publish, including deduplicated retries.
8#[derive(Debug, Clone, Copy, PartialEq, Eq)]
9#[non_exhaustive]
10#[must_use = "check duplicate() to distinguish a stored message from an idempotent retry or an id collision"]
11pub struct PublishReceipt {
12    stream_sequence: u64,
13    duplicate: bool,
14}
15
16impl PublishReceipt {
17    pub fn new(stream_sequence: u64, duplicate: bool) -> Self {
18        Self {
19            stream_sequence,
20            duplicate,
21        }
22    }
23
24    /// The stored message's sequence, or the original sequence for a duplicate.
25    pub fn stream_sequence(&self) -> u64 {
26        self.stream_sequence
27    }
28
29    /// Whether this id was already stored within the stream's deduplication window.
30    pub fn duplicate(&self) -> bool {
31        self.duplicate
32    }
33}
34
35#[derive(Debug, Clone, PartialEq, Eq)]
36pub struct StreamDelivery {
37    pub stream_seq: u64,
38    /// How many times this message has been delivered to this durable,
39    /// counting this delivery: 1 on the first delivery, and one more after
40    /// every nak or ack-wait expiry. JetStream reports it as `num_delivered`.
41    ///
42    /// On a stream whose max-deliver is unlimited this is the only way a
43    /// consumer can tell a message that keeps failing from a new one, so it is
44    /// what a consumer uses to give up and [`StreamCursor::term`] it.
45    pub delivery_count: u64,
46    pub message: Message,
47}
48
49/// A cursor for an already-created durable, selected by its associated registry identity.
50///
51/// At most one delivery is in flight per cursor: `ack`, `nak` and `term` each
52/// settle it, after which `next` may be called again. Each of them, and
53/// `in_progress`, returns `Ok` only once the server has recorded it.
54#[async_trait]
55pub trait StreamCursor: Send {
56    async fn next(&mut self) -> BusResult<Option<StreamDelivery>>;
57    async fn ack(&mut self) -> BusResult<()>;
58    async fn nak(&mut self, delay: Duration) -> BusResult<()>;
59
60    /// Settles the current delivery terminally: it is not acknowledged as
61    /// processed and it is never redelivered (JetStream `+TERM`). `reason`
62    /// is recorded with the termination where the backend can carry one
63    /// (JetStream's `+TERM <reason>`), for example `body_absent`.
64    async fn term(&mut self, reason: Option<&str>) -> BusResult<()>;
65
66    /// Reports that the current delivery is still being worked on (JetStream
67    /// `+WPI`): the ack-wait clock restarts from now, so a handler slower than
68    /// one ack wait is not redelivered under it. The delivery stays in flight
69    /// and must still be settled with `ack`, `nak` or `term`.
70    async fn in_progress(&mut self) -> BusResult<()>;
71}
72
73/// Whose durable a cursor binds: an agent or a module, with its id. The
74/// backend derives the durable name from the kind and the id (`c_` or `m_`
75/// followed by the id), so a cursor can bind only an agent's or a module's
76/// durable and never an arbitrary consumer name.
77#[derive(Debug, Clone, Copy, PartialEq, Eq)]
78pub enum DurableOwner<'a> {
79    /// An agent's durable on an agent stream (WAKE, PEER, EFFECT), named
80    /// `c_{agent_id}`.
81    Agent(&'a str),
82    /// A module's durable on a non-agent stream (EVENT, or prefrontal-core's
83    /// one durable on ROOM), named `m_{module_id}`.
84    Module(&'a str),
85}
86
87#[async_trait]
88pub trait Stream: Send + Sync {
89    type Cursor: StreamCursor;
90
91    /// Publishes a message, returning whether it was stored or deduplicated.
92    ///
93    /// A duplicate means the stream already stored this id within its
94    /// deduplication window; the receipt carries that original message's sequence
95    /// and no new message was stored. Callers may accept it as an idempotent retry,
96    /// or treat it as an id collision when a new message was intended. Use distinct
97    /// ids for distinct messages, even across subjects in the same stream.
98    async fn publish(
99        &self,
100        subject: &str,
101        id: &str,
102        digest: ContentDigest,
103        headers: Headers,
104    ) -> BusResult<PublishReceipt>;
105
106    /// Attaches to the owner's durable, created separately; this client cannot
107    /// create workload consumers.
108    async fn consumer(&self, owner: DurableOwner<'_>) -> BusResult<Self::Cursor>;
109}
cortexkit_bus_trait/stream.rs

cortexkit_bus_trait/
stream.rs
