Skip to main content

Crate queuey

Crate queuey 

Source
Expand description

Type-safe job queues for Rust on RabbitMQ.

Queues are an enum with Queues; jobs are structs with Job. A job statically knows its queue, so Producer::enqueue and WorkerBuilder::handler are checked at compile time. A job belonging to another application’s queue set does not compile.

This crate is a facade: it re-exports queuey_core, the derive macros from queuey-macros, and (behind the default rabbitmq feature) queuey_rabbitmq. Depending on it alone is enough.

use queuey::prelude::*;

#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Queues)]
#[queues(prefix = "myapp")]
enum AppQueues {
    #[queue(prefetch = 10)]
    Emails,
    #[queue(retry(max_attempts = 3, backoff = "exponential", base = "1s", max = "2m"))]
    Images,
}

#[derive(Debug, Serialize, Deserialize, Job)]
#[job(queue = AppQueues::Emails, retry(max_attempts = 5))]
struct SendEmail { to: String, body: String }

struct EmailHandler;

#[async_trait]
impl JobHandler for EmailHandler {
    type Job = SendEmail;
    async fn handle(&self, job: SendEmail, ctx: JobContext) -> Result<(), JobError> {
        tracing::info!(to = %job.to, attempt = ctx.attempt, "sending");
        Ok(())
    }
}

#[tokio::main]
async fn main() -> queuey::Result<()> {
    let backend = Arc::new(RabbitMqBackend::connect("amqp://guest:guest@localhost:5672/%2f").await?);

    let producer = Producer::<AppQueues, _>::new(backend.clone()).await?;
    producer.enqueue(&SendEmail { to: "a@b.c".into(), body: "hi".into() }).await?;

    let worker = Worker::<AppQueues, _>::builder(backend)
        .handler(EmailHandler)
        // The worker leaves the shared backend open by default; this process has
        // nothing else to do with it.
        .close_backend_on_shutdown(true)
        .build()
        .await?;
    worker.run().await?;
    Ok(())
}

§Deferral (rate limits, Retry-After)

A job that cannot run yet, because the API answered 429 Too Many Requests with Retry-After: 30, is not a failure. The handler returns JobError::deferred / JobError::deferred_msg and the job is parked for exactly that long, then comes back ahead of the backlog that piled up meanwhile. It costs no attempt: ctx.attempt is unchanged and the retry policy is never consulted, so a job may defer itself indefinitely. JobContext::deferrals counts how often it happened, which is how a handler caps it (return JobError::Fatal once it has had enough).

Coming back first is a broker priority: #[queue(max_priority = n)] declares how many levels the queue has, and deferred jobs return at the top while ordinary work sits at 0. The default is DEFAULT_MAX_PRIORITY; 0 turns priorities off, and deferred jobs then come back FIFO. Producer::defer is the producer-side twin: same hold, same priority, without a first run.

use std::time::Duration;

use queuey::prelude::*;

#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Queues)]
#[queues(prefix = "docs")]
enum ApiQueues {
    #[queue(prefetch = 4, max_priority = 10)]
    Calls,
}

#[derive(Debug, Serialize, Deserialize, Job)]
#[job(queue = ApiQueues::Calls)]
struct CallApi { url: String }

struct CallHandler;

#[async_trait]
impl JobHandler for CallHandler {
    type Job = CallApi;
    async fn handle(&self, job: CallApi, ctx: JobContext) -> Result<(), JobError> {
        // Pretend the API answered `429` with `Retry-After: 30`.
        let retry_after: Option<Duration> = (ctx.deferrals == 0).then_some(Duration::from_secs(30));
        match retry_after {
            // Not a failure: wait exactly that long, keep the attempt, come back first.
            Some(delay) if ctx.deferrals < 5 => Err(JobError::deferred_msg(delay, "rate limited")),
            Some(_) => Err(JobError::fatal_msg("still rate limited after five deferrals")),
            None => Ok(()),
        }
    }
}

