Quickstart

Four things make up every queuey application:

1. a queue set, one enum deriving Queues;
2. one or more jobs, structs deriving Job, each bound to a variant;
3. a handler per job type;
4. a producer that enqueues and a worker that consumes.

The whole thing

RabbitMQ

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(())
}

Start a broker first:

docker run --rm -d -p 5672:5672 -p 15672:15672 rabbitmq:4-management

In memory

The same program with MemoryBackend in place of RabbitMqBackend. Nothing else changes, and no broker is involved.

use queuey::prelude::*;

#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Queues)]
#[queues(prefix = "myapp")]
enum AppQueues {
    #[queue(prefetch = 10)]
    Emails,
}

#[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(MemoryBackend::new());

    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.clone())
        .handler(EmailHandler)
        .build()
        .await?;
    let handle = worker.handle();
    let running = tokio::spawn(worker.run());

    // ... do work, then stop.
    handle.shutdown();
    running.await.unwrap()?;

    assert_eq!(backend.pending("myapp.emails"), 0);
    Ok(())
}

What each piece did

#[queues(prefix = "myapp")] made the broker queue names myapp.emails and myapp.images. Without a prefix they would be emails and images. The variant name is snake-cased unless #[queue(name = "...")] says otherwise.

#[queue(prefetch = 10)] told the consumer to hold at most ten unacknowledged messages at a time. The default is 16.

#[queue(retry(...))] gave every job on Images three attempts with exponential backoff. #[job(retry(max_attempts = 5))] overrode that for SendEmail specifically. Without either, a failure is dead-lettered on the first attempt.

Producer::<AppQueues, _>::new declared every queue in the set before returning. Producer::new_undeclared skips that if something else owns the topology.

Worker::<AppQueues, _>::builder collected handlers, then build() declared the queues it consumes and rejected any duplicate handler registration. run() consumes until the handle says to stop.

The worker does not close the backend

The backend is an Arc, normally shared with a producer and possibly other workers, so run() leaves it open. close_backend_on_shutdown(true) opts in when the process exists only to run the worker.

Runnable examples

Both live in the repository and are the source of the code on this page:

cargo run -p queuey --example memory_quickstart      # the whole library in one file, no broker
RUST_LOG=debug cargo run -p queuey --example memory_quickstart

docker run --rm -d -p 5672:5672 -p 15672:15672 rabbitmq:4-management
cargo run -p queuey --example rabbitmq_end_to_end    # the same tour against a real broker
AMQP_URL=amqp://user:pass@host:5672/%2f cargo run -p queuey --example rabbitmq_end_to_end

memory_quickstart enqueues six jobs: three that succeed, one that fails twice before succeeding, one that fails fatally, and one that a rate limit defers before it succeeds. It then waits for all of them to settle, shuts the worker down gracefully and prints what the backend saw. It is the fastest way to see every mechanism at once.

Next

• Queues and Jobs for the attributes in full.
• Handlers for what a handler can return and what each outcome means.
• Testing your jobs for asserting on all of it without a broker or wall-clock waits.