Message middleware · WebSocket

Add behavior before and after messages, and use an independent cleanup path on interruption.

Message middleware groups shared message policies and observation. before_message is entry, after_message is normal exit, and on_message_termination is the cleanup path for interrupted processing.

Middleware example#

middleware.rs contains a timer and the scope guard explained on the next page. The facade and Tokio dependencies from the first application without DI are sufficient.

use std::sync::Arc;
use lilyrs::websocket::{CleanupCancellation, ExecutionCancellation, Extensions,
    WebSocketErrorCode, WsAppBuilder, async_trait};
use lilyrs::websocket::guard::{GuardInitializationError, WebSocketGuardRejection, WsGuard};
use lilyrs::websocket::middleware::{MiddlewareDescriptor, MiddlewareErrorCode, MiddlewareKind,
    WsMessageDecision, WsMessageExchange, WsMessageMiddleware, WsMessageOutcome,
    WsMessageTerminationContext, WsMiddlewareError, WsMiddlewareInitError};

#[derive(Clone)]
pub struct StartedAt(pub tokio::time::Instant);
pub struct MessageTimer;

#[async_trait]
impl WsMessageMiddleware for MessageTimer {
    async fn new(_: Arc<Extensions>) -> Result<Self, WsMiddlewareInitError> { Ok(Self) }
    fn descriptor(&self) -> MiddlewareDescriptor {
        MiddlewareDescriptor::new("message_timer", MiddlewareKind::WebSocketMessage)
    }
    async fn before_message(&self, exchange: &mut WsMessageExchange,
        _: ExecutionCancellation) -> Result<WsMessageDecision, WsMiddlewareError>
    {
        exchange.insert_message_local(StartedAt(tokio::time::Instant::now()))
            .map_err(|_| WsMiddlewareError::internal(MiddlewareErrorCode::INTERNAL))?;
        Ok(WsMessageDecision::Continue)
    }
    async fn after_message(&self, exchange: &mut WsMessageExchange,
        _outcome: WsMessageOutcome, _: ExecutionCancellation)
        -> Result<WsMessageDecision, WsMiddlewareError>
    {
        if let Some(StartedAt(start)) = exchange.message_local::<StartedAt>() {
            let _elapsed = start.elapsed(); // Send to a bounded application metric.
        }
        Ok(WsMessageDecision::Continue)
    }
    async fn on_message_termination(&self, context: WsMessageTerminationContext<'_>,
        cancellation: CleanupCancellation) -> Result<(), WsMiddlewareError>
    {
        let _normal_exit = context.normal_exit();
        if cancellation.is_cancelled() { return Ok(()); }
        // Perform cleanup here without repeating the partial side effects of normal exit.
        Ok(())
    }
}

pub struct ChatWriteGuard;
#[async_trait]
impl WsGuard for ChatWriteGuard {
    async fn new(_: Arc<Extensions>) -> Result<Self, GuardInitializationError> { Ok(Self) }
    async fn can_activate(&self, exchange: &mut WsMessageExchange,
        _: ExecutionCancellation) -> Result<(), WebSocketGuardRejection>
    {
        if exchange.principal().is_some_and(|p| p.has_scope("chat:write")) {
            Ok(())
        } else {
            Err(WebSocketGuardRejection::error(
                WebSocketErrorCode::new("CHAT_FORBIDDEN").expect("sabit kod"),
                "Bu işlem için yetkiniz yok.",
            ).expect("sabit mesaj"))
        }
    }
}
pub fn with_policies(builder: WsAppBuilder) -> WsAppBuilder {
    builder.message_middleware::<MessageTimer>().guard::<ChatWriteGuard>()
}

Download the complete example.

The StartedAt message local is available within the same message. _elapsed is not sent to a metrics exporter yet; connect your application's bounded measurement here. The example's with_policies function installs both the timer and guard; the guard also requires identity. For only the timer, use .message_middleware::<MessageTimer>().

Registration levels#

Global registration is .message_middleware::<M>() on the builder; controller or action registration is #[message_middleware(M)]. Effective order is global → controller → action. Do not repeat the same concrete type on one route. A middleware plan contains at most 64 entries.

The constructor is fallible and runs once per app. The descriptor returns MiddlewareKind::WebSocketMessage. If a Scoped service is needed, resolve it from the active message scope with exchange.service::<T>().await instead of retaining it in the constructor.

Normal exit#

Middleware becomes entered when before_message returns Continue. A before callback ending in Reject, an error, panic, or timeout is not itself considered successfully entered. The entered prefix runs after_message in reverse order.

Continue preserves the current result. Reject can replace only a successful result. The first selected Close is not overwritten by a later Close chosen by outer middleware. Normal exits share the message deadline; after callbacks receive no fresh duration.

Termination cleanup#

Entered middleware whose normal exit has not started or was interrupted may receive on_message_termination. A normal exit that completed, returned an error, or panicked is not repeated on this path. context.normal_exit() reports evidence of the first poll; it does not undo partial side effects.

Even the callback's first poll is not guaranteed once the cleanup budget is exhausted. Callbacks must be idempotent and bounded and observe an independent CleanupCancellation signal. The response cannot be changed on this path. See deadlines and shutdown for details.