Producer

pub struct Producer<Q: QueueSet, B: Backend>;
impl<Q: QueueSet, B: Backend> Clone for Producer<Q, B>;

Cheap to clone — a clone shares the same backend — and Send + Sync, so one producer in application state serves every request handler.

Constructing

pub async fn new(backend: Arc<B>) -> Result<Self>;
pub fn new_undeclared(backend: Arc<B>) -> Self;
pub fn backend(&self) -> &Arc<B>;

new declares every queue in Q before returning, which is why it is async and fallible. new_undeclared declares nothing and cannot fail.

new_undeclared cannot delay or defer

On RabbitMQ, a hold queue needs the work queue to exist so it has something to dead-letter into. An undeclared producer’s enqueue_after and defer therefore fail with Error::UnknownQueue until something else declares the queue. Plain enqueue works, provided the queue really does exist.

let producer = Producer::<AppQueues, _>::new(backend.clone()).await?;

The turbofish is usually needed: Q cannot be inferred from the backend, only from the jobs you later enqueue.

Enqueueing

pub async fn enqueue<J: Job<Queue = Q>>(&self, job: &J) -> Result<uuid::Uuid>;
pub async fn enqueue_after<J: Job<Queue = Q>>(&self, job: &J, delay: Duration) -> Result<uuid::Uuid>;
pub async fn defer<J: Job<Queue = Q>>(&self, job: &J, delay: Duration) -> Result<uuid::Uuid>;

All three return the new job_id, which is stable for the life of the job across every retry and deferral. The J: Job<Queue = Q> bound is what rejects a job belonging to another queue set at compile time.

Call	Delay	Priority on arrival	Position
enqueue	none	0	back of the queue
enqueue_after	waits delay	0	back of the queue
defer	waits delay	the queue’s max_priority, or 0 if it has none	front of the queue

All three start the job at attempt = 1 and deferrals = 0. A producer-side defer does not count as a deferral — the producer scheduled the job, it did not defer it — but it still arrives at the elevated priority.

// Now.
let id = producer.enqueue(&SendEmail { to: "a@b.c".into(), body: "hi".into() }).await?;

// In five minutes, behind whatever is queued by then.
producer.enqueue_after(&SendEmail { .. }, Duration::from_secs(300)).await?;

// In thirty seconds, ahead of whatever is queued by then.
producer.defer(&CallApi { url }, Duration::from_secs(30)).await?;

enqueue_after is for “not before then”; defer is for “as soon as allowed”. See Deferral.

Delay limits

A delay is rounded up to the backend’s granularity, so a job is never released early. On RabbitMQ the ceiling is about 24.8 days; a longer delay is refused with an error rather than silently released early.

enqueue_after rounds with retry_granularity, defer with deferred_granularity. Both default to one second. See Broker topology.

Errors

Error	Cause
Error::Serde	The job failed to serialize
Error::UnknownQueue(q)	A delayed publish onto a queue this backend never declared
Error::Backend(_)	Transport failure, a nack, or an unroutable return
Error::ShutDown	The backend has been closed

A returned Ok means the broker confirmed the publish, so an enqueue that returns successfully is on the queue.

In a web application

Build the backend and producer once at startup, put the producer in shared state, and clone it per request:

let backend = Arc::new(RabbitMqBackend::connect(&url).await?);
let producer = Producer::<AppQueues, _>::new(backend.clone()).await?;

// ... store `producer` in application state; it is Clone + Send + Sync.

Do not construct a producer per request: new declares queues on every call, which is a needless round trip.

See also

Jobs, Deferral, and Producer on docs.rs.