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: that one shares a single wait queue per
59    /// queue (on RabbitMQ `q.retry`, where mixed per-message TTLs block each other at
60    /// the head) and the message returns with whatever priority it carries. A deferral
61    /// is held per delay, so equal delays drain strictly in order, and the envelope is
62    /// expected to carry the queue's top priority so it overtakes the backlog.
63    ///
64    /// Hold queue naming and lifetime are backend-specific.
65    async fn defer(&self, envelope: &Envelope, delay: Duration) -> Result<()>;
66
67    /// Start consuming `queue` with the given prefetch. The stream ends when the
68    /// backend is closed or the connection is lost.
69    async fn consume(&self, queue: &QueueConfig) -> Result<DeliveryStream>;
70
71    /// Graceful shutdown: stop all consumers, flush, close connections.
72    async fn close(&self) -> Result<()>;
73}
queuey_core/backend.rs

queuey_core/
backend.rs
