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
//! Logging every message with the app-global `TracingLayer`, written without the `macros`
//! feature: named handler types grouped in a `Router`, and a hand-written `main`.
//!
//! ```text
//! RUST_LOG=ruststream=debug,info cargo run --example manual_logging_middleware \
//! --no-default-features --features memory,json,logging
//! ```
//!
//! `TracingLayer` wraps every handler and emits a `tracing` event on each delivery (DEBUG on
//! arrival, INFO on ack, WARN on nack), so the handlers stay free of logging calls. Middleware is
//! runtime API, so none of it changes with the macros off; what does change is who installs the
//! console subscriber that renders the events - the generated CLI calls `logging::init` on `run`,
//! and here `main` calls it.
//!
//! The app-global `.layer(..)` reaches router handlers: `include_router` wraps each with the app's
//! stack, which must be a `BlanketLayer` (every bundled layer is). `TracingLayer` here applies to
//! both `Confirm` and `Reject`, mounted through the `routes` group.
use std::error::Error;
use std::future::{Future, ready};
use ruststream::memory::prelude::*;
use ruststream::runtime::layers::TracingLayer;
use ruststream::runtime::{Identity, Stack};
use serde::Deserialize;
#[derive(Debug, Deserialize, schemars::JsonSchema)]
struct Order {
id: u64,
quantity: u32,
}
/// Accepts an order. The middleware logs the arrival and the resulting ack; no logging here.
struct Confirm;
impl Handle<Order> for Confirm {
fn handle(
&self,
order: &Order,
_outs: &(),
_ctx: &mut Context<'_>,
) -> impl Future<Output = Result<(), HandlerOutcome>> {
let _ = order.id;
ready(Ok(()))
}
}
/// Rejects empty orders by requeueing. The middleware logs the nack at WARN with `requeue=true`.
struct Reject;
impl Handle<Order> for Reject {
fn handle(
&self,
order: &Order,
_outs: &(),
_ctx: &mut Context<'_>,
) -> impl Future<Output = Result<(), HandlerOutcome>> {
if order.quantity == 0 {
return ready(Err(HandlerOutcome::retry()));
}
ready(Ok(()))
}
}
/// Builds the orders router. Broker-agnostic and middleware-agnostic: the app's global layer wraps
/// these handlers when the router is mounted.
// --8<-- [start:layered_router]
fn routes() -> impl RouterDef<MemoryBroker> {
Router::new()
.include(subscriber("orders", Confirm).build())
.include(subscriber("returns", Reject).build())
}
// --8<-- [end:layered_router]
fn app() -> RustStream<Stack<TracingLayer, Identity>> {
// The global layer is added before with_broker; include_router applies it to the router handlers.
RustStream::new(AppInfo::new("orders", "0.1.0"))
.layer(TracingLayer::default())
.with_broker(MemoryBroker::new(), |b| b.include_router(routes()))
}
#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
// What the generated CLI does on `run`: the layer only emits events, something has to render
// them.
ruststream::logging::init()?;
app().run().await?;
Ok(())
}