queuey_core/
backend.rs

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

queuey_core/
backend.rs
