Skip to main content

Middleware and observability

The app wraps every subscriber in two layers: one logs each delivery and its outcome, one of the shop's own warns about slow handlers.

A Layer wraps a handler in another handler. It runs code before the body, after it, or instead of it, and passes the outcome on. A layer on the app wraps every subscriber the app mounts; a layer on a router wraps that router's subscribers only. RustStream emits its own events through tracing: the service starting, each subscription opening, the service stopping. TracingLayer adds an event for every delivery: delivery received at DEBUG, then its outcome, an ack at INFO, a retry or a drop at WARN. The logging feature prints them, as chapter 1 showed; any other tracing subscriber works in its place. Observability covers Prometheus metrics and OpenTelemetry export, which mount as layers the same way.

A layer of the shop's own​

SlowHandlers times each handler and warns when one runs longer than its threshold. The warning names the subscription, read off the delivery's context. A layer has no macro form: it is the same plain type in both tabs. BlanketLayer is what lets the app apply it to handlers whose types it hides.

observability.rs
/// Warns about every delivery whose handler ran longer than the threshold.
#[derive(Clone)]
pub(crate) struct SlowHandlers {
    threshold: Duration,
}

impl SlowHandlers {
    pub(crate) const fn over(threshold: Duration) -> Self {
        Self { threshold }
    }
}

/// The handler `SlowHandlers` makes of the one it wraps.
pub(crate) struct Timed<H> {
    inner: H,
    threshold: Duration,
}

impl<H> Layer<H> for SlowHandlers {
    type Handler = Timed<H>;

    fn layer(&self, inner: H) -> Timed<H> {
        Timed {
            inner,
            threshold: self.threshold,
        }
    }
}

impl<M, C, S, H> Handler<M, C, S> for Timed<H>
where
    M: Send + Sync,
    C: Send,
    S: Send + Sync,
    H: Handler<M, C, S>,
{
    async fn handle(&self, msg: &M, ctx: &mut Context<'_, C, S>) -> HandlerOutcome {
        let started = Instant::now();
        let outcome = self.inner.handle(msg, ctx).await;
        let took = started.elapsed();
        if took > self.threshold {
            tracing::warn!(subscription = ctx.name(), ?took, "slow handler");
        }
        outcome
    }
}

// The app wraps handlers whose types it hides, through this generic method.
impl BlanketLayer for SlowHandlers {
    fn apply<M, C, S, H>(&self, handler: H) -> impl Handler<M, C, S> + 'static
    where
        M: Send + Sync + 'static,
        C: Send + 'static,
        S: Send + Sync + 'static,
        H: Handler<M, C, S> + 'static,
    {
        self.layer(handler)
    }
}

The app​

service.rs assembles the shop. layer adds a layer to the app; the first one added is the outermost. Each module's router mounts with include_router, and the app's layers wrap its subscribers. on_startup opens the shared state of chapter 7, and info() is the description of chapter 9. The function takes the broker as a parameter, so the command-line entry point and the HTTP host of chapter 10 build the same shop.

service.rs
/// The whole shop on `broker`. The command-line entry point and the HTTP host both build it here.
pub(crate) fn shop(
    broker: impl BrokerRegistration<Broker = MemoryBroker>,
) -> impl App<State = Shop> {
    RustStream::new(info())
        .on_startup(async move |()| Shop::open().await)
        .layer(TracingLayer::default())
        .layer(SlowHandlers::over(Duration::from_millis(250)))
        .with_broker(broker, |b| {
            b.include_router(orders::routes());
            b.include_router(payments::routes());
            b.include_router(shipping::routes());
            b.include_router(notifications::routes());
        })
}

What the log shows​

An order placed with the request id req-1 leaves one line per step:

INFO guide_http::orders order placed request=req-1 order=1 total_cents=14400
INFO ruststream::dispatch handler ack
INFO ruststream::dispatch handler ack
INFO ruststream::dispatch handler ack
INFO guide_http::notifications mail sent server=smtp://mail.internal:25 to="ada@example.com" text="We have your order 1."
INFO ruststream::dispatch handler ack
INFO ruststream::dispatch handler ack
INFO ruststream::dispatch handler ack
INFO guide_http::notifications mail sent server=smtp://mail.internal:25 to="ada@example.com" text="Order 1 is on its way: TRK-1-2."
INFO ruststream::dispatch handler ack

RUST_LOG tunes what is printed, for example RUST_LOG=ruststream=debug,info.