Typed extractors · WebSocket

Receive message metadata, services, and a single payload as owned action parameters.

Extractors make an action's requirements visible in its signature. After finding the route, the framework runs all parts extractors, then the single payload consumer. The action focuses on its business rule.

Choose the right extractor#

Need	Extractor
Connection targets and identity view	`WebSocketContext`, `Arc<WebSocketContext>`
Message ID, deadline, and metadata	`WebSocketMessageContext`
Route information	`ConnectionId`, `Namespace`, `EventName`
Envelope metadata	`MessageHeaders`
Verified identity	`Principal`, `Option<Principal>`
Typed local data	`ConnectionLocal<T>`, `MessageLocal<T>`
A service in the active scope	`Service<T>`, `Service<dyn Trait>`
JSON DTO	`Payload<T>`
Typed text or binary	`TextPayload`, `BinaryPayload`
Decoded payload enum	`RawPayload`
Original application frame	`RawEnvelope`
Operation control	`ExecutionCancellation`, `MessageDeadline`

RawPayload is not the complete wire bytes. RawEnvelope takes the single consumption right without triggering payload decoding. Combining two of these in one signature is not a way to obtain two views; it is rejected as two payload consumers.

Complete example#

The example includes a custom parts extractor, JSON validation, a binary ACK, Close, and a bounded action. It uses the first application's dependencies. AuthorizedSubject only checks that a previously verified Principal exists; without identity middleware, the validated action is not admitted.

use std::{future::Future, sync::Arc};
use lilyrs::websocket::{Ack, BinaryPayload, CloseConnection, ExecutionCancellation,
    Extensions, FromWebSocketMessageParts, MessageDeadline, NoReply, Payload,
    WebSocketActionError, WebSocketController, WebSocketControllerInitError,
    WebSocketControllerTrait, WebSocketErrorCode, WebSocketMessageInvocation,
    async_trait, websocket_controller};
use serde::Deserialize;

pub struct AuthorizedSubject(String);
impl FromWebSocketMessageParts for AuthorizedSubject {
    type Rejection = WebSocketActionError;
    fn from_message_parts(invocation: &mut WebSocketMessageInvocation)
        -> impl Future<Output = Result<Self, Self::Rejection>> + Send
    {
        let result = invocation.principal().map(|p| Self(p.subject().to_owned()))
            .ok_or_else(|| WebSocketActionError::rejected(
                WebSocketErrorCode::new("AUTH_REQUIRED").expect("sabit kod"),
                "Kimlik doğrulaması gerekli.",
            ).expect("sabit mesaj"));
        std::future::ready(result)
    }
}

#[derive(Deserialize)]
struct TextInput { text: String }
#[derive(WebSocketController)]
#[namespace("techniques")]
#[timeout(seconds = 10)]
#[asyncapi(tag = "techniques", summary = "Mesaj teknikleri")]
struct TechniquesController;
#[async_trait]
impl WebSocketControllerTrait for TechniquesController {
    async fn new(_: Arc<Extensions>) -> Result<Self, WebSocketControllerInitError> { Ok(Self) }
}
#[websocket_controller]
impl TechniquesController {
    #[message("validated")]
    #[asyncapi(summary = "Boş olmayan metin", operation_id = "validatedText")]
    async fn validated(&self, Payload(input): Payload<TextInput>, subject: AuthorizedSubject)
        -> Result<Ack<String>, WebSocketActionError>
    {
        if input.text.trim().is_empty() {
            return Err(WebSocketActionError::rejected(
                WebSocketErrorCode::new("TEXT_REQUIRED").expect("sabit kod"),
                "Metin boş olamaz.",
            ).expect("sabit mesaj"));
        }
        let _ = subject;
        Ok(Ack::new(input.text))
    }
    #[message("binary")]
    async fn binary(&self, BinaryPayload(bytes): BinaryPayload)
        -> Result<Ack<Vec<u8>>, WebSocketActionError>
    { Ok(Ack::binary(bytes)) }
    #[message("stop")]
    async fn stop(&self) -> Result<CloseConnection, WebSocketActionError> {
        CloseConnection::normal("completed").map_err(WebSocketActionError::internal)
    }
    #[message("bounded")]
    #[timeout(seconds = 2)]
    async fn bounded(&self, stopping: ExecutionCancellation, deadline: MessageDeadline)
        -> Result<NoReply, WebSocketActionError>
    {
        let work = tokio::time::sleep(std::time::Duration::from_millis(10));
        let _remaining = deadline.remaining();
        tokio::select! {
            biased;
            _ = stopping.cancelled() => Ok(NoReply),
            _ = work => Ok(NoReply),
        }
    }
}

Download the complete example.

Signature rules#

At most 16 owned extractors are allowed. Reference parameters, raw WsRequest, and direct invocation parameters are not accepted in action signatures. Payload<T>, Service<S> is a valid order; Service is a parts extractor and resolves first. Each action has only one payload consumer.

The example's custom FromWebSocketMessageParts implementation is a public extension point. Invocation is the input to that implementation, not an action parameter. Define DTOs requiring DeserializeOwned with Serde.

Optional values and errors#

Optional support does not automatically apply to every type. Option<Principal>, Option<ConnectionLocal<T>>, and Option<MessageLocal<T>> can represent absence. Do not assume the same behavior for Option<Payload<T>> or Option<Service<T>>. Malformed data is not swallowed as None.

Serde parsing and domain validation are separate. In the example, requiring text to be a string is parsing; requiring it to be nonempty is a business rule. Errors and validation turns this distinction into public errors. Lifecycle hooks have no payload, and their cancellation phase must follow the hook contract.