MemoryLeaseStore

MemoryLeaseStore is the default store: correct for one service instance, and empty again after a restart.

A broker that names no lease store creates one with MemoryLeaseStore::new. It keeps leases and checkpoints in the memory of the process, so one instance reads every shard of its streams and orders a split's children after their parents. Brokers of one process that share it through an Arc share its leases and checkpoints. A new process starts with an empty store. So a restarted service opens each shard at the tip, unless start_at names another position. The records published while the service was down are not read, and a held record does not come back. A service that must resume where it stopped, or runs in more than one instance, uses a store outside the process, such as DynamoLeaseStore.

Example

The audit service runs as one instance and reads from the tip after each start.

Macros
Manual

examples/kinesis_leases_memory.rs
/// Records each audit event. One instance of the service reads every shard.
#[subscriber(KinesisStream::new("audit-trail"))]
async fn record(event: &AuditEvent) -> HandlerOutcome {
    tracing::info!(event.id, "recorded");
    HandlerOutcome::ack()
}

fn service(broker: KinesisBroker) -> RustStream {
    // The store a broker uses when it names none, named here: its checkpoints live in this
    // process and go with it.
    let leases = Arc::new(MemoryLeaseStore::new());
    RustStream::new(AppInfo::new("audit", "0.1.0")).with_broker(broker.lease_store(leases), |b| {
        b.include(record);
    })
}

examples/manual/kinesis_leases_memory.rs
/// Records each audit event. One instance of the service reads every shard.
struct Record;

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

fn service(broker: KinesisBroker) -> RustStream {
    // The store a broker uses when it names none, named here: its checkpoints live in this
    // process and go with it.
    let leases = Arc::new(MemoryLeaseStore::new());
    RustStream::new(AppInfo::new("audit", "0.1.0")).with_broker(broker.lease_store(leases), |b| {
        b.include(subscriber(KinesisStream::new("audit-trail"), Record).build());
    })
}

Example​

Example