ruststream-kinesis 0.7.1

Amazon Kinesis Data Streams broker implementation for the RustStream messaging framework.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158

<h1 align="center">ruststream-kinesis</h1>

<p align="center">
  <i>The Amazon Kinesis Data Streams broker for the <a href="https://github.com/powersemmi/ruststream">RustStream</a> messaging framework: a sharded, retained log with leases, checkpoints, and replay.</i>
</p>

<p align="center">
  <a href="https://github.com/powersemmi/ruststream-kinesis/actions/workflows/ci.yml"><img src="https://github.com/powersemmi/ruststream-kinesis/actions/workflows/ci.yml/badge.svg" alt="CI"></a>
  <a href="https://crates.io/crates/ruststream-kinesis"><img src="https://img.shields.io/crates/v/ruststream-kinesis.svg" alt="crates.io"></a>
  <a href="https://crates.io/crates/ruststream-kinesis"><img src="https://img.shields.io/crates/dr/ruststream-kinesis" alt="Recent downloads"></a>
  <a href="https://docs.rs/ruststream-kinesis"><img src="https://img.shields.io/docsrs/ruststream-kinesis" alt="docs.rs"></a>
  <img src="https://img.shields.io/badge/MSRV-1.95-blue.svg" alt="MSRV 1.95">
  <img src="https://img.shields.io/badge/license-Apache--2.0-blue.svg" alt="License">
  <a href="https://t.me/ruststream_community"><img src="https://img.shields.io/badge/-Telegram-blue?logo=telegram&label=News" alt="Telegram news channel"></a>
  <a href="https://t.me/ruststream_communuty_ru_chat"><img src="https://img.shields.io/badge/-Telegram-blue?logo=telegram&label=RU" alt="Telegram RU chat"></a>
</p>

<p align="center">
  <b><a href="https://powersemmi.github.io/ruststream-kinesis/latest/">Documentation</a></b>
</p>

---

`ruststream-kinesis` connects a RustStream service to Amazon Kinesis Data Streams over the
official [`aws-sdk-kinesis`](https://crates.io/crates/aws-sdk-kinesis). It also does the
coordination the vendor's consumer library does on other platforms: shard discovery across splits
and merges, shard leases, and per-shard checkpoints. Handlers, routing, codecs and middleware come
from the framework; this crate is the transport.

## Features

- **Checkpoints as acknowledgement:** a shard's checkpoint advances once every earlier record is
  handled, so delivery is at least once.
- **Shards handled by the crate:** one reader per owned shard, and children start after their
  parents, so per-key order survives resharding.
- **Leases per stream and shard:** a lease and its checkpoint belong to a stream and a shard
  together, so one lease store serves every stream a broker consumes. Instances share them
  through DynamoDB behind the `dynamodb-lease` feature.
- **Batches** sized at the mount site and read with one `GetRecords` call.
- **Start positions and repositioning:** the horizon, the tip, a timestamp or a captured record.
- **The partition key** as a per-message setting.
- **Deferred retries,** with retry caps and dead-letter streams.
- **AsyncAPI** with the stream, read pause and shards, behind the `asyncapi` feature.
- **Tests on the production app** (feature `testing`): `TestApp` runs it with `KinesisBroker`
  connected in process, with no AWS account, and `TestApp::start_live` runs the same test against
  LocalStack.

## Install

```toml
[dependencies]
ruststream = { version = "0.7", features = ["macros", "json"] }
ruststream-kinesis = "0.7"
serde = { version = "1", features = ["derive"] }

[dev-dependencies]
ruststream-kinesis = { version = "0.7", features = ["testing"] }
```

## Write a service

```rust
use ruststream_kinesis::prelude::*;
use serde::{Deserialize, Serialize};

#[derive(Debug, Deserialize, Outgoing, PartialEq, Serialize)]
struct Order {
    id: u64,
}

#[derive(Debug, Deserialize, Outgoing, PartialEq, Serialize)]
struct Receipt {
    order: u64,
}

#[subscriber(
    KinesisStream::new("orders"),
    reply("receipts"),
    start_at(KinesisPosition::horizon())
)]
async fn confirm(order: &Order) -> Receipt {
    Receipt { order: order.id }
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(KinesisBroker::new(), |b| {
        b.include(confirm)
            .out_reply(Publish::default())
            .partition_key("receipts-v1");
    })
}
```

`#[ruststream::app]` generates `main`, so the binary understands `run` and `asyncapi gen`.

## Upgrading a DynamoDB lease table

The table keeps one row per stream and shard, keyed `stream:shard` in the `lease_key` attribute. Versions before this one keyed a row by the shard id alone, so a table they wrote holds rows that name no stream. After the upgrade:

1. If the table served one stream, name it: `DynamoLeaseStore::new(&config, "orders-leases").legacy_stream("orders")`. A shard of that stream with no row of its own resumes from its bare row, and the first read copies that checkpoint under `orders:shardId-...`. Other streams ignore the bare rows.
2. Without that setting, a shard whose bare row holds a checkpoint is not read: the subscription reports a lease error naming the table, the shard and the stream, so no stream starts from another stream's progress or from the tip by accident.
3. Once the upgraded service has started every stream it consumes, each bare row that held progress has its copy. Delete the rows whose `lease_key` has no `:` and drop `legacy_stream`.

A table that served more than one stream before the upgrade holds, in each bare row, whichever stream checkpointed last, so neither stream can resume from it: delete the bare rows and open the streams with `start_at(..)` at the position they should read from. Stop the instances of the old version before starting the new one: the two versions lease different rows, so running both at once lets both read the same shard (duplicates, not losses).

## Test it

`TestApp::start` runs the service's own app with `KinesisBroker` connected in process, with no AWS
account. `TestApp::start_live(app())` runs the same test against LocalStack.

```rust
use ruststream::testing::TestApp;

let tb = TestApp::start(app()).await?;

tb.broker::<KinesisBroker>()
    .message(&Order { id: 42 })
    .to("orders")
    .publish()
    .await?;

tb.broker::<KinesisBroker>()
    .subscriber("orders")
    .assert_called_once()
    .with(&Order { id: 42 })
    .settled(HandlerOutcome::ack());

tb.broker::<KinesisBroker>()
    .published::<Receipt>("receipts")
    .assert_called_once()
    .with(&Receipt { order: 42 });
```

## Documentation

- [Getting started](https://powersemmi.github.io/ruststream-kinesis/latest/guides/getting-started/) - installation and a first service against a
  local Kinesis.
- [Guides](https://powersemmi.github.io/ruststream-kinesis/latest/) - consuming a stream, shards and ordering, checkpoints, retries, replay,
  several instances, testing and production.
- [Components](https://powersemmi.github.io/ruststream-kinesis/latest/components/) - one page per part of the crate: the broker, subscriptions,
  settlement, leases, positions, publishing and testing.
- [Architecture](https://powersemmi.github.io/ruststream-kinesis/latest/architecture/) - how the crate sits between the AWS SDK and the runtime.
- [Benchmarks](https://powersemmi.github.io/ruststream-kinesis/latest/architecture/benchmarks/) - what the crate costs over the raw client.
- [API reference](https://docs.rs/ruststream-kinesis) - every item on docs.rs.
- The framework: <https://powersemmi.github.io/ruststream/latest/>

## Minimum supported Rust version

The MSRV is **1.95**, edition 2024, the floor of the RustStream core.

## Contributing

See [CONTRIBUTING.md](./CONTRIBUTING.md).

## License

Licensed under the [Apache-2.0](./LICENSE) license.