Struct AxumWebSocketTransport 

pub struct AxumWebSocketTransport { /* private fields */ }

Axum WebSocket transport implementation

Implementations

impl AxumWebSocketTransport

pub fn new() -> Self

Create a transport with the default rate-limit configuration.

See RateLimitConfig::default for the limits applied.

pub fn with_rate_limit_config(config: RateLimitConfig) -> Self

Create a transport with an explicit rate-limit configuration.

Use RateLimitConfig::high_traffic or RateLimitConfig::low_resource for preset profiles, or construct a custom RateLimitConfig.

Spawns a background sweep that periodically aborts streaming sessions older than SESSION_MAX_AGE via AdaptiveStreamController::cleanup_expired_sessions; the sweep holds only a std::sync::Weak reference to the controller, so it exits once every Arc<AdaptiveStreamController> (including this transport’s own) is dropped, instead of keeping the controller alive forever.

pub fn with_allowed_origins(self, allowed_origins: Vec<String>) -> Self

Restrict WebSocket upgrades to the given Origin allow-list.

Reuses the config semantics of HttpServerConfig::allowed_origins’s CORS allow-list:

• [] (the default) — deny all cross-origin upgrades (fail-closed)
• ["*"] — allow any origin. More dangerous here than the equivalent CORS Any: browsers attach ambient credentials to a WebSocket handshake regardless of the server’s Origin response, unlike CORS, so a wildcard here fully re-enables the CSWSH this allow-list exists to prevent.
• "*" mixed with explicit origins — treated as deny-all (fail closed); unlike the CORS layer this cannot be surfaced as a construction error, since this builder returns Self

Explicit entries that can never match a real Origin header (no scheme://, a trailing path, or uppercase letters) are kept fail-closed but logged with warn!, since they’d otherwise silently deny every browser connection with no diagnostic.

This only governs requests that carry an Origin header. A request without one is always allowed to upgrade regardless of this list — see Self::upgrade_handler for why that is safe.

pub async fn upgrade_handler( ws: WebSocketUpgrade, __arg1: ConnectInfo<SocketAddr>, headers: HeaderMap, __arg3: State<Arc<Self>>, ) -> Response

Handle WebSocket upgrade for Axum.

Extracts the peer address via ConnectInfo and rejects upgrade requests that exceed the per-IP request budget with HTTP 429 before any WebSocket frames are exchanged.

Also rejects, with HTTP 403, upgrades carrying an Origin header not in Self::with_allowed_origins’s allow-list — see that method and the check’s own doc comment below for the CSWSH threat model and why a missing Origin header is allowed.

Configures axum/tungstenite’s transport-level max_message_size and max_frame_size from the transport’s RateLimitConfig::max_frame_size, so an oversized frame is rejected during frame assembly instead of being fully buffered first and only rejected afterward by the application-level check_message call (which remains as defense-in-depth for messages under the transport cap but still over policy in other ways).

The router must be served with into_make_service_with_connect_info::<SocketAddr>() so the peer address is populated; otherwise the upgrade response is HTTP 500.

pub async fn handle_socket( self: Arc<Self>, socket: WebSocket, client_ip: IpAddr, )

Handle WebSocket connection lifecycle

pub fn controller(&self) -> Arc<AdaptiveStreamController>

Returns a shared handle to the underlying AdaptiveStreamController.

pub async fn active_connection_count(&self) -> usize

Returns the number of currently active WebSocket connections.

Useful for observability, health endpoints, and integration tests.

Trait Implementations

impl Default for AxumWebSocketTransport

fn default() -> Self

Returns the “default value” for a type.

impl WebSocketTransport for AxumWebSocketTransport

fn send_frame( &self, connection: Arc<Self::Connection>, message: WsMessage, ) -> Self::SendFrameFuture<'_>

The channel this queues onto is drained by the same tokio::select! loop in Self::handle_socket that also awaits handle_websocket_message inline. Calling send_frame from within that inline handling path (directly or transitively) would deadlock the connection: the loop can’t reach outgoing_rx.recv() again until the in-flight branch finishes, so a blocking send would wait forever on a receiver that can’t run. Using try_send here keeps that latent hazard from becoming a real deadlock — see WebSocketTransport::send_frame’s doc for the general contract.

Always returns Ok(()) even when the frame is dropped (channel full, or larger than MAX_QUEUED_OUTGOING_BYTES) — this mirrors the underlying channel’s own fire-and-forget delivery guarantee (an Ok try_send on a normal mpsc channel doesn’t promise the receiver will ever read the item either) and matches how the broadcast-based frame_rx delivery path also has no per-frame delivery acknowledgment. Both drop reasons are logged via warn!.

fn handle_message( &self, connection: Arc<Self::Connection>, message: WsMessage, ) -> Self::HandleMessageFuture<'_>

The StreamInit arm records the created session under connection in connection_sessions so Self::handle_socket’s teardown can abort it on disconnect. Nothing else drains that map: a caller that drives this method directly, bypassing handle_socket (e.g. a test, or a future non-axum trait caller), leaves its session’s entry there indefinitely — WebSocketTransport::close_stream removes the session from the controller but does not touch connection_sessions.

type Connection = String

Concrete connection type the implementor uses for I/O.

type StartStreamFuture<'a> = impl Future<Output = Result<String, Error>> + Send + 'a where Self: 'a

Future type for starting stream

type SendFrameFuture<'a> = impl Future<Output = Result<(), Error>> + Send + 'a where Self: 'a

Future type for sending frame

type HandleMessageFuture<'a> = impl Future<Output = Result<(), Error>> + Send + 'a where Self: 'a

Future type for handling message

type CloseStreamFuture<'a> = impl Future<Output = Result<(), Error>> + Send + 'a where Self: 'a

Future type for closing stream

fn start_stream( &self, _connection: Arc<Self::Connection>, data: Value, options: StreamOptions, ) -> Self::StartStreamFuture<'_>

Start streaming session

fn close_stream(&self, session_id: &str) -> Self::CloseStreamFuture<'_>

Close streaming session