nest_rs_queue/
producer.rs

1//! Backend + producer seams.
2//!
3//! Every queue backend implements [`QueueBackend`] (the boot identity used by
4//! diagnostics) and exposes a [`JobProducer`] surface so any feature can
5//! enqueue without naming the backend. The first-class backend is `apalis-redis`
6//! (shipped as `nest-rs-queue`); third-party backends provide their own
7//! `*Module` that registers a `JobProducer` in the container the same way.
8
9use async_trait::async_trait;
10use serde::Serialize;
11
12use crate::error::QueueError;
13use crate::processor::Job;
14use crate::queue_name::QueueName;
15
16/// A queue backend, identified by name for boot diagnostics. The actual
17/// runtime work happens through [`JobProducer`] (enqueue) and
18/// [`JobConsumer`](crate::consumer::JobConsumer) (drain `ProcessMethod` and
19/// dispatch). A backend's module typically:
20///
21/// 1. seeds an `Arc<dyn JobProducer>` in the container (so any service can
22///    inject it generically), and
23/// 2. contributes a `Transport` whose `serve` runs the backend's
24///    [`JobConsumer`] driver.
25pub trait QueueBackend: Send + Sync + 'static {
26    /// Stable display name (e.g. `"apalis-redis"`, `"sqs"`, `"in-memory"`).
27    /// Logged at boot when the consumer attaches.
28    fn name(&self) -> &'static str;
29}
30
31/// Backend-agnostic producer: push a JSON-serialized job onto a named queue.
32/// Concrete backends implement this; typed pushes are a convenience built on
33/// top via [`JobProducerExt`].
34#[async_trait]
35pub trait JobProducer: Send + Sync + 'static {
36    /// Push a JSON-encoded job onto `queue`. The wire format is always JSON;
37    /// a backend may re-encode internally but must round-trip the payload. A
38    /// backend failure surfaces as [`QueueError::Backend`].
39    async fn push_json(&self, queue: &str, payload: serde_json::Value) -> Result<(), QueueError>;
40}
41
42/// Typed-push convenience over any [`JobProducer`]. Lives as an extension trait
43/// so the producer trait stays object-safe (`Arc<dyn JobProducer>`).
44#[async_trait]
45pub trait JobProducerExt: JobProducer {
46    /// Push a job onto a **typed** queue handle — the default enqueue path. The
47    /// queue name and the payload type are both taken from `Q`
48    /// ([`QueueName::NAME`] and [`QueueName::Job`]), so the compiler rejects an
49    /// enqueue onto the wrong queue or with the wrong payload before it ever
50    /// runs. Declare `Q` once at the feature port with the
51    /// [`queue`](crate::queue) macro; both the producer here and the consumer's
52    /// `#[process(queue = Q)]` name the same type.
53    ///
54    /// Fails with [`QueueError::Serialize`] if the job won't serialize, else
55    /// with whatever [`push_json`](JobProducer::push_json) returns.
56    ///
57    /// Passing a job of the wrong type is a compile error, not a runtime
58    /// surprise:
59    ///
60    /// ```compile_fail
61    /// use nest_rs_queue::{queue, JobProducer, JobProducerExt};
62    ///
63    /// #[queue(name = "transcode", job = String)]
64    /// struct TranscodeQueue;
65    ///
66    /// async fn demo<P: JobProducer>(producer: &P) {
67    ///     // `TranscodeQueue::Job` is `String`; a `u32` does not compile.
68    ///     producer.push_to::<TranscodeQueue>(42u32).await.unwrap();
69    /// }
70    /// ```
71    async fn push_to<Q: QueueName>(&self, job: Q::Job) -> Result<(), QueueError> {
72        let value = serde_json::to_value(&job)?;
73        self.push_json(Q::NAME, value).await
74    }
75
76    /// Push a job onto a queue named by a raw string — the dynamic-name escape
77    /// hatch. Prefer [`push_to`](JobProducerExt::push_to): a typed handle
78    /// compile-checks both the name and the payload type. Reach for this only
79    /// when the queue name genuinely isn't known until runtime. Fails with
80    /// [`QueueError::Serialize`] if the job won't serialize, else with whatever
81    /// [`push_json`](JobProducer::push_json) returns.
82    async fn push<J: Job + Serialize>(&self, queue: &str, job: J) -> Result<(), QueueError> {
83        let value = serde_json::to_value(&job)?;
84        self.push_json(queue, value).await
85    }
86}
87
88impl<T: JobProducer + ?Sized> JobProducerExt for T {}
nest_rs_queue/producer.rs

nest_rs_queue/
producer.rs
