ruststream 0.7.5

Async messaging framework for Rust: broker-agnostic traits, router, codecs, and a conformance harness for broker authors.
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
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177

<h1 align="center">RustStream</h1>

<p align="center">
  <i>An async messaging framework for Rust: broker-agnostic traits, a router runtime, codecs, AsyncAPI generation, Prometheus and OpenTelemetry observability, and a conformance harness for broker authors.</i>
</p>

<p align="center">
  <a href="https://github.com/powersemmi/ruststream/actions/workflows/ci.yml"><img src="https://github.com/powersemmi/ruststream/actions/workflows/ci.yml/badge.svg" alt="CI"></a>
  <a href="https://coveralls.io/github/powersemmi/ruststream?branch=main"><img src="https://coveralls.io/repos/github/powersemmi/ruststream/badge.svg?branch=main" alt="Coverage"></a>
  <a href="https://crates.io/crates/ruststream"><img src="https://img.shields.io/crates/v/ruststream.svg" alt="crates.io"></a>
  <a href="https://crates.io/crates/ruststream"><img src="https://img.shields.io/crates/dr/ruststream" alt="Recent downloads"></a>
  <a href="https://docs.rs/ruststream"><img src="https://img.shields.io/docsrs/ruststream" 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">
  <img src="https://img.shields.io/badge/unsafe-none-success.svg" alt="100% safe Rust">
  <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>
  <a href="https://context7.com/powersemmi/ruststream"><img src="https://img.shields.io/badge/Context7-Ask_AI-6200ee" alt="Ask AI"></a>
  <a href="https://www.greptile.com/?utm_source=oss_badge&amp;utm_medium=readme&amp;utm_campaign=greptile_for_open_source"><img src="https://www.greptile.com/badge.svg" alt="Greptile: The War on Bugs"></a>
</p>

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

---

RustStream connects your service to a message broker through a small set of generic traits, then
gives you a router, middleware, codecs, and tooling on top. The core depends on no broker: each
broker is an independent crate held to one contract. The core is 100% safe Rust.

## Features

- **Broker-agnostic core.** Brokers are separate crates, checked by a conformance harness.
- **Misuse does not compile.** Double acks, out-of-order lifecycle calls and under-specified
  publishes are compile errors that name the fix.
- **Pluggable codecs:** JSON, MessagePack, and CBOR behind cargo features, or raw bytes with none.
- **Zero-boilerplate binaries.** `#[subscriber]` and `#[ruststream::app]` macros, and a CLI that
  scaffolds, runs and documents a service.
- **Observability:** AsyncAPI 3.1, Prometheus metrics, a health probe, and OpenTelemetry.
- **Tests without a broker.** The service's own app runs in process under a test harness.
- **Capability traits** for batches, transactions, request-reply, partitioning and repositioning;
  a broker implements only what it supports.

## Install

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

The CLI ships with the crate behind the `cli` feature:

```bash
cargo install ruststream --features cli
```

## Agent skills

The repository ships skills that teach a coding agent to write RustStream services, their tests
and broker crates the recommended way. In Claude Code:

```text
/plugin marketplace add powersemmi/ruststream
/plugin install ruststream@ruststream
```

Cursor and Codex read them from [`.agents/skills/`](.agents/skills); the
[Agent skills](https://powersemmi.github.io/ruststream/latest/guides/getting-started/agent-skills/)
page lists them.

## Write a service

```rust
use ruststream::memory::prelude::*;
use serde::Deserialize;

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

#[subscriber("orders")]
async fn handle(order: &Order) -> HandlerOutcome {
    println!("got order {}", order.id);
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> RustStream {
    RustStream::new(AppInfo::new("orders", "0.1.0"))
        .with_broker(MemoryBroker::new(), |b| { b.include(handle); })
}
```

`#[ruststream::app]` generates `main`, so there is no runtime boilerplate.

## Run it

```bash
ruststream run                 # start the service (or: cargo run -- run)
ruststream asyncapi gen        # print the AsyncAPI document
```

Scaffold a fresh project with `cargo generate --git https://github.com/powersemmi/ruststream
templates/memory --name my-service` (each broker crate ships its own template). See the
[quick start](https://powersemmi.github.io/ruststream/latest/guides/getting-started/quickstart/).

## Test it

`TestApp` runs the service's own app in process, with no external broker.

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

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

// Inject an order; the harness drives the handler to completion before returning.
tb.broker::<MemoryBroker>()
    .message(&Order { id: 42 })
    .to("orders")
    .publish()
    .await?;

// The handler ran once, decoded the order, and acked.
tb.broker::<MemoryBroker>()
    .subscriber("orders")
    .assert_called_once()
    .with(&Order { id: 42 })
    .settled(HandlerOutcome::ack());
```

Full compiling example: `examples/testing.rs`.

## Brokers

| Broker | Crate |
|---|---|
| NATS | [`ruststream-nats`](https://github.com/powersemmi/ruststream-nats) |
| Redis / Valkey | [`ruststream-fred`](https://github.com/powersemmi/ruststream-fred) |
| RabbitMQ (AMQP 0.9.1) | [`ruststream-lapin`](https://github.com/powersemmi/ruststream-lapin) |
| Apache Kafka | [`ruststream-rdkafka`](https://github.com/powersemmi/ruststream-rdkafka) |
| AMQP 1.0 | [`ruststream-amqp`](https://github.com/powersemmi/ruststream-amqp) |
| Google Cloud Pub/Sub | [`ruststream-gcp-pubsub`](https://github.com/powersemmi/ruststream-gcp-pubsub) |
| Amazon SQS / SNS | [`ruststream-sqs-sns`](https://github.com/powersemmi/ruststream-sqs-sns) |
| Apache Pulsar | [`ruststream-pulsar`](https://github.com/powersemmi/ruststream-pulsar) |
| MQTT 5 | [`ruststream-rumqttc`](https://github.com/powersemmi/ruststream-rumqttc) |
| ZeroMQ | [`ruststream-zeromq`](https://github.com/powersemmi/ruststream-zeromq) |
| Files and stdio | [`ruststream-sea-file`](https://github.com/powersemmi/ruststream-sea-file) |
| Amazon Kinesis | [`ruststream-kinesis`](https://github.com/powersemmi/ruststream-kinesis) |
| SQL databases (Postgres, MySQL / MariaDB, SQLite) | [`ruststream-sqlx`](https://github.com/powersemmi/ruststream-sqlx) |

What each broker supports is on the
[broker index](https://powersemmi.github.io/ruststream/latest/brokers/). To write a broker, start at
the [`Broker` trait](https://docs.rs/ruststream/latest/ruststream/trait.Broker.html).

## Documentation

- Site: <https://powersemmi.github.io/ruststream/latest>
- API reference and topic overviews: <https://docs.rs/ruststream>

## Minimum supported Rust version

The MSRV is **1.95**, edition 2024. The policy is a rolling one, as in tower: the MSRV rises only
to a Rust release at least six months old, and only in a minor release. A broker crate may require
a newer toolchain when its client does.

## Contributing

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

## License

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

<sub>Inspired by [FastStream](https://github.com/ag2ai/faststream).</sub>