跳到主要内容
英文版本为准

本页译自英文，由模型翻译。若与英文原文存在差异，以英文原文为准。

中间件与可观测性

应用用两层包裹每个订阅者：一层记录每次投递及其结果，另一层是商店自己的，对运行过慢的处理器发出警告。

Layer 把一个处理器包裹在另一个处理器里。 它在主体之前、之后或者代替主体运行代码，并把结果传递下去。 应用上的层包裹应用挂载的每个订阅者，路由器上的层只包裹该路由器的订阅者。 RustStream 通过 tracing 发出自己的事件：服务启动、每个订阅打开、服务停止。 TracingLayer 为每次投递添加事件：先在 DEBUG 级别记录 delivery received，然后记录它的结果。ack 在 INFO 级别，重试或丢弃在 WARN 级别。 logging feature 会打印这些事件，第 1 章已经展示过。任何其他 tracing 订阅者都可以代替它。 可观测性讲解 Prometheus 指标和 OpenTelemetry 导出，它们同样以层的形式挂载。

商店自己的层​

SlowHandlers 为每个处理器计时，运行时间超过阈值时发出警告。 警告中指明订阅，订阅名取自投递的上下文。 层没有宏形式：两个标签页中是同一个普通类型。 应用的处理器类型是隐藏的，BlanketLayer 让应用能把这个层应用到它们上。

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)
    }
}

应用​

service.rs 组装商店。 layer 给应用添加一个层，最先添加的层位于最外层。 每个模块的路由器用 include_router 挂载，应用的层包裹其中的订阅者。 on_startup 打开第 7 章的共享状态，info() 是第 9 章的描述。 函数以 Broker 作为参数，因此命令行入口和第 10 章的 HTTP 宿主构建的是同一个商店。

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());
        })
}

日志的内容​

用请求 id req-1 下单后，每一步留下一行日志：

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 调节打印哪些内容，例如 RUST_LOG=ruststream=debug,info。