ruststream 0.2.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

# Brokers

A broker connects RustStream to a message transport. The framework ships an in-memory broker for
development and tests; production brokers are separate crates you add as a dependency.

Handlers, routers, codecs, and middleware are broker-agnostic, so moving between brokers is a
one-line change at `with_broker`.

## In-memory (built-in)

The `memory` feature provides `MemoryBroker`, an in-process broadcast broker. It needs no external
service, which makes it ideal for examples, unit tests, and prototypes.

```toml
ruststream = { version = "0.2", features = ["macros", "memory"] }
```

```rust
use ruststream::memory::MemoryBroker;

let broker = MemoryBroker::new();
```

It does core routing only and does not emulate any real broker's delivery semantics. See
[Testing](../guides/testing.md).

## NATS

[`ruststream-nats`](https://github.com/powersemmi/ruststream-nats) is the NATS broker. It covers
Core NATS subjects and JetStream durable consumers.

```toml
ruststream = { version = "0.2", features = ["macros"] }
ruststream-nats = "0.2"
serde = { version = "1", features = ["derive"] }
```

`NatsBroker::new` is synchronous and does no I/O, so a NATS service is assembled with the same
`#[ruststream::app]` macro as any other broker. The runtime connects the broker once at startup,
before opening subscriptions.

### Core subscription

A `#[subscriber("subject")]` handler binds straight to a NATS subject:

```rust
use ruststream::codec::JsonCodec;
use ruststream::runtime::{AppInfo, HandlerResult, RustStream};
use ruststream::subscriber;
use ruststream_nats::NatsBroker;
use serde::Deserialize;

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

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

Wire it onto the broker; the `with_broker` / `include` part is identical to the in-memory broker.

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

### JetStream durable consumer

To consume from JetStream instead, override the handler's by-name source with `SubscribeOptions`,
naming the stream and a durable consumer so progress survives restarts. The handler's
`HandlerResult::Ack` acks back to JetStream.

```rust
use ruststream_nats::SubscribeOptions;

// inside with_broker(broker, |b| { ... }):
b.include_on(
    SubscribeOptions::new("orders.*")
        .jetstream("ORDERS")
        .durable("orders-worker"),
    handle,
    JsonCodec,
);
```

See the crate's documentation for connection options, authentication, and JetStream configuration.
For how a NATS broker implements the contract from the inside, read the
[worked example](../broker-authors/example-nats.md).

## Switching brokers

The same handlers and routers run on either broker, and both build synchronously, so only the
broker construction differs by one line inside `with_broker`.

=== "Memory"

    ```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()))
    }
    ```

=== "NATS"

    ```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())
            })
    }
    ```