LeaseKey

LeaseKey names one shard of one stream; a store of your own keys its rows by the key's stream:shard form.

Every LeaseStore call names its shard by a LeaseKey. The stream is the name the subscription's KinesisStream was given, and the shard is the id the service assigned. Shard ids repeat across streams, so a store keys by both, and one store serves every stream of a broker. LeaseKey displays as stream:shard; a stream name holds no colon, so the form is unambiguous, and a store that keeps one string key writes it. A store of your own implements the five methods of LeaseStore and is installed with KinesisBroker::lease_store. A store for one instance grants every lease; a store for several instances makes acquire, renew and checkpoint conditional on the owner, as DynamoLeaseStore does.

Example

A store for one instance keeps its checkpoints in a map keyed by the key's string form.

examples/kinesis_leases_key.rs
/// Checkpoints in a string-keyed map, for a single instance that owns every shard.
#[derive(Debug, Default)]
struct Checkpoints(Mutex<HashMap<String, String>>);

impl LeaseStore for Checkpoints {
    fn checkpoint<'a>(
        &'a self,
        key: &'a LeaseKey,
        _owner: &'a str,
        sequence: &'a str,
    ) -> BoxFuture<'a, Result<bool, LeaseError>> {
        Box::pin(async move {
            self.0
                .lock()
                .map_err(|err| err.to_string())?
                .insert(key.to_string(), sequence.to_owned());
            Ok(true)
        })
    }

    fn read<'a>(&'a self, key: &'a LeaseKey) -> BoxFuture<'a, Result<LeaseState, LeaseError>> {
        Box::pin(async move {
            let checkpoint = self
                .0
                .lock()
                .map_err(|err| err.to_string())?
                .get(&key.to_string())
                .cloned();
            let mut state = LeaseState::default();
            state.checkpoint = checkpoint;
            Ok(state)
        })
    }

    // One instance owns every shard, so a lease is always granted and never lapses.
    fn acquire<'a>(
        &'a self,
        _key: &'a LeaseKey,
        _owner: &'a str,
        _ttl: Duration,
    ) -> BoxFuture<'a, Result<bool, LeaseError>> {
        Box::pin(async { Ok(true) })
    }

    fn renew<'a>(
        &'a self,
        _key: &'a LeaseKey,
        _owner: &'a str,
        _ttl: Duration,
    ) -> BoxFuture<'a, Result<bool, LeaseError>> {
        Box::pin(async { Ok(true) })
    }

    fn release<'a>(
        &'a self,
        _key: &'a LeaseKey,
        _owner: &'a str,
    ) -> BoxFuture<'a, Result<(), LeaseError>> {
        Box::pin(async { Ok(()) })
    }
}

The service installs it on its broker.

Macros
Manual

examples/kinesis_leases_key.rs
fn service(broker: KinesisBroker, checkpoints: Arc<Checkpoints>) -> RustStream {
    RustStream::new(AppInfo::new("loyalty", "0.1.0")).with_broker(
        broker.lease_store(checkpoints),
        |b| {
            b.include(credit);
        },
    )
}

examples/manual/kinesis_leases_key.rs
fn service(broker: KinesisBroker, checkpoints: Arc<Checkpoints>) -> RustStream {
    RustStream::new(AppInfo::new("loyalty", "0.1.0")).with_broker(
        broker.lease_store(checkpoints),
        |b| {
            b.include(subscriber(KinesisStream::new("loyalty"), Credit).build());
        },
    )
}

Example​

Example