persistent

persistent sets the delivery mode of one message; LapinPublish::persistent sets it for every message of a policy.

A persistent message (delivery mode 2) on a durable queue survives a broker restart. A transient message (delivery mode 1) does not. The publish policies default to persistent. LapinRequest defaults to transient: a request nobody waits for after its timeout gains nothing from a restart.

Example

Live meter readings are transient, because the next tick replaces them. The reading a bill is made from is persistent.

Macros

examples/lapin_publishing_persistent.rs

#[derive(OutSlot)]
#[publishes(Reading)]
struct Readings;

/// Live readings are transient: a broker restart may lose them, and the next tick replaces them.
/// The reading a bill is made from is persistent.
#[subscriber(RabbitQueue::new("meters"))]
async fn record(
    tick: &MeterTick,
    Out(readings): Out<impl Publisher<Options = LapinPublishOptions>, Readings>,
) -> HandlerOutcome {
    let reading = Reading {
        meter: tick.meter.clone(),
        kwh: tick.kwh,
    };
    let sent = if tick.billing {
        readings.message(&reading).persistent(true).publish().await
    } else {
        readings.message(&reading).publish().await
    };
    if sent.is_err() {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("metering", "0.1.0")).with_broker(
        LapinBroker::new("amqp://localhost:5672"),
        |b| {
            b.include(record)
                .out(Readings, Publish::default().persistent(false))
                .build();
        },
    )
}

Manual

examples/manual/lapin_publishing_persistent.rs

#[derive(OutSlot)]
#[publishes(Reading)]
struct Readings;

/// Live readings are transient: a broker restart may lose them, and the next tick replaces them.
/// The reading a bill is made from is persistent.
struct Record;

impl<Egress> Handle<MeterTick, (), Outs<(Egress,)>> for Record
where
    Egress: OutEntry<Readings, Wire: Publisher<Options = LapinPublishOptions>>,
{
    async fn handle(
        &self,
        tick: &MeterTick,
        outs: &Outs<(Egress,)>,
        _ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        let reading = Reading {
            meter: tick.meter.clone(),
            kwh: tick.kwh,
        };
        let readings = outs.get(Readings);
        let sent = if tick.billing {
            readings.message(&reading).persistent(true).publish().await
        } else {
            readings.message(&reading).publish().await
        };
        if sent.is_err() {
            return Err(HandlerOutcome::retry());
        }
        Ok(())
    }
}

fn app() -> impl App {
    RustStream::new(AppInfo::new("metering", "0.1.0")).with_broker(
        LapinBroker::new("amqp://localhost:5672"),
        |b| {
            b.include(subscriber(RabbitQueue::new("meters"), Record).build())
                .out(Readings, Publish::default().persistent(false))
                .build();
        },
    )
}