poll_interval

poll_interval sets the pause between reads of an idle shard, one second by default; the service allows five reads a second per shard.

Each shard reader asks GetRecords for the next records of its shard. When a read returns records, the reader reads again at once; when a read comes back empty, it waits poll_interval first. So the setting bounds how long a new record waits on a quiet shard, and how often a quiet shard is read. The five reads a second are a limit of the shard, shared by every consumer that reads it. A read over that limit is throttled; the reader then waits one interval and reads on from where it stopped, and the handler sees a pause rather than an error. A shorter pause lowers the latency on a quiet stream and spends more of the shared limit. The same setting is a mount-site step through KinesisSubscriberExt. The AsyncAPI document reports it on the channel as pollIntervalMs.

Example

Quotes are repriced within 200 ms of their arrival on a quiet stream.

Macros
Manual

examples/kinesis_subscriptions_poll.rs
/// Reprices on every quote. A quiet shard is read again 200 ms after an empty read, so a new
/// quote waits at most that long; every read counts against the shard's five a second.
#[subscriber(KinesisStream::new("quotes").poll_interval(Duration::from_millis(200)))]
async fn reprice(quote: &Quote) -> HandlerOutcome {
    tracing::info!(quote.symbol, quote.price, "repriced");
    HandlerOutcome::ack()
}

fn service(broker: KinesisBroker) -> RustStream {
    RustStream::new(AppInfo::new("pricing", "0.1.0")).with_broker(broker, |b| {
        b.include(reprice);
    })
}

examples/manual/kinesis_subscriptions_poll.rs
/// Reprices on every quote. A quiet shard is read again 200 ms after an empty read, so a new
/// quote waits at most that long; every read counts against the shard's five a second.
struct Reprice;

impl Handle<Quote> for Reprice {
    fn handle(
        &self,
        quote: &Quote,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(quote.symbol, quote.price, "repriced");
        ready(Ok(()))
    }
}

fn service(broker: KinesisBroker) -> RustStream {
    RustStream::new(AppInfo::new("pricing", "0.1.0")).with_broker(broker, |b| {
        let quotes = KinesisStream::new("quotes").poll_interval(Duration::from_millis(200));
        b.include(subscriber(quotes, Reprice).build());
    })
}

Example​

Example