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
//! The Prometheus example written without the `macros` feature: the reply body is a named type
//! whose `impl Handle` names the reply type, bound to its subject by the `subscriber` constructor
//! and given its destination and reply publisher on the same chain.
//!
//! ```text
//! cargo run --example manual_metrics_http --no-default-features --features memory,metrics,json
//! ```
//!
//! Publish an order, then read the metrics:
//!
//! ```text
//! curl -X POST http://127.0.0.1:8080/orders -d '{"id":1,"quantity":3}'
//! curl http://127.0.0.1:8080/metrics
//! ```
//!
//! Each published order is consumed (incrementing the consume counter) and replied to on
//! `confirmations` through the metrics publish layer (incrementing the publish counter).
use std::convert::Infallible;
use std::future::{Future, ready};
use std::sync::Arc;
use axum::Router;
use axum::body::Bytes;
use axum::extract::State;
use axum::routing::{get, post};
use ruststream::memory::MemoryPublisher;
use ruststream::memory::prelude::*;
use ruststream::metrics::Metrics;
use ruststream::runtime::{MessageWire, SerializedWire};
use ruststream::{CallerName, MessageHeaders, NoHeaders, OutgoingDestination};
use serde::{Deserialize, Serialize};
#[derive(Debug, Deserialize, schemars::JsonSchema)]
struct Order {
id: u64,
quantity: u32,
}
#[derive(Debug, Serialize, schemars::JsonSchema)]
struct Confirmation {
id: u64,
accepted: bool,
}
impl OutgoingDestination for Confirmation {
type Form = CallerName;
}
/// The HTTP request body on its way to the bus: bytes that arrived from outside, with no model of
/// their own. What `#[derive(Outgoing, Serialized)]` writes is these three impls - the bytes are
/// already the payload so no codec runs on them, the wire spelling routes the publish onto that
/// lane, and no destination is declared, so the call site names one. The axum buffer moves in
/// whole; nothing is copied on the way to the broker.
struct Ingest(Bytes);
impl Serialized for Ingest {
type Error = Infallible;
fn wire_bytes(&self, _buf: &mut BytesMut) -> Result<WireBytes<'_>, Infallible> {
Ok(WireBytes::Own(&self.0))
}
}
impl MessageWire for Ingest {
type Wire = SerializedWire;
}
impl OutgoingDestination for Ingest {
type Form = CallerName;
}
impl MessageHeaders for Ingest {
type Contract = NoHeaders;
}
/// The body `#[subscriber("orders", reply("confirmations"))]` generates. A reply-publishing
/// handler returns the value rather than publishing it, so the runtime encodes and sends it -
/// which is what puts it through the app's publish pipeline.
struct Confirm;
impl Handle<Order, Confirmation> for Confirm {
// `Err(result)` skips the publish and settles by the returned outcome; `Ok(reply)` publishes
// and acks.
fn handle(
&self,
order: &Order,
_outs: &(),
_ctx: &mut Context<'_>,
) -> impl Future<Output = Result<Confirmation, HandlerOutcome>> {
ready(Ok(Confirmation {
id: order.id,
accepted: order.quantity > 0,
}))
}
}
struct AppState {
metrics: Metrics,
ingest: MemoryPublisher,
}
async fn publish_order(State(state): State<Arc<AppState>>, body: Bytes) -> &'static str {
let _ = state
.ingest
.message(&Ingest(body))
.to("orders")
.publish()
.await;
"published\n"
}
async fn serve_metrics(State(state): State<Arc<AppState>>) -> String {
state.metrics.export().unwrap_or_default()
}
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
// --8<-- [start:wiring]
let metrics = Metrics::new()?;
let broker = MemoryBroker::new();
let ingest = broker.publisher();
let app = RustStream::new(AppInfo::new("orders", "0.1.0"))
.layer(metrics.consume_layer())
.publish_layer(metrics.publish_layer())
.with_broker(broker, |b| {
// The attribute reads the reply publisher off the broker's default publish policy;
// naming it on the chain is the same step written out.
b.include(
subscriber("orders", Confirm)
.reply()
.to("confirmations")
.build(),
)
.out_reply(Publish);
});
// --8<-- [end:wiring]
// The messaging side starts in the background and shares the metric collectors with the
// HTTP state; a startup failure surfaces here, before HTTP accepts traffic.
let running = app.start().await?;
let state = Arc::new(AppState { metrics, ingest });
let router = Router::new()
.route("/orders", post(publish_order))
.route("/metrics", get(serve_metrics))
.with_state(state);
let listener = tokio::net::TcpListener::bind("127.0.0.1:8080").await?;
println!("metrics on http://127.0.0.1:8080/metrics");
// The host owns the signals. HTTP stops on Ctrl+C, or when the messaging side tears itself
// down (fail-fast); 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(())
}