priority

priority sets the AMQP priority of one message; LapinPublish::priority sets it for every message of a policy.

A policy that sets no priority sends none. RabbitMQ orders deliveries by priority only on a queue declared with x-max-priority, and a higher priority counts as that maximum. On any other queue the priority travels to the consumer and orders nothing. A delivery reports it under PRIORITY_HEADER as decimal digits, and the handler reads it like any other header. A delayed redelivery keeps the priority its original delivery reported.

Example

A reading over 90 degrees raises an alert at priority 9, a warm one at the policy's 1. The alerts queue orders by priority, and its consumer pages someone for the urgent ones.

Macros

examples/lapin_publishing_priority.rs

/// A reading over 90 degrees raises an alert at priority 9; a warm one takes the policy's 1.
#[subscriber(RabbitQueue::new("readings"))]
async fn watch(
    reading: &Reading,
    Out(alerts): Out<impl Publisher<Options = LapinPublishOptions>, Alerts>,
) -> HandlerOutcome {
    if reading.celsius < 60 {
        return HandlerOutcome::ack();
    }
    let alert = Alert {
        sensor: reading.sensor.clone(),
    };
    let sent = if reading.celsius > 90 {
        alerts.message(&alert).priority(9).publish().await
    } else {
        alerts.message(&alert).publish().await
    };
    if sent.is_err() {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

/// The alerts queue orders by priority, and the consumer pages someone for the urgent ones.
#[subscriber(RabbitQueue::new("alerts").argument("x-max-priority", AMQPValue::LongLongInt(10)))]
async fn triage(
    alert: &Alert,
    ctx: &mut Context<'_>,
    Out(pager): Out<impl Publisher>,
) -> HandlerOutcome {
    let priority = ctx.headers().get_str(PRIORITY_HEADER).unwrap_or("0");
    if priority != "9" {
        tracing::info!(alert.sensor, priority, "alert logged");
        return HandlerOutcome::ack();
    }
    let page = Page {
        sensor: alert.sensor.clone(),
    };
    if pager.message(&page).publish().await.is_err() {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    let broker = LapinBroker::new("amqp://localhost:5672").declare_topology(true);
    RustStream::new(AppInfo::new("monitoring", "0.1.0")).with_broker(broker, |b| {
        b.include(watch)
            .out(Alerts, Publish::default().priority(1))
            .build();
        b.include(triage)
            .out(DefaultSlot, Publish::default())
            .build();
    })
}

Manual

examples/manual/lapin_publishing_priority.rs

/// A reading over 90 degrees raises an alert at priority 9; a warm one takes the policy's 1.
struct Watch;

impl<Egress> Handle<Reading, (), Outs<(Egress,)>> for Watch
where
    Egress: OutEntry<Alerts, Wire: Publisher<Options = LapinPublishOptions>>,
{
    async fn handle(
        &self,
        reading: &Reading,
        outs: &Outs<(Egress,)>,
        _ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        if reading.celsius < 60 {
            return Ok(());
        }
        let alert = Alert {
            sensor: reading.sensor.clone(),
        };
        let alerts = outs.get(Alerts);
        let sent = if reading.celsius > 90 {
            alerts.message(&alert).priority(9).publish().await
        } else {
            alerts.message(&alert).publish().await
        };
        if sent.is_err() {
            return Err(HandlerOutcome::retry());
        }
        Ok(())
    }
}

/// The alerts queue orders by priority, and the consumer pages someone for the urgent ones.
struct Triage;

impl<Egress> Handle<Alert, (), Outs<(Egress,)>> for Triage
where
    Egress: OutEntry<DefaultSlot, Wire: Publisher>,
{
    async fn handle(
        &self,
        alert: &Alert,
        outs: &Outs<(Egress,)>,
        ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        let priority = ctx.headers().get_str(PRIORITY_HEADER).unwrap_or("0");
        if priority != "9" {
            tracing::info!(alert.sensor, priority, "alert logged");
            return Ok(());
        }
        let page = Page {
            sensor: alert.sensor.clone(),
        };
        let pager = outs.get(DefaultSlot);
        if pager.message(&page).publish().await.is_err() {
            return Err(HandlerOutcome::retry());
        }
        Ok(())
    }
}

fn app() -> impl App {
    let broker = LapinBroker::new("amqp://localhost:5672").declare_topology(true);
    RustStream::new(AppInfo::new("monitoring", "0.1.0")).with_broker(broker, |b| {
        let alerts =
            RabbitQueue::new("alerts").argument("x-max-priority", AMQPValue::LongLongInt(10));
        b.include(subscriber(RabbitQueue::new("readings"), Watch).build())
            .out(Alerts, Publish::default().priority(1))
            .build();
        b.include(subscriber(alerts, Triage).build())
            .out(DefaultSlot, Publish::default())
            .build();
    })
}