Struct KinesisStream 

pub struct KinesisStream { /* private fields */ }

A subscription descriptor for one Kinesis stream.

Every shard resumes from its stored checkpoint, and a shard without one starts at the tip (a child whose parent was read to its end starts at its own beginning). To open somewhere else, wrap the descriptor in the framework’s start_at(..) clause with a KinesisPosition.

A string literal names a stream too: #[subscriber("orders")] builds this descriptor with its defaults. An invalid descriptor (an empty stream name, a zero shard count) fails the mount before any I/O, and a stream the service does not have fails it when the subscription opens, unless the descriptor creates it.

The subscription lists the stream’s shards and lists them again as splits and merges change the set, takes a lease per shard, and runs one reader per shard it holds. A child of a split or a merge starts only after its parents are fully consumed, which keeps per-key ordering across resharding.

A broker reads a stream through one subscription: a second subscription to the same stream on one broker is refused when it opens. A second KinesisBroker with a lease store of its own reads the stream again. A KPL-aggregated record is reported as KinesisError::AggregatedRecord.

Implements SubscriptionSource, so it can sit inline in the #[subscriber(..)] decorator, and IntoSource, so the manual path’s subscriber constructor names a stream with it too:

use std::error::Error;
use std::time::Duration;

use ruststream::prelude::*;
use ruststream_kinesis::{KinesisBroker, KinesisStream};
use schemars::JsonSchema;
use serde::Deserialize;

#[derive(Deserialize, JsonSchema)]
struct Order {
    id: u64,
}

struct Audit;

impl Handle<Order> for Audit {
    async fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        tracing::info!(order.id, "audited");
        Ok(())
    }
}

pub async fn run() -> Result<(), Box<dyn Error>> {
    let orders = KinesisStream::new("orders").poll_interval(Duration::from_millis(500));
    RustStream::new(AppInfo::new("orders", "0.1.0"))
        .with_broker(KinesisBroker::new(), |b| {
            b.include(subscriber(orders, Audit).build());
        })
        .run()
        .await?;
    Ok(())
}

Implementations

impl KinesisStream

pub fn new(stream: impl Into<String>) -> Self

Names the stream by its name, which is also the stream half of every LeaseKey the subscription’s shards are leased under.

A stream is named by its name; every call the subscription makes passes it as the stream name.

pub fn poll_interval(self, interval: Duration) -> Self

The pause between reads on an idle shard. Defaults to 1 second, the service’s own recommendation; lower values spend the 5-reads-per-second budget faster.

pub fn create_if_missing(self, shards: i32) -> Self

Creates the stream with shards provisioned shards on subscribe when it does not exist yet, then waits up to a minute for it to become active. Meant for local development and tests; production streams are usually managed as infrastructure.

Without it a stream the service does not have refuses the subscription when it opens, so the name is checked while a service is starting rather than after it has started.

pub fn stream(&self) -> &str

The stream name this descriptor resolves.

Trait Implementations

impl Clone for KinesisStream

fn clone(&self) -> Self

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Debug for KinesisStream

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl Eq for KinesisStream

impl IntoSource for KinesisStream

The manual path’s constructor takes a subject string or a source, and this is what makes the crate’s own descriptor the second: a service without the macros feature names its stream, and the settings that price a read, exactly as an attribute one does.

type Source = KinesisStream

The source the constructor stores.

fn into_source(self) -> Self

Builds the source.

impl PartialEq for KinesisStream

fn eq(&self, other: &Self) -> bool

Equality operator ==.

fn ne(&self, other: &Rhs) -> bool

Inequality operator !=.

impl RedeliveryAddressed<ConnectedKinesisBroker> for KinesisStream

fn redelivery_address( &self, _connected: &ConnectedKinesisBroker, ) -> impl Future<Output = Result<RedeliveryAddress, KinesisError>> + Send

The stream this descriptor names, which is where a deferred copy and a spent delivery of this registration are published.

The copy keeps the record’s partition key and lands on that key’s shard, at the tip rather than in the place the original held, so a deferred record loses its position in the stream’s order. Where that order matters, declare no cap and let HandlerOutcome::retry() hold the shard: the checkpoint stops at the record, and the shard is read again from it.

max_attempts(n) counts the first delivery, and under it HandlerOutcome::retry() becomes a published copy as well, because only a copy carries the attempt count forward. dead_letter(name) is the stream a spent record leaves for, payload and headers as it arrived; a cap declared without one checkpoints past the spent record, as drop() does.

Examples

use std::time::Duration;

use ruststream_kinesis::prelude::*;
use serde::Deserialize;

#[derive(Deserialize)]
struct Order {
    id: u64,
}

/// Asks for a pause when the order cannot be settled yet, and counts nothing: how many pauses
/// it gets is the mount site's to say.
#[subscriber(KinesisStream::new("orders"))]
async fn reconcile(order: &Order) -> HandlerOutcome {
    if order.id.is_multiple_of(2) {
        HandlerOutcome::ack()
    } else {
        HandlerOutcome::retry_after(Duration::from_secs(30))
    }
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(KinesisBroker::new(), |b| {
        // Four deliveries per order, then the record leaves for a stream an operator reads.
        b.include(reconcile)
            .max_attempts(nonzero!(4))
            .dead_letter("orders.unsettled");
    })
}

impl StructuralPartialEq for KinesisStream

impl SubscriptionSource<ConnectedKinesisBroker> for KinesisStream

type Copies = AddressedCopies

Kinesis moves nothing on its own: it holds no redelivery timer, no delivery counter and no dead-letter mechanism, so the framework publishes every copy. A stream is what a subscription reads and what a publish writes to, so this descriptor knows where a copy goes.

type Subscriber = KinesisSubscriber

The subscriber type this source opens.

fn name(&self) -> &str

The name (subject / channel) this subscription binds to.

async fn subscribe( self, connected: &ConnectedKinesisBroker, ) -> Result<KinesisSubscriber, KinesisError>

Opens the subscription against the connected broker. Called once at startup.

fn channel_bindings(&self) -> Bindings

What this subscription adds to its channel in the generated AsyncAPI document.

fn declare_retry(self, declaration: &RetryDeclaration) -> Self

where Self: Sized,

Takes the registration’s retry declaration into the descriptor, before the subscription opens.

fn declare_retry_on( &self, connected: &C, declaration: &RetryDeclaration, ) -> Result<(), DeclareRetryError>

Carries the registration’s declaration to the broker, at the same point and before the subscription opens.

fn operation_bindings(&self) -> Bindings

What this subscription adds to its receive operation in the document.

fn message_bindings(&self) -> Bindings

What this subscription adds to the messages that arrive on it.