expiration

expiration gives one message a time to live; LapinPublish::expiration gives one to every message of a policy.

A message nobody consumed within its TTL leaves the queue. The broker drops it, or dead-letters it where the queue has a dead-letter exchange. RabbitMQ expires a message when it reaches the head of its queue. A policy that sets no TTL sends messages that do not expire. AMQP counts the TTL in whole milliseconds, and a shorter non-zero TTL becomes one millisecond. A delivery reports it under EXPIRATION_HEADER in milliseconds.

Example

A quote is worth a minute and a flash quote ten seconds. The broker drops a quote nobody consumed in time.

Macros

examples/lapin_publishing_expiration.rs

#[derive(OutSlot)]
#[publishes(Quote)]
struct Quotes;

/// A quote is worth a minute; a flash quote, ten seconds. The broker drops a quote nobody
/// consumed in time.
#[subscriber(RabbitQueue::new("prices"))]
async fn publish_quote(
    change: &PriceChange,
    Out(quotes): Out<impl Publisher<Options = LapinPublishOptions>, Quotes>,
) -> HandlerOutcome {
    let quote = Quote {
        sku: change.sku.clone(),
        cents: change.cents,
    };
    let sent = if change.flash {
        quotes
            .message(&quote)
            .expiration(Duration::from_secs(10))
            .publish()
            .await
    } else {
        quotes.message(&quote).publish().await
    };
    if sent.is_err() {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("pricing", "0.1.0")).with_broker(
        LapinBroker::new("amqp://localhost:5672"),
        |b| {
            b.include(publish_quote)
                .out(
                    Quotes,
                    Publish::default().expiration(Duration::from_mins(1)),
                )
                .build();
        },
    )
}

Manual

examples/manual/lapin_publishing_expiration.rs

#[derive(OutSlot)]
#[publishes(Quote)]
struct Quotes;

/// A quote is worth a minute; a flash quote, ten seconds. The broker drops a quote nobody
/// consumed in time.
struct PublishQuote;

impl<Egress> Handle<PriceChange, (), Outs<(Egress,)>> for PublishQuote
where
    Egress: OutEntry<Quotes, Wire: Publisher<Options = LapinPublishOptions>>,
{
    async fn handle(
        &self,
        change: &PriceChange,
        outs: &Outs<(Egress,)>,
        _ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        let quote = Quote {
            sku: change.sku.clone(),
            cents: change.cents,
        };
        let quotes = outs.get(Quotes);
        let sent = if change.flash {
            quotes
                .message(&quote)
                .expiration(Duration::from_secs(10))
                .publish()
                .await
        } else {
            quotes.message(&quote).publish().await
        };
        if sent.is_err() {
            return Err(HandlerOutcome::retry());
        }
        Ok(())
    }
}

fn app() -> impl App {
    RustStream::new(AppInfo::new("pricing", "0.1.0")).with_broker(
        LapinBroker::new("amqp://localhost:5672"),
        |b| {
            b.include(subscriber(RabbitQueue::new("prices"), PublishQuote).build())
                .out(
                    Quotes,
                    Publish::default().expiration(Duration::from_mins(1)),
                )
                .build();
        },
    )
}