Type-safe job queues for Rust on 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)
        .close_backend_on_shutdown(true)
        .build()
        .await?;
    worker.run().await?;
    Ok(())
}

use queuey::prelude::*;
// Same queue set, same job, same handler as the RabbitMQ tab.

#[tokio::test(start_paused = true)]
async fn flaky_email_is_retried_then_delivered() -> queuey::Result<()> {
    let backend = Arc::new(MemoryBackend::new());
    let producer = Producer::<AppQueues, _>::new(backend.clone()).await?;
    let worker = Worker::<AppQueues, _>::builder(backend.clone())
        .handler(EmailHandler)
        .build()
        .await?;
    let handle = worker.handle();
    let running = tokio::spawn(worker.run());

    producer.enqueue(&SendEmail { to: "a@b.c".into(), body: "hi".into() }).await?;
    tokio::time::sleep(Duration::from_secs(10)).await; // virtual

    handle.shutdown();
    running.await.unwrap()?;

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

if response.status() == 429 {
    // A handler that wants a cap enforces its own: nothing else will.
    if ctx.deferrals >= 5 {
        return Err(JobError::fatal_msg("still rate limited after five deferrals"));
    }
    let retry_after = parse_retry_after(&response).unwrap_or(Duration::from_secs(30));
    return Err(JobError::deferred_msg(retry_after, "rate limited"));
}

retry(
    max_attempts = 3,            // total attempts including the first; 1 means no retries
    backoff = "exponential",     // "none" | "fixed" | "exponential" (default)
    delay = "1s",                // fixed only, required
    base = "1s",                 // exponential only, default "1s", must be <= max
    factor = 2.0,                // exponential only, default 2.0
    max = "5m",                  // exponential only, default "5m"
    jitter = true,               // exponential only, default true
)