plugin_dme

Delay::plugin_dme hands a delayed copy to RabbitMQ's delayed-message-exchange plugin instead of a waiting queue.

The crate publishes the copy to an x-delayed-message exchange with the delay in its x-delay header, then acknowledges the original. The plugin holds each message for its own delay and then routes it to the origin queue. So a short delay never waits behind a long one, and one exchange serves every delay. The exchange is <queue>.delay; Delay::plugin_dme_named names another, which several queues may share. The released copy carries x-ruststream-retry-count raised by one, as with a waiting queue.

The backend sits behind the crate's plugin-dme feature, and the plugin has to be enabled on the server. With declare_topology(true) the service declares the exchange and binds the origin queue to it. On a broker that declares nothing, the operator provisions both, and the subscription checks at startup that the exchange exists.

Example

Each reminder waits as long as it asks for. The in-process transport releases the copy like a waiting queue; the plugin's own release order shows only against a server with the plugin.

Macros
Manual

examples/lapin_settlement_plugin_dme.rs
/// Each reminder waits as long as it asks, in the plugin's exchange `reminders.delay`. A short
/// wait is released on time even while a long one is still held.
#[subscriber(RabbitQueue::new("reminders").delay(Delay::plugin_dme()))]
async fn remind(reminder: &Reminder, ctx: &mut Context<'_>) -> HandlerOutcome {
    if ctx.headers().get_str(RETRY_COUNT_HEADER).is_none() {
        return HandlerOutcome::retry_after(Duration::from_secs(reminder.after_secs));
    }
    tracing::info!(reminder.id, "reminder sent");
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    // The service declares the delayed exchange; the plugin has to be enabled on the server.
    let broker = LapinBroker::new("amqp://localhost:5672").declare_topology(true);
    RustStream::new(AppInfo::new("reminders", "0.1.0")).with_broker(broker, |b| {
        b.include(remind);
    })
}

examples/manual/lapin_settlement_plugin_dme.rs
/// Each reminder waits as long as it asks, in the plugin's exchange `reminders.delay`. A short
/// wait is released on time even while a long one is still held.
struct Remind;

impl Handle<Reminder> for Remind {
    fn handle(
        &self,
        reminder: &Reminder,
        _outs: &(),
        ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        if ctx.headers().get_str(RETRY_COUNT_HEADER).is_none() {
            let wait = Duration::from_secs(reminder.after_secs);
            return ready(Err(HandlerOutcome::retry_after(wait)));
        }
        tracing::info!(reminder.id, "reminder sent");
        ready(Ok(()))
    }
}

fn app() -> impl App {
    // The service declares the delayed exchange; the plugin has to be enabled on the server.
    let broker = LapinBroker::new("amqp://localhost:5672").declare_topology(true);
    RustStream::new(AppInfo::new("reminders", "0.1.0")).with_broker(broker, |b| {
        let reminders = RabbitQueue::new("reminders").delay(Delay::plugin_dme());
        b.include(subscriber(reminders, Remind).build());
    })
}

Example​

Example