cortexkit_bus_trait/
stream.rs

1use std::time::Duration;
2
3use async_trait::async_trait;
4
5use crate::{BusResult, ContentDigest, Headers, Message};
6
7#[derive(Debug, Clone, Copy, PartialEq, Eq)]
8pub struct PublishAck {
9    pub stream_seq: u64,
10}
11
12#[derive(Debug, Clone, PartialEq, Eq)]
13pub struct StreamDelivery {
14    pub stream_seq: u64,
15    /// How many times this message has been delivered to this durable,
16    /// counting this delivery: 1 on the first delivery, and one more after
17    /// every nak or ack-wait expiry. JetStream reports it as `num_delivered`.
18    ///
19    /// On a stream whose max-deliver is unlimited this is the only way a
20    /// consumer can tell a message that keeps failing from a new one, so it is
21    /// what a consumer uses to give up and [`StreamCursor::term`] it.
22    pub delivery_count: u64,
23    pub message: Message,
24}
25
26/// A cursor for an already-created durable, selected by its associated registry identity.
27///
28/// At most one delivery is in flight per cursor: `ack`, `nak` and `term` each
29/// settle it, after which `next` may be called again. Each of them, and
30/// `in_progress`, returns `Ok` only once the server has recorded it.
31#[async_trait]
32pub trait StreamCursor: Send {
33    async fn next(&mut self) -> BusResult<Option<StreamDelivery>>;
34    async fn ack(&mut self) -> BusResult<()>;
35    async fn nak(&mut self, delay: Duration) -> BusResult<()>;
36
37    /// Settles the current delivery terminally: it is not acknowledged as
38    /// processed and it is never redelivered (JetStream `+TERM`). `reason`
39    /// is recorded with the termination where the backend can carry one
40    /// (JetStream's `+TERM <reason>`), for example `body_absent`.
41    async fn term(&mut self, reason: Option<&str>) -> BusResult<()>;
42
43    /// Reports that the current delivery is still being worked on (JetStream
44    /// `+WPI`): the ack-wait clock restarts from now, so a handler slower than
45    /// one ack wait is not redelivered under it. The delivery stays in flight
46    /// and must still be settled with `ack`, `nak` or `term`.
47    async fn in_progress(&mut self) -> BusResult<()>;
48}
49
50/// Whose durable a cursor binds: an agent or a module, with its id. The
51/// backend derives the durable name from the kind and the id (`c_` or `m_`
52/// followed by the id), so a cursor can bind only an agent's or a module's
53/// durable and never an arbitrary consumer name.
54#[derive(Debug, Clone, Copy, PartialEq, Eq)]
55pub enum DurableOwner<'a> {
56    /// An agent's durable on an agent stream (WAKE, PEER, EFFECT), named
57    /// `c_{agent_id}`.
58    Agent(&'a str),
59    /// A module's durable on a non-agent stream (EVENT, or prefrontal-core's
60    /// one durable on ROOM), named `m_{module_id}`.
61    Module(&'a str),
62}
63
64#[async_trait]
65pub trait Stream: Send + Sync {
66    type Cursor: StreamCursor;
67
68    async fn publish(
69        &self,
70        subject: &str,
71        id: &str,
72        digest: ContentDigest,
73        headers: Headers,
74    ) -> BusResult<PublishAck>;
75
76    /// Attaches to the owner's durable, created separately; this client cannot
77    /// create workload consumers.
78    async fn consumer(&self, owner: DurableOwner<'_>) -> BusResult<Self::Cursor>;
79}
cortexkit_bus_trait/stream.rs

cortexkit_bus_trait/
stream.rs
