ruststream 0.7.3

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

import Tabs from '@theme/Tabs';
import TabItem from '@theme/TabItem';

# Brokers

Handlers, routers, codecs, and middleware are broker-agnostic. You can move a service to another
broker by changing one line at `with_broker`.

The framework ships a full in-memory broker for queues that stay inside a single application.
Brokers backed by an external service are separate crates you add as a dependency.

Each broker crate has its own documentation site, linked in the Docs column and from the
**Brokers** menu.

| Broker | Crate | Transport | Docs |
|---|---|---|---|
| [Memory](memory.md) | `ruststream` (feature `memory`) | in-process queue, no external service | this site |
| NATS | [`ruststream-nats`](https://github.com/powersemmi/ruststream-nats) | Core NATS and JetStream | [powersemmi.github.io/ruststream-nats](https://powersemmi.github.io/ruststream-nats/) |
| Redis | [`ruststream-fred`](https://github.com/powersemmi/ruststream-fred) | Redis Streams (standalone, cluster, sentinel) | [powersemmi.github.io/ruststream-fred](https://powersemmi.github.io/ruststream-fred/) |
| RabbitMQ | [`ruststream-lapin`](https://github.com/powersemmi/ruststream-lapin) | AMQP 0.9.1 (queues, exchanges, publisher confirms, direct reply-to) | [powersemmi.github.io/ruststream-lapin](https://powersemmi.github.io/ruststream-lapin/) |
| Kafka | [`ruststream-rdkafka`](https://github.com/powersemmi/ruststream-rdkafka) | Apache Kafka (consumer groups, tracked commits, transactions, exactly-once pipelines) | [powersemmi.github.io/ruststream-rdkafka](https://powersemmi.github.io/ruststream-rdkafka/) |
| AMQP 1.0 | [`ruststream-amqp`](https://github.com/powersemmi/ruststream-amqp) | ActiveMQ Artemis, RabbitMQ 4.x, Azure Service Bus, and the wider AMQP 1.0 family (request/reply, transactions) | [powersemmi.github.io/ruststream-amqp](https://powersemmi.github.io/ruststream-amqp/) |
| Google Cloud Pub/Sub | [`ruststream-gcp-pubsub`](https://github.com/powersemmi/ruststream-gcp-pubsub) | Pub/Sub over the official client (ordering keys, exactly-once acknowledgement, dead-letter policies) | [powersemmi.github.io/ruststream-gcp-pubsub](https://powersemmi.github.io/ruststream-gcp-pubsub/) |
| AWS SQS / SNS | [`ruststream-sqs-sns`](https://github.com/powersemmi/ruststream-sqs-sns) | SQS queues with SNS fan-out (FIFO groups, visibility management, native deferred retry) | [powersemmi.github.io/ruststream-sqs-sns](https://powersemmi.github.io/ruststream-sqs-sns/) |
| Apache Pulsar | [`ruststream-pulsar`](https://github.com/powersemmi/ruststream-pulsar) | Pulsar topics and patterns (subscription modes, dead-letter policies, repositioning) | [powersemmi.github.io/ruststream-pulsar](https://powersemmi.github.io/ruststream-pulsar/) |
| MQTT 5 | [`ruststream-rumqttc`](https://github.com/powersemmi/ruststream-rumqttc) | MQTT v5 (QoS levels, shared groups, retained messages) | [powersemmi.github.io/ruststream-rumqttc](https://powersemmi.github.io/ruststream-rumqttc/) |
| ZeroMQ | [`ruststream-zeromq`](https://github.com/powersemmi/ruststream-zeromq) | Brokerless PUSH/PULL, PUB/SUB, and DEALER/ROUTER request/reply over TCP and IPC | [powersemmi.github.io/ruststream-zeromq](https://powersemmi.github.io/ruststream-zeromq/) |
| Stream files / stdio | [`ruststream-sea-file`](https://github.com/powersemmi/ruststream-sea-file) | Persistent replayable stream files and shell pipelines; zero infrastructure, full repositioning | [powersemmi.github.io/ruststream-sea-file](https://powersemmi.github.io/ruststream-sea-file/) |
| AWS Kinesis | [`ruststream-kinesis`](https://github.com/powersemmi/ruststream-kinesis) | Kinesis data streams (shard leasing, checkpointing, repositioning) | [powersemmi.github.io/ruststream-kinesis](https://powersemmi.github.io/ruststream-kinesis/) |
| SQL databases | [`ruststream-sqlx`](https://github.com/powersemmi/ruststream-sqlx) | Task queues in the service's own Postgres, MySQL/MariaDB and SQLite tables, and a transactional outbox over any broker; statements checked at startup or at compile time | [powersemmi.github.io/ruststream-sqlx](https://powersemmi.github.io/ruststream-sqlx/) |

To implement a broker for another transport, see [Broker authors](../broker-authors/index.md).

## Switching brokers

Every broker is constructed synchronously. The runtime connects it when the application starts.
The examples below differ only in the line that constructs the broker.

<Tabs groupId="broker">
<TabItem value="memory" label="Memory">

{/* inline-rust: side-by-side broker-switch comparison; the NATS half depends on the external ruststream-nats crate and has no in-repo compiled home, so both halves stay inline to read in parallel */}
```rust
use ruststream::memory::MemoryBroker;
use ruststream::runtime::{AppInfo, RustStream};

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

</TabItem>
<TabItem value="nats" label="NATS">

{/* inline-rust: NATS half of the broker-switch comparison; depends on the external ruststream-nats crate, no in-repo compiled home */}
```rust
use ruststream::runtime::{AppInfo, RustStream};
use ruststream_nats::NatsBroker;

#[ruststream::app]
fn app() -> RustStream {
    RustStream::new(AppInfo::new("orders", "0.1.0"))
        .with_broker(NatsBroker::new("nats://localhost:4222"), |b| {
            b.include_router(routes::orders())
        })
}
```

</TabItem>
<TabItem value="redis" label="Redis">

{/* inline-rust: Redis half of the broker-switch comparison; depends on the external ruststream-fred crate, no in-repo compiled home */}
```rust
use ruststream::runtime::{AppInfo, RustStream};
use ruststream_fred::RedisBroker;

#[ruststream::app]
fn app() -> RustStream {
    RustStream::new(AppInfo::new("orders", "0.1.0"))
        .with_broker(RedisBroker::standalone("redis://localhost:6379"), |b| {
            b.include_router(routes::orders())
        })
}
```

</TabItem>
<TabItem value="rabbitmq" label="RabbitMQ">

{/* inline-rust: RabbitMQ half of the broker-switch comparison; depends on the external ruststream-lapin crate, no in-repo compiled home */}
```rust
use ruststream::runtime::{AppInfo, RustStream};
use ruststream_lapin::LapinBroker;

#[ruststream::app]
fn app() -> RustStream {
    RustStream::new(AppInfo::new("orders", "0.1.0"))
        .with_broker(LapinBroker::new("amqp://localhost:5672"), |b| {
            b.include_router(routes::orders())
        })
}
```

</TabItem>
<TabItem value="kafka" label="Kafka">

{/* inline-rust: Kafka half of the broker-switch comparison; depends on the external ruststream-rdkafka crate, no in-repo compiled home */}
```rust
use ruststream::runtime::{AppInfo, RustStream};
use ruststream_rdkafka::KafkaBroker;

#[ruststream::app]
fn app() -> RustStream {
    RustStream::new(AppInfo::new("orders", "0.1.0"))
        .with_broker(
            KafkaBroker::new(["localhost:9092"]).default_group("orders"),
            |b| {
                b.include_router(routes::orders())
            },
        )
}
```

</TabItem>
</Tabs>

Each broker crate documents its own connection options. When a subscription needs broker-specific
options (consumer groups, durable names), you can write that broker's descriptor in the
`#[subscriber(..)]` attribute; see
[broker-specific descriptors](https://docs.rs/ruststream/latest/ruststream/runtime/index.html#the-subscription-source).