#[tokio::main]
async fn main() -> queuey::Result<()> {
    let backend = Arc::new(MemoryBackend::new());
    let producer = Producer::<ApiQueues, _>::new(backend.clone()).await?;
    let _worker = Worker::<ApiQueues, _>::builder(backend.clone())
        .handler(CallHandler)
        .build()
        .await?;

    // The same hold, without a first run: released in 30s, at the queue's top priority.
    producer
        .defer(&CallApi { url: "https://example.com".into() }, Duration::from_secs(30))
        .await?;

    assert_eq!(backend.deferred("docs.calls"), 1, "held, not yet deliverable");
    assert_eq!(backend.pending("docs.calls"), 0);
    assert_eq!(ApiQueues::Calls.config().max_priority, Some(DEFAULT_MAX_PRIORITY));
    Ok(())
}

§Where to look next

§How the derive macros find this crate

Generated code needs a path to queuey-core. The macros read the calling crate’s Cargo.toml and prefer a dependency on queuey, emitting ::queuey::__core, a hidden re-export of the core crate. Depending only on this facade therefore needs no crate = "..." attribute. A crate that depends on queuey-core directly gets ::queuey_core instead.

Re-exports§

pub use queuey_rabbitmq as rabbitmq;

Modules§

backend
Transport abstraction: Backend, Delivery and DeliveryStream.
envelope
The Envelope: the wire format every job travels in.
error
Error types: Error for infrastructure, JobError for handlers.
handler
JobHandler, the JobContext it receives and the FnHandler adapter.
job
The Job trait: a serializable payload bound to one queue.
memory
In-memory backend for tests and local development.
prelude
Everything a typical application needs, in one import.
producer
The Producer: type-safe publishing into a queue set.
queue
Queue declarations: QueueSet and QueueConfig.
retry
Retry configuration: RetryPolicy, Backoff and the backoff math.
worker
Worker runtime.

Structs§

Envelope
Wire format for a job message. Serialized as JSON in the message body.
FnHandler
Blanket adapter so plain async closures can be handlers: builder.handler(FnHandler::<SendEmail, _>::new(|job, ctx| async move { ... })).
JobContext
Metadata about the current execution, available to handlers.
MemoryBackend
An in-memory Backend. Cloning gives another handle onto the same state.
Producer
Type-safe publisher. Q pins the producer to one queue set so a job from a different application cannot be enqueued by accident.
QueueConfig
Static configuration of a single queue.
RabbitMqBackend
A Backend backed by a single RabbitMQ connection.
RabbitMqOptions
Configuration for RabbitMqBackend::with_options.
RetryPolicy
Retry configuration attached to a queue or a job.
Worker
Consumes one or more queues of Q and dispatches jobs to registered handlers.
WorkerBuilder
Configures a Worker. Created by Worker::builder.
WorkerHandle
Remote control for a running Worker. Cheap to clone and send across tasks.

Enums§

Backoff
How long to wait between attempts.
Error
Infrastructure-level errors (serialization, transport, configuration).
JobError
Error returned by a crate::JobHandler.
RetryDecision
Outcome of consulting a policy after a failure.

Constants§

DEFAULT_MAX_PRIORITY
Default for QueueConfig::max_priority: RabbitMQ recommends at most 10 levels.

Traits§

Backend
A transport. Implementations: MemoryBackend (this crate), RabbitMqBackend.
Delivery
A message received from the broker. Must be acked or nacked exactly once.
Job
A unit of work. Normally implemented via #[derive(Job)] from the macros crate; the hand-written equivalent is:
JobHandler
Processes jobs of one type. Register with crate::WorkerBuilder::handler.
QueueSet
A closed set of queues, normally an enum with #[derive(Queues)] from the macros crate; the hand-written equivalent is:

Type Aliases§

DeliveryStream
Stream of deliveries produced by Backend::consume.
Result
Result alias defaulting to this crate’s Error.

Derive Macros§

Job
Implement queuey_core::Job for a struct or enum.
Queues
Implement queuey_core::QueueSet for a fieldless enum.