Layer

A layer wraps handlers in code of your own: every handler of the app, every handler of a router, or one registration.

A Layer<H> turns a handler H into a wrapper of its own type, and that type implements Handler. The wrapper gets the input and the same &mut Context the handler gets, and calls the handler it wraps. It runs code before and after the call, and may return an outcome without making it. Three scopes take a layer. RustStream::layer, before with_broker, wraps every handler the app mounts, router handlers included. Router::layer, before the router's first include, wraps every handler of that router. .layer(..) right after an include on a router wraps that registration alone, outside its decode step. The app and a router hide their handlers' concrete types, so a layer for those scopes also implements BlanketLayer; a registration keeps its type, and Layer<H> is enough there. The app's layers wrap outside a router's. In the app and in a router, the first layer added runs outermost; on a registration, the last one added does. The chain is one type, so a layer adds nothing to the hot path beyond its own code. App and router layers wrap single-message handlers; batch handlers are not wrapped. A layer has no macro form: the same impl serves both forms, and only the mounting differs. A chain chosen when the app starts goes into a DynStack.

Example

The layer drops a delivery that names no tenant before the handler sees it.

examples/layer.rs

/// Drops a delivery that names no tenant before the handler sees it.
#[derive(Clone)]
struct RequireTenant;

/// The handler `RequireTenant` makes of the one it wraps.
struct Tenanted<H>(H);

impl<H> Layer<H> for RequireTenant {
    type Handler = Tenanted<H>;

    fn layer(&self, inner: H) -> Tenanted<H> {
        Tenanted(inner)
    }
}

impl<M, C, S, H> Handler<M, C, S> for Tenanted<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 {
        if ctx.headers().get("tenant").is_none() {
            tracing::warn!(subscription = ctx.name(), "no tenant header, dropped");
            return HandlerOutcome::drop();
        }
        self.0.handle(msg, ctx).await
    }
}

// The app and a router wrap handlers whose types they hide, through this generic method.
impl BlanketLayer for RequireTenant {
    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,
    {
        Tenanted(handler)
    }
}

Mounted on the app, it wraps every handler.

Macros

examples/layer.rs

#[subscriber("orders")]
async fn accept(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "accepted");
    HandlerOutcome::ack()
}

fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0"))
        // Every handler the app mounts below runs inside the layer.
        .layer(RequireTenant)
        .with_broker(MemoryBroker::new(), |b| {
            b.include(accept);
        })
}

Manual

examples/manual/layer.rs

fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0"))
        // Every handler the app mounts below runs inside the layer.
        .layer(RequireTenant)
        .with_broker(MemoryBroker::new(), |b| {
            b.include(subscriber("orders", Log).build());
        })
}

A router and a registration

On a fresh router the layer wraps every handler of the router. After an include it wraps that registration alone.

Macros

examples/layer.rs

fn app() -> impl App {
    RustStream::new(AppInfo::new("billing", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
        // Before the first `include`: every handler of this router.
        b.include_router(Router::new().layer(RequireTenant).include(refund));
        // After an `include`: that registration alone; `audit` runs unwrapped.
        b.include_router(
            Router::new()
                .include(invoice)
                .layer(RequireTenant)
                .include(audit),
        );
    })
}

Manual

examples/manual/layer.rs

fn app() -> impl App {
    RustStream::new(AppInfo::new("billing", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
        // Before the first `include`: every handler of this router.
        b.include_router(
            Router::new()
                .layer(RequireTenant)
                .include(subscriber("refunds", Log).build()),
        );
        // After an `include`: that registration alone; `audit` runs unwrapped.
        b.include_router(
            Router::new()
                .include(subscriber("invoices", Log).build())
                .layer(RequireTenant)
                .include(subscriber("audit", Log).build()),
        );
    })
}