Skip to main content

DynStack

A DynStack is a handler middleware chain assembled when the app starts, mounted on one registration as one layer.

A static layer chain is fixed when the code compiles. A DynStack holds a list chosen when the app is built: from configuration, a loop or a feature flag. Each element implements DynMiddleware. It gets the input, the context and a Next; next.run(input, ctx) runs the rest of the chain and the handler. It may return an outcome without that call. DynStack::new(list) freezes the list, and the first element runs outermost. An empty list leaves the handler's behaviour as it is. The stack is one ordinary Layer, fixed to one input type, so it wraps one registration: .layer(stack) right after an include on a router. The app and a router take only layers that wrap any input, which a stack fixed to one type is not. A registration's layer runs outside its decode step, so a stack over the broker's message type, MemoryMessage here, sees the raw delivery. Each element costs one boxed future and one dynamic call per delivery, and the stack one more of each for the handler it wraps; the static layers around it stay free. The middleware has no macro form: the same impl serves both forms, and only the mounting differs.

Example​

The middleware drops the deliveries of one tenant.

examples/dyn_stack.rs
/// Drops the deliveries of one tenant.
struct Block {
    tenant: String,
}

impl<I: Send + Sync> DynMiddleware<I> for Block {
    fn handle<'a>(
        &'a self,
        input: &'a I,
        ctx: &'a mut Context<'_>,
        next: Next<'a, I>,
    ) -> Pin<Box<dyn Future<Output = HandlerOutcome> + Send + 'a>> {
        Box::pin(async move {
            if ctx.headers().get("tenant") == Some(self.tenant.as_bytes()) {
                tracing::warn!(subscription = ctx.name(), tenant = %self.tenant, "blocked");
                return HandlerOutcome::drop();
            }
            next.run(input, ctx).await
        })
    }
}

The app builds one per tenant the configuration blocks.

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

/// The chain comes from configuration: one `Block` per tenant the operator listed.
fn app(blocked: &[&str]) -> impl App {
    let chain = blocked.iter().map(|tenant| {
        let block = Block {
            tenant: (*tenant).to_owned(),
        };
        Arc::new(block) as Arc<dyn DynMiddleware<MemoryMessage>>
    });
    let stack = DynStack::new(chain);
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
        // The stack wraps one registration, which sees the broker's own message.
        b.include_router(Router::new().include(accept).layer(stack));
    })
}