create_if_missing

create_if_missing creates the stream with the shard count it names when it is missing, then waits up to a minute for it to become active.

The check runs once, when the subscription opens at startup. A stream the service already has is left as it is, whatever its shard count. A missing one is created with CreateStream, and the subscription opens once the stream is active. Two instances that start together may both try to create it; the one that loses the race uses the stream the other created. A stream that is not active within the minute, or a creation the service refuses, stops the service at startup with a KinesisError::Stream that names the stream. A shard count below one is refused before any request, also at startup. Without this setting a missing stream stops the service at startup, which catches a mistyped stream name before the service runs. The setting suits local stands and tests; a production stream is usually managed as infrastructure. The service's credentials need the kinesis:CreateStream permission for it. The in-process transport of a TestApp test has every stream a service names, so there the setting changes nothing. The AsyncAPI document reports the shard count on the channel as shardCount.

Example

On a stand without the stream parcels, the subscription creates it with two shards.

Macros
Manual

examples/kinesis_subscriptions_create.rs
/// Tracks each parcel. On a stand that lacks the stream, the subscription creates it with two
/// shards before it opens.
#[subscriber(KinesisStream::new("parcels").create_if_missing(2))]
async fn track(parcel: &Parcel) -> HandlerOutcome {
    tracing::info!(parcel.id, "parcel tracked");
    HandlerOutcome::ack()
}

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

examples/manual/kinesis_subscriptions_create.rs
/// Tracks each parcel. On a stand that lacks the stream, the subscription creates it with two
/// shards before it opens.
struct Track;

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

fn service(broker: KinesisBroker) -> RustStream {
    RustStream::new(AppInfo::new("parcels", "0.1.0")).with_broker(broker, |b| {
        let parcels = KinesisStream::new("parcels").create_if_missing(2);
        b.include(subscriber(parcels, Track).build());
    })
}

Example​

Example