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
//! Accept orders over HTTP with axum and hand them to the same service's subscriber through a
//! transactional outbox.
//!
//! ```text
//! cargo run --example http_outbox --features macros,memory,json
//! ```
//!
//! Place an order and watch the subscriber pick it up:
//!
//! ```text
//! curl -X POST http://127.0.0.1:8080/orders \
//! -H 'content-type: application/json' -d '{"id":1,"item":"book"}'
//! ```
use std::collections::VecDeque;
use std::sync::Arc;
use std::time::Duration;
use axum::extract::State;
use axum::http::StatusCode;
use axum::routing::{get, post};
use axum::{Json, Router};
use ruststream::codec::JsonCodec;
use ruststream::memory::MemoryPublisher;
use ruststream::memory::prelude::*;
use ruststream::runtime::{HealthProbe, HealthState};
use serde::{Deserialize, Serialize};
use tokio::sync::Mutex;
// --8<-- [start:event]
/// The integration event the HTTP side hands to the messaging side.
#[derive(Debug, Clone, Outgoing, Serialize, Deserialize)]
#[outgoing(name = "orders.placed")]
struct OrderPlaced {
id: u64,
item: String,
}
// --8<-- [end:event]
// --8<-- [start:handler]
/// The same service consumes what its HTTP endpoints produce; any other service subscribed to
/// the broker would see the event too.
#[subscriber("orders.placed")]
async fn fulfil(order: &OrderPlaced) -> HandlerOutcome {
tracing::info!(order.id, order.item, "fulfilling");
HandlerOutcome::ack()
}
// --8<-- [end:handler]
// --8<-- [start:store]
/// The business records and the pending outbox rows, committed together. With a real database
/// both inserts run in one SQL transaction; the mutex over the pair plays that role here, so an
/// order is never stored without its event, and no event exists without its order.
#[derive(Default)]
struct Store {
orders: Vec<OrderPlaced>,
outbox: VecDeque<OrderPlaced>,
}
// --8<-- [end:store]
// --8<-- [start:endpoint]
/// The request path only writes to the store: recording the order and queueing its event is one
/// atomic step, and no broker I/O can fail or stall the HTTP response.
async fn place_order(
State(store): State<Arc<Mutex<Store>>>,
Json(order): Json<OrderPlaced>,
) -> &'static str {
let mut store = store.lock().await;
store.orders.push(order.clone());
store.outbox.push_back(order);
"accepted\n"
}
// --8<-- [end:endpoint]
// --8<-- [start:relay]
/// Drains the outbox into the broker. A row is removed only after its publish succeeds, so a
/// broker outage delays events instead of losing them; a crash between publish and removal
/// re-publishes the row on restart, which is the usual at-least-once contract of an outbox.
async fn relay_outbox(store: Arc<Mutex<Store>>, egress: MemoryPublisher) {
let mut tick = tokio::time::interval(Duration::from_millis(200));
loop {
tick.tick().await;
loop {
let Some(event) = store.lock().await.outbox.front().cloned() else {
break;
};
if egress
.message(&event)
.with_codec(JsonCodec)
.publish()
.await
.is_err()
{
// Keep the row and retry on the next tick.
break;
}
store.lock().await.outbox.pop_front();
}
}
}
// --8<-- [end:relay]
// --8<-- [start:healthz]
/// The liveness endpoint: the probe outlives `shutdown`, so this route reports the terminal
/// state for as long as requests still reach it. Here that window is the graceful drain: a
/// fail-fast also resolves the `stopping()` future below and stops the HTTP server. A host
/// that should keep serving after the messaging side dies would omit that select arm.
///
/// This service is unhealthy once a subscription stops consuming, even though the rest of it
/// still runs: the orders it no longer reads would wait unanswered.
async fn healthz(State(health): State<HealthProbe>) -> (StatusCode, String) {
let count = health.subscriptions();
match health.state() {
HealthState::Running if count.active == count.started => {
(StatusCode::OK, "running".to_owned())
}
HealthState::Running => (
StatusCode::SERVICE_UNAVAILABLE,
format!(
"{} of {} subscriptions consuming",
count.active, count.started
),
),
state => (StatusCode::SERVICE_UNAVAILABLE, format!("{state:?}")),
}
}
// --8<-- [end:healthz]
// --8<-- [start:wiring]
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
let broker = MemoryBroker::new().bindable();
// A token, not a publisher: minted before registration, paired after start.
let egress = broker.bind(Publish);
let app = RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
b.include(fulfil);
});
// The messaging side starts in the background; a startup failure (a broker refusing to
// connect, a subscription failing to open) surfaces here, before HTTP accepts traffic.
let running = app.start().await?;
// The handle existing witnesses that startup connected the broker, so the relay task gets a
// live publisher, never a "not connected" state.
let egress = running.publisher(egress).await?;
let store = Arc::new(Mutex::new(Store::default()));
tokio::spawn(relay_outbox(store.clone(), egress));
let router = Router::new()
.route("/orders", post(place_order))
.with_state(store)
// The health probe is `Clone` and outlives the app handle; /healthz flips to 503 the
// moment the messaging side fail-fasts, answering through the graceful drain below.
.route("/healthz", get(healthz).with_state(running.health()));
let listener = tokio::net::TcpListener::bind("127.0.0.1:8080").await?;
tracing::info!("orders API on http://127.0.0.1:8080/orders");
// The host owns the signals. HTTP stops on Ctrl+C, or when the messaging side tears itself
// down (fail-fast) so the process does not keep serving with a dead consumer; either way the
// messaging side then drains gracefully.
let stopping = running.stopping();
axum::serve(listener, router)
.with_graceful_shutdown(async move {
tokio::select! {
_ = tokio::signal::ctrl_c() => {}
() = stopping => {}
}
})
.await?;
running.shutdown().await?;
Ok(())
}
// --8<-- [end:wiring]