rainier-queue

Deferred work: the [Job] contract, the [Queue] port and its drivers, and the [Worker] that runs them.

use rainier_queue::{Job, JobContext, JobRegistry, MemoryQueue, Queue, QueueManager};
use serde::{Deserialize, Serialize};
use std::sync::Arc;

#[derive(Serialize, Deserialize)]
struct SendWelcomeEmail { user_id: u64 }

#[async_trait::async_trait]
impl Job for SendWelcomeEmail {
    const NAME: &'static str = "mail.welcome";
    const QUEUE: &'static str = "mail";

    async fn handle(&self, _context: &JobContext) -> rainier_support::Result<()> {
        Ok(())
    }
}

# #[tokio::main] async fn main() -> rainier_support::Result<()> {
let registry = Arc::new(JobRegistry::new().with::<SendWelcomeEmail>());
let queue = Arc::new(MemoryQueue::new());
let dispatcher = QueueManager::new(queue.clone(), registry);

dispatcher.dispatch(SendWelcomeEmail { user_id: 1 }).await?;
assert_eq!(queue.size("mail").await?, 1);
# Ok(()) }

A job crosses a process boundary

Everything about the design follows from that. A job is written by a web request and read, later, by a worker that may be a different process on a different machine — so a job is a serialisable payload plus a stable name, not a closure, and the worker needs a [JobRegistry] to turn the name back into code. Dependencies cannot be captured either, so a running job resolves them from the container through its [JobContext].

Drivers

Testing

[QueueManager::fake] records dispatches instead of performing them:

# use rainier_queue::{Job, JobContext, QueueManager};
# use serde::{Deserialize, Serialize};
# #[derive(Serialize, Deserialize)] struct SendInvoice;
# #[async_trait::async_trait] impl Job for SendInvoice {
#     const NAME: &'static str = "billing.invoice";
#     async fn handle(&self, _: &JobContext) -> rainier_support::Result<()> { Ok(()) }
# }
# #[tokio::main] async fn main() -> rainier_support::Result<()> {
let queue = QueueManager::fake();
queue.dispatch(SendInvoice).await?;
queue.assert_pushed_times::<SendInvoice>(1);
# Ok(()) }