1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
//! Transport abstraction: [`Backend`], [`Delivery`] and [`DeliveryStream`].
use std::{pin::Pin, time::Duration};
use async_trait::async_trait;
use futures::Stream;
use crate::{envelope::Envelope, error::Result, queue::QueueConfig};
/// A message received from the broker. Must be acked or nacked exactly once.
#[async_trait]
pub trait Delivery: Send + 'static {
/// The message this delivery carries.
fn envelope(&self) -> &Envelope;
/// Successfully processed; remove from broker.
async fn ack(self: Box<Self>) -> Result<()>;
/// Failed permanently (or attempts exhausted); route to dead-letter storage.
async fn dead_letter(self: Box<Self>, reason: &str) -> Result<()>;
/// Failed transiently; schedule `next` (already `attempt + 1`) to be redelivered
/// after `delay`. Implementations must ack the original *after* the retry is
/// durably scheduled so no message is lost.
async fn retry(self: Box<Self>, next: Envelope, delay: Duration) -> Result<()>;
/// Did not fail, but must run again in `delay`: durably schedule `next` (already
/// `deferrals + 1` with its priority set, `attempt` unchanged) to reappear on
/// `next.queue`, **then** ack the original.
///
/// Same "publish before ack" rule as [`Delivery::retry`]: if the scheduling fails,
/// the original must be left unacked so the broker redelivers it.
///
/// How the message is held is backend-specific (a dedicated hold queue per delay
/// on RabbitMQ, a timer in `MemoryBackend`), but the observable contract is the
/// same: nothing is delivered before `delay` has passed, and when it comes back it
/// carries `next.priority`, so it overtakes normally enqueued work on a queue that
/// supports priorities. See [`crate::JobError::Deferred`].
async fn defer(self: Box<Self>, next: Envelope, delay: Duration) -> Result<()>;
}
/// Stream of deliveries produced by [`Backend::consume`].
pub type DeliveryStream = Pin<Box<dyn Stream<Item = Result<Box<dyn Delivery>>> + Send>>;
/// A transport. Implementations: `MemoryBackend` (this crate), `RabbitMqBackend`.
#[async_trait]
pub trait Backend: Send + Sync + 'static {
/// Idempotently create all queues (plus any retry / dead-letter infrastructure).
async fn declare(&self, queues: &[QueueConfig]) -> Result<()>;
/// Publish `envelope` to `envelope.queue`, optionally delayed.
async fn publish(&self, envelope: &Envelope, delay: Option<Duration>) -> Result<()>;
/// Publish `envelope` into a hold that releases it onto `envelope.queue` after
/// `delay`. This is the publish half of [`Delivery::defer`], also used by
/// [`crate::Producer::defer`].
///
/// Differs from `publish` with a delay only in intent, and backends may treat
/// the two differently in detail: a deferral is expected to carry its queue's
/// top priority so it overtakes the backlog when it returns, and on RabbitMQ the
/// delay is rounded to a separate, typically finer, granularity because a
/// `Retry-After` is a contract while a backoff is a heuristic. Neither path
/// releases a job early, and neither lets different delays block each other.
///
/// Hold naming and lifetime are backend-specific.
async fn defer(&self, envelope: &Envelope, delay: Duration) -> Result<()>;
/// Start consuming `queue` with the given prefetch. The stream ends when the
/// backend is closed or the connection is lost.
async fn consume(&self, queue: &QueueConfig) -> Result<DeliveryStream>;
/// Graceful shutdown: stop all consumers, flush, close connections.
async fn close(&self) -> Result<()>;
}