prefetch

prefetch caps unacknowledged deliveries per subscription on the broker or one descriptor; unset, the server applies no limit.

The server pushes deliveries to a consumer without waiting to be asked. The prefetch window (basic.qos) is how many deliveries it may have out to one subscription before the service settles some of them. So the window is the back-pressure between a queue and a service: a slow handler holds the rest of the queue on the server, where other instances can take it.

LapinBroker::prefetch sets the window for every subscription of the broker. RabbitQueue::prefetch and RabbitQuorumQueue::prefetch set it for one subscription and take precedence. The count is a NonZeroU16, because AMQP reads a window of 0 as no limit. A subscription with neither setting has no window, and the server pushes it the whole queue. A batch handler needs a window of at least its batch size, or its batches stay short.

The window is the server's: the in-process transport has none and hands a subscription everything its queue holds. A test of the window runs against a server with TestApp::start_live.

Example

Every subscription holds up to 64 deliveries, and the slow report queue holds one.

Macros
Manual

examples/lapin_subscriptions_prefetch.rs
use ruststream_lapin::prelude::*;

#[subscriber("orders")]
async fn place(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "order placed");
    HandlerOutcome::ack()
}

/// A report takes minutes to render, so this instance holds one at a time and leaves the rest
/// to the other instances.
#[subscriber(RabbitQueue::new("reports").prefetch(nonzero!(1)))]
async fn render(report: &Report) -> HandlerOutcome {
    tracing::info!(report.id, "report rendered");
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    // Every subscription holds up to 64 unacknowledged deliveries, unless its descriptor says
    // otherwise.
    let broker = LapinBroker::new("amqp://localhost:5672").prefetch(nonzero!(64));
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        b.include(place);
        b.include(render);
    })
}

examples/manual/lapin_subscriptions_prefetch.rs
use ruststream_lapin::prelude::*;

struct Place;

impl Handle<Order> for Place {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(order.id, "order placed");
        ready(Ok(()))
    }
}

/// A report takes minutes to render, so this instance holds one at a time and leaves the rest
/// to the other instances.
struct Render;

impl Handle<Report> for Render {
    fn handle(
        &self,
        report: &Report,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(report.id, "report rendered");
        ready(Ok(()))
    }
}

fn app() -> impl App {
    // Every subscription holds up to 64 unacknowledged deliveries, unless its descriptor says
    // otherwise.
    let broker = LapinBroker::new("amqp://localhost:5672").prefetch(nonzero!(64));
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        b.include(subscriber("orders", Place).build());
        let reports = RabbitQueue::new("reports").prefetch(nonzero!(1));
        b.include(subscriber(reports, Render).build());
    })
}

Example​

Example