cortexkit_bus_trait/
work_queue.rs

1use std::time::Duration;
2
3use async_trait::async_trait;
4
5use crate::{BusResult, ContentDigest, Headers, Message, PublishAck, Stream};
6
7pub const DEAD_LETTER_REASON_MAX_DELIVERIES: &str = "max_deliveries_exceeded";
8
9#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
10pub struct DeliveryToken(pub u64);
11
12#[derive(Debug, Clone, PartialEq, Eq)]
13pub struct WorkItem {
14    pub token: DeliveryToken,
15    pub message: Message,
16    pub delivery_count: u32,
17}
18
19#[derive(Debug, Clone, PartialEq, Eq)]
20pub struct MaxDeliveriesExceeded {
21    pub item: WorkItem,
22    pub max_deliveries: u32,
23}
24
25#[derive(Debug, Clone, PartialEq, Eq)]
26pub enum ClaimOutcome {
27    Item(WorkItem),
28    Empty,
29    MaxDeliveriesExceeded(MaxDeliveriesExceeded),
30}
31
32/// A claimant on a bounded work queue.
33///
34/// `ack`, `nak` and `term` return `Ok` only once the server has recorded the
35/// decision, so a claimant may exit straight after one without losing it.
36#[async_trait]
37pub trait WorkQueue: Send + Sync {
38    async fn claim(&self) -> BusResult<ClaimOutcome>;
39    async fn ack(&self, token: DeliveryToken) -> BusResult<()>;
40    async fn nak(&self, token: DeliveryToken, delay: Duration) -> BusResult<()>;
41    async fn term(&self, token: DeliveryToken) -> BusResult<()>;
42}
43
44#[derive(Debug, Clone, PartialEq, Eq)]
45pub struct DeadLetterRecord {
46    pub original_subject: String,
47    pub message_id: String,
48    pub digest: ContentDigest,
49    pub delivery_count: u32,
50    pub reason: String,
51}
52
53impl DeadLetterRecord {
54    pub fn from_exhaustion(exhausted: &MaxDeliveriesExceeded) -> Self {
55        Self {
56            original_subject: exhausted.item.message.subject.clone(),
57            message_id: exhausted.item.message.id.clone(),
58            digest: exhausted.item.message.digest,
59            delivery_count: exhausted.item.delivery_count,
60            reason: DEAD_LETTER_REASON_MAX_DELIVERIES.into(),
61        }
62    }
63
64    pub fn headers(&self) -> Headers {
65        Headers::from([
66            ("original-subject".into(), self.original_subject.clone()),
67            ("message-id".into(), self.message_id.clone()),
68            ("content-digest".into(), self.digest.to_string()),
69            ("delivery-count".into(), self.delivery_count.to_string()),
70            ("reason".into(), self.reason.clone()),
71        ])
72    }
73}
74
75pub fn effect_dead_subject(account: &str) -> String {
76    format!("ck.{account}.effect.dead")
77}
78
79/// Publishes a persistent dead-letter record before terminating the exhausted delivery.
80pub async fn terminally_dispose<S, Q>(
81    stream: &S,
82    queue: &Q,
83    account: &str,
84    exhausted: &MaxDeliveriesExceeded,
85    dead_letter_digest: ContentDigest,
86) -> BusResult<PublishAck>
87where
88    S: Stream,
89    Q: WorkQueue,
90{
91    let record = DeadLetterRecord::from_exhaustion(exhausted);
92    let ack = stream
93        .publish(
94            &effect_dead_subject(account),
95            &record.message_id,
96            dead_letter_digest,
97            record.headers(),
98        )
99        .await?;
100    queue.term(exhausted.item.token).await?;
101    Ok(ack)
102}
cortexkit_bus_trait/work_queue.rs

cortexkit_bus_trait/
work_queue.rs
