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
178
179
//! The handler forms from the Subscribers guide: the basic contract, the context parameter, the
//! settings a mount site fills in, and the manual (macro-free) registration.
//!
//! ```text
//! cargo run --example subscribers --features macros,memory,json -- run
//! ```
use std::future::{Future, ready};
// The attribute and the value constructor share the name in different namespaces, so the one glob
// brings both into scope.
use ruststream::memory::prelude::*;
use serde::Deserialize;
// The manual registration at the bottom is documented by default, which is where the schema
// derive is owed; the attribute path asks for nothing.
#[derive(Debug, Deserialize, schemars::JsonSchema)]
struct Order {
id: u64,
}
// --8<-- [start:contract]
#[subscriber("orders")]
async fn handle(order: &Order) -> HandlerOutcome {
println!("got order {}", order.id);
HandlerOutcome::ack()
}
// --8<-- [end:contract]
// --8<-- [start:context]
#[subscriber("orders")]
async fn with_context(order: &Order, ctx: &mut Context<'_>) -> HandlerOutcome {
if let Some(id) = ctx.headers().correlation_id() {
println!("order {} correlates to {id}", order.id);
}
HandlerOutcome::ack()
}
// --8<-- [end:context]
// --8<-- [start:deferred_name]
/// The by-name source with its value left out: the mount site names the subscription.
#[subscriber]
async fn audit(order: &Order) -> HandlerOutcome {
println!("auditing order {}", order.id);
HandlerOutcome::ack()
}
// --8<-- [end:deferred_name]
// --8<-- [start:named_kind]
/// A named kind carrying only what it needs to exist; the value arrives at the mount site.
#[subscriber(MemorySource)]
async fn archive(order: &Order) -> HandlerOutcome {
println!("archiving order {}", order.id);
HandlerOutcome::ack()
}
// --8<-- [end:named_kind]
// --8<-- [start:batch]
/// Settles a whole batch of orders in one go: the slice parameter is what says so.
#[subscriber("orders")]
async fn settle(orders: &[Order]) -> HandlerOutcome {
println!("settling {} orders", orders.len());
HandlerOutcome::ack()
}
// --8<-- [end:batch]
// --8<-- [start:raw_batch]
/// The raw element type: the derive gives the newtype the delivery's bytes, and with them both
/// input spellings, so a batch of frames is `&[Frame<'_>]`.
#[derive(Deserialized)]
struct Frame<'a>(&'a [u8]);
/// A batch of payloads: the batch shape without the decode step.
#[subscriber("frames")]
async fn ingest(frames: &[Frame<'_>]) -> HandlerOutcome {
let bytes: usize = frames.iter().map(|frame| frame.0.len()).sum();
println!("ingesting {} frames ({bytes} bytes)", frames.len());
HandlerOutcome::ack()
}
// --8<-- [end:raw_batch]
// --8<-- [start:workers]
/// Up to 16 orders processed concurrently; global order is lost by design.
#[subscriber("orders", workers(16))]
async fn fan_out(order: &Order) -> HandlerOutcome {
println!("processing order {}", order.id);
HandlerOutcome::ack()
}
// --8<-- [end:workers]
// --8<-- [start:workers_by_key]
/// 16 lanes keyed by the message's partition key: per-key order is preserved.
#[subscriber("orders", workers(16, by_key))]
async fn per_customer(order: &Order) -> HandlerOutcome {
println!("processing order {}", order.id);
HandlerOutcome::ack()
}
// --8<-- [end:workers_by_key]
// --8<-- [start:batch_selective]
/// Retries only the entries that are not ready yet; the rest of the batch settles.
#[subscriber("orders")]
async fn reconcile(orders: &[Order]) -> Vec<HandlerOutcome> {
orders
.iter()
.map(|order| {
if order.id == 0 {
HandlerOutcome::retry()
} else {
HandlerOutcome::ack()
}
})
.collect()
}
// --8<-- [end:batch_selective]
/// How big a batch is, is the mount site's word, so both the name and the size land there.
#[subscriber]
async fn drain(orders: &[Order]) -> HandlerOutcome {
println!("draining {} orders", orders.len());
HandlerOutcome::ack()
}
/// A handler whose declarative settings are all left to the mount site.
#[subscriber]
async fn bill(order: &Order) -> HandlerOutcome {
println!("billing order {}", order.id);
HandlerOutcome::ack()
}
// The manual snippet below defines its handler type where the guide shows it, inside the builder
// closure, so the item follows the `include` statements above it on purpose.
#[allow(clippy::items_after_statements)]
#[ruststream::app]
fn app() -> RustStream {
let shard = 7;
RustStream::new(AppInfo::new("subscribers", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
b.include(handle);
b.include(with_context);
// --8<-- [start:name_mount]
b.include(audit.name(format!("audit-{shard}")));
// --8<-- [end:name_mount]
b.include(archive.name("archive"));
// --8<-- [start:batch_mount]
// The batch size is the one parameter a batch mount owes the broker: at most 64 orders per
// call, whatever the broker builds its batches out of.
b.include(settle.batch(nonzero!(64)));
// --8<-- [end:batch_mount]
b.include(ingest.batch(nonzero!(32)));
b.include(reconcile.batch(nonzero!(64)));
b.include(drain.name("orders").batch(nonzero!(128)));
// --8<-- [start:builder_settings]
b.include(
bill.name("orders")
.workers(nonzero!(4))
.on_failure(FailurePolicies::default().with_decode(FailurePolicy::Skip)),
);
// --8<-- [end:builder_settings]
b.include(fan_out);
b.include(per_customer);
// --8<-- [start:manual]
struct Inline;
impl Handle<Order> for Inline {
fn handle(
&self,
order: &Order,
_outs: &(),
_ctx: &mut Context<'_>,
) -> impl Future<Output = Result<(), HandlerOutcome>> {
println!("got order {}", order.id);
ready(Ok(()))
}
}
b.include(subscriber("orders", Inline).build());
// --8<-- [end:manual]
})
}