WebSocket Support

In an HTTP exchange, the client sends a request and the server sends a response. An open WebSocket connection lets either side send a message at any time. Use it for interactive applications such as chat and live notifications. It can also carry another protocol, such as MQTT, over an ordinary web connection.

WebSocket support in ntex

Most web applications only need web::ws. ntex provides four APIs for different ways of handling a WebSocket connection:

All four share the codec and protocol types in ntex::ws. WebSocket support is enabled by default through the ws Cargo feature.

Every WebSocket connection starts with an HTTP/1.1 request. The client asks to upgrade the connection, and the server accepts with 101 Switching Protocols. The peers then exchange WebSocket frames instead of HTTP messages. The APIs in this chapter support HTTP/1 upgrades, not HTTP/2 extended CONNECT.

A ws:// URL uses a plain connection; wss:// uses TLS. For wss://, configure a TLS connector with a backend such as OpenSSL, rustls, or Windows Schannel. Changing the URL alone does not enable TLS on WsClient. See WebSocket TLS configuration.

The optional compress feature adds permessage-deflate support to the codec and client. web::ws::start() does not negotiate compression on its own.

Frames, messages, and fragmentation

The frame API gives you incoming data as a Frame. To send data, use a Message. Both types have variants for text, binary, ping, pong, close, and continuation frames.

A sender can split one message into several frames. The first fragment is Frame::Continuation(Item::FirstText(_)) or FirstBinary(_). Zero or more Continue(_) fragments may follow. The final fragment is Last(_).

The codec checks fragment order but does not join fragments. If your handler needs whole messages, collect the fragments yourself or reject fragmented input.

The codec also checks text encoding and size limits:

• Text frames arrive as raw bytes. The codec checks each text message for valid UTF-8 and fails the connection if the check fails. A multi-byte character can span fragments, so convert the joined message to a string rather than converting each fragment. Codec::validate_utf8(false) disables this check.
• Each frame is limited to 64 KiB by Codec::max_size(). Each message, including all its fragments, is limited to 4 MiB by Codec::max_message_size(). A larger frame or message fails with ProtocolError::Overflow. Raise both limits to accept a single-frame message over 4 MiB.

Servers with web::ws

Add a WebSocket endpoint as an ordinary route with a GET handler. In the handler, call web::ws::start with the request and a service for incoming frames. The service returns Some(Message) to reply to a frame, or None to send nothing.

This echo server handles text and binary messages that arrive in a single frame:

use std::io;

use ntex::web::{self, HttpRequest, ws};

async fn echo(frame: ws::Frame) -> Result<Option<ws::Message>, io::Error> {
    let message = match frame {
        ws::Frame::Text(bytes) => match String::from_utf8(bytes.to_vec()) {
            Ok(text) => ws::Message::Text(text.into()),
            Err(_) => ws::Message::Close(Some(ws::CloseCode::Invalid.into())),
        },
        ws::Frame::Binary(bytes) => ws::Message::Binary(bytes),
        ws::Frame::Ping(bytes) => ws::Message::Pong(bytes),
        ws::Frame::Pong(_) => return Ok(None),
        ws::Frame::Close(reason) => ws::Message::Close(reason),
        ws::Frame::Continuation(_) => {
            ws::Message::Close(Some(ws::CloseCode::Unsupported.into()))
        }
    };
    Ok(Some(message))
}

async fn socket(req: HttpRequest) {
    if let Err(err) = ws::start(&req, None, echo).await {
        eprintln!("WebSocket connection failed: {err:?}");
    }
}

#[ntex::main]
async fn main() -> io::Result<()> {
    web::server(async |_| {
        web::App::new().route("/ws/", web::get().to(socket))
    })
    .bind("127.0.0.1:8080", ntex::SharedCfg::default())?
    .run()
    .await
}

Run the server, then connect a WebSocket client to ws://127.0.0.1:8080/ws/. The client example below sends a text message and reads the echo. An ordinary HTTP client does not switch protocols just because the path ends in /ws/; it must send the WebSocket upgrade headers.

Follow one text message through the server: the route calls socket(), which hands the connection to start(). The dispatcher decodes an incoming frame and calls echo(). That function turns the text bytes into an outgoing Message::Text, and the dispatcher writes it back. The Pong branch returns None because a pong is an acknowledgement, not something to echo again. The continuation branch closes rather than silently losing part of a message.

This is an echo endpoint, not a chat room: each connection replies only to its own sender. To broadcast to other users, keep authorized room membership and a sink for each connected peer. Remove a peer when its connection ends, and bound any per-peer queue. A slow reader should not make every other room member wait indefinitely or let queued messages exhaust server memory.

Your handler does not build an HTTP response. start() validates the handshake, sends the 101 response, and takes over the connection from the HTTP layer. It then runs the WebSocket dispatcher until the connection closes. The await lasts for the connection’s lifetime, beyond the initial setup.

During the handoff, start() cancels the HTTP layer’s read timer. It also switches to an I/O configuration without idle keepalive. If the handshake is invalid, it sends the HTTP error response itself.

After takeover, returning an HttpResponse from your handler sends nothing else. On failure, start() returns a WsError. This covers frame-service errors, protocol errors, read and write timeouts, and I/O errors.

Sending outside the frame handler

Use a WsSink to send messages without waiting for an incoming frame. For example, another task can use it to push a notification. Returning a message from the frame handler remains the easiest way to reply to a frame.

The frame service receives the sink as its state. Use fn_service_st to access it, or implement Service<WsSink, Frame> yourself.

A WsSink is cheap to clone. Every clone sends to the same connection. Give a clone to tasks that push messages, such as timers, pub/sub subscriptions, or chat room broadcasters. Each task can call sink.send(...) independently of the frame handler and the other tasks.

This service answers pings through the sink instead of returning the pong:

use std::io;
use ntex::service::fn_service_st;
use ntex::web::ws;

let service = fn_service_st(async |sink: &ws::WsSink, frame: ws::Frame| {
    if let ws::Frame::Ping(bytes) = frame {
        sink.send(ws::Message::Pong(bytes))
            .await
            .map_err(io::Error::other)?;
    }
    Ok::<Option<ws::Message>, io::Error>(None)
});

The handler returns None because it has already sent the pong. start() does not send another one.

Sink clones share the connection’s state. Once any clone sends a close message, none of them can send more messages. The sink uses Rc for reference counting and stays on the connection’s worker thread. Start sender tasks with ntex::rt::spawn on that worker rather than on a separate thread.

send() waits while the connection is under write backpressure. If you keep your own outgoing message queue, limit its size so a slow client cannot make it grow without bound.

The basic WebSocket example uses a sink for its heartbeat and includes a browser page, a Rust client, and the server. The server’s readiness hook starts the heartbeat task once. Its shutdown hook stops the task.

The task sends pings and records when the client last sent a ping or pong. If the client stays quiet for too long, the task closes the connection.

Choosing a subprotocol

Use a subprotocol to agree on what the connection will carry. For example, mqtt identifies MQTT. The client lists the protocols it supports, and the server chooses one.

Read the client’s offers with web::ws::subprotocols. Pass your choice to start():

let chosen = ntex::web::ws::subprotocols(&req)
    .find(|protocol| *protocol == "my-application");
ntex::web::ws::start(&req, chosen, service).await?;

Pass Some(protocol) only for a valid token that the client offered. None accepts the connection without a subprotocol. If your application needs a subprotocol, reject clients that do not offer one you support.

The handshake checks the WebSocket protocol; it does not authorize clients. Authenticate the client before calling start(). For browser clients, check the Origin header before that call too.

Heartbeats and closing the connection

You can let start() answer control frames or handle them in your frame service. Ping, pong, and close frames reach the service just like data frames.

If your service returns None for a ping without sending a pong through the sink, start() sends a pong with the same payload. Return your own message to answer differently.

When the peer sends a close frame and your service returns None, start() echoes the peer’s close code unless you have already sent a close. It closes the I/O stream after your service handles the peer’s close.

Answering pings is not the same as sending heartbeats. start() can answer a peer’s ping, but it does not schedule periodic pings or decide how long a silent peer may stay connected. If your application needs to detect broken connections promptly, keep a last-seen timestamp per connection and run a small timer task. For example, send a ping every five seconds and consider thirty seconds without a ping or pong a failed connection. These are application policy values, not ntex defaults.

Update that timestamp in the frame service when a ping or pong arrives. Keep the timer and frame service on the same worker, and arrange for the timer to stop when the dispatcher finishes. Decide whether ordinary data messages also count as activity; an application-level acknowledgement may matter more than a WebSocket pong for some protocols. The HTTP keepalive setting is not a replacement for this policy after the upgrade.

To close from your side, return or send Message::Close. This starts the closing-handshake timeout. Set it with WsClientConfig::set_close_timeout() in the connection’s shared configuration. Despite the type’s name, this setting also applies to server-side sinks.

Stop background tasks such as heartbeats when the connection ends, as the basic example does. If a peer stops responding, close the I/O stream with sink.io().close(). Use the same call whenever you need to close without waiting for further application messages.

Use web::ws::start_with to handle dispatcher events yourself. It performs the same upgrade, but your service receives DispatchItem<WsSink> instead of Frame. Your service must handle control and stop events, including errors and shutdown. start_with() does not answer pings or close frames for you.

Testing the Upgrade and Message Exchange

Use a real test server for a WebSocket endpoint: the upgrade transfers an actual connection, which an in-memory handler test cannot provide. Put this test alongside the echo and socket functions from the server example:

use futures::StreamExt;
use ntex::{
    SharedCfg, http::StatusCode, time::Seconds, util::Bytes, ws::WsClient,
};

#[ntex::test]
async fn echo_round_trip() {
    let server = web::test::server(async |_| {
        web::App::new().route("/ws/", web::get().to(socket))
    });
    ntex::time::timeout(Seconds(5), async {
        let connection = WsClient::new(server.url("/ws/"), SharedCfg::default())
            .connect().await.unwrap();
        assert_eq!(connection.response().status(), StatusCode::SWITCHING_PROTOCOLS);
        let sink = connection.sink();
        let mut frames = connection.seal().receiver();

        sink.send(ws::Message::Text("hello".into())).await.unwrap();
        assert_eq!(
            frames.next().await.unwrap().unwrap(),
            ws::Frame::Text(Bytes::from_static(b"hello")),
        );
        sink.send(ws::Message::Ping(Bytes::from_static(b"alive"))).await.unwrap();
        assert_eq!(
            frames.next().await.unwrap().unwrap(),
            ws::Frame::Pong(Bytes::from_static(b"alive")),
        );
        sink.send(ws::Message::Close(Some(ws::CloseCode::Normal.into())))
            .await.unwrap();
        assert!(matches!(
            frames.next().await.unwrap().unwrap(),
            ws::Frame::Close(Some(reason)) if reason.code == ws::CloseCode::Normal
        ));
    }).await.expect("WebSocket exchange timed out");
}

The server binds an available port, so tests can run without claiming a fixed port or starting a separate process. WsClient accepts the http:// URL returned by server.url() for this plain WebSocket connection. The test waits for actual protocol events rather than sleeping: it checks the upgrade, text reply, pong payload, and close acknowledgement. The deadline makes a missing reply fail the test instead of leaving it waiting indefinitely.

For a shorter setup, server.ws_at("/ws/").await connects with the server’s configured WsClient for you. Use the explicit client when you need custom headers, subprotocols, or connection limits. See Testing for test-server configuration and the distinction between in-memory and network tests.

Low-level servers with an HTTP control service

An HTTP control service can accept WebSocket connections without a web application. Use it for a lower-level server, a listener shared by several protocols, or your own I/O loop.

Attach the service with HttpService::h1(...).control(...). For the combined HTTP service, use HttpService::new(...).h1_control(...).

Upgrade requests arrive as h1::Control::Upgrade. Inspect the request to check its path, authentication, Origin, and any required subprotocol. ws::handshake validates the WebSocket handshake and returns a response builder. You still handle authorization and subprotocol selection yourself.

Choose how to handle each upgrade request:

• Reject the request with upgrade.fail(error) or upgrade.fail_with(response).
• Pass it on to the regular application service with upgrade.ack(). This is the path a web::ws handler relies on.
• Take over the connection with upgrade.handle().

handle() gives you the acknowledgement to return. It also returns the I/O stream, request, and HTTP/1 codec. This excerpt from the low-level WebSocket example passes those three values to a separate ws_service:

.h1_control(async move |control: h1::Control<_, _>| {
    let ack = if let h1::Control::Upgrade(upgrade) = control {
        let (ack, io, request, http_codec) = upgrade.handle();
        ws_service.call((request, io, http_codec)).await?;
        ack
    } else {
        control.ack()
    };
    Ok::<_, io::Error>(ack)
})

ws_service checks the handshake and writes the HTTP response with the HTTP/1 codec it received. It then switches to ws::Codec::new() for WebSocket frames. That codec uses the server’s masking rules. A client needs the client rules instead.

handle() transfers ownership without sending a response. Your code must send either an HTTP rejection or the 101 response. Keep using the I/O stream you received. It may already hold the start of the first WebSocket frame, buffered immediately after the HTTP headers.

You also manage the connection’s lifetime:

• Stop the HTTP timer with io.stop_timer() and choose I/O settings for the new protocol. Otherwise the old header-read timer can fire in the middle of the WebSocket session.
• Drive the connection with io.recv() and io.send(), or use an I/O dispatcher. Your loop must handle ping/pong, fragmented messages, close frames, and protocol and transport errors.
• Finish the closing handshake, shut down the I/O stream, and stop any background tasks that belong to the connection.

The example changes I/O settings with Io::set_config(). This call is unsafe. Before calling it, ensure no reference borrowed from the old configuration remains alive. The I/O chapter explains why.

The complete example includes an OpenSSL acceptor and a web application that serves static files for ordinary HTTP requests. Its WebSocket loop echoes text and binary frames, answers pings, and uses pongs for its heartbeat. It rejects fragmented messages instead of assembling them.

The example’s README lists the server and client commands. It also explains how to set up the test certificate.

Clients with ntex::ws

Create a WsClient and call connect().await to connect to a WebSocket server. Store client settings in a WsClientConfig inside a SharedCfg. A successful handshake returns a WsConnection holding the server’s response and a sink for sending messages.

This client connects to the echo server above, sends "hello", prints the reply, and closes the connection:

use std::io;

use futures::StreamExt;
use ntex::{SharedCfg, time::Seconds, ws};

#[ntex::main]
async fn main() -> io::Result<()> {
    let cfg = SharedCfg::new("WS-CLIENT").add(
        ws::WsClientConfig::new()
            .set_max_frame_size(128 * 1024)
            .set_handshake_timeout(Seconds(5))
            .set_close_timeout(Seconds(5)),
    );
    let connection = ws::WsClient::new("ws://127.0.0.1:8080/ws/", cfg)
        .connect()
        .await
        .map_err(io::Error::other)?;
    println!("Handshake: {}", connection.response().status());

    let sink = connection.sink();
    let mut frames = connection.seal().receiver();
    sink.send(ws::Message::Text("hello".into()))
        .await
        .map_err(io::Error::other)?;

    while let Some(frame) = frames.next().await {
        match frame.map_err(|err| io::Error::other(format!("{err:?}")))? {
            ws::Frame::Text(bytes) => {
                println!("Echo: {bytes:?}");
                sink.send(ws::Message::Close(Some(ws::CloseCode::Normal.into())))
                    .await
                    .map_err(io::Error::other)?;
            }
            ws::Frame::Close(_) => break,
            _ => {}
        }
    }
    Ok(())
}

The client reads frames with receiver() and futures::StreamExt, as in the basic WebSocket client. The call to seal() erases the concrete I/O filter type. Calling receiver() then starts the dispatcher in a background task and returns a channel of frames and errors.

The receiver answers the server’s pings and close frame for you. Pings still reach your loop. Dropping the receiver starts the closing handshake.

To use a frame service instead of a channel, call connection.seal().start(service).await. The service works like the server-side service, but its state is () rather than a sink. It returns Option<Message> and must reply to both pings and close frames.

Await or spawn start() to run the dispatcher. Keeping a connection or sink alone does not run it.

Configuration and TLS

Use WsClientConfig to choose handshake options and connection limits:

• Offer subprotocols and add headers, such as Origin or authentication.
• Limit frame size with set_max_frame_size.
• Limit message size with set_max_message_size.
• Set the handshake and close timeouts.

The handshake timeout starts after the connection is established. It covers only the HTTP exchange. Wrap the operation in your own timeout if you need a deadline for the whole connection attempt.

If the server selects a subprotocol, the client checks that it is valid and was offered. The server may also accept without selecting one. If your application requires a particular subprotocol, check the response before sending data.

For wss://, enable the matching Cargo feature and configure a TLS connector. Use WsClient::openssl() for OpenSSL or WsClient::rustls() for rustls. The low-level TLS client shows an OpenSSL setup for a local self-signed certificate. It disables certificate verification for the demo. A real client must verify the server’s certificate and hostname.

With the compress feature, call set_deflate(DeflateConfig::new()) to offer permessage-deflate compression. The server may accept or decline, so enabling it does not guarantee compression.

WebSocket as a byte transport

Use a byte transport to run another protocol without handling WebSocket frames in that protocol. MQTT, for example, already decodes its own packets, acknowledges messages, and keeps the connection alive.

WsTransport is an I/O filter that removes binary WebSocket framing from incoming data and adds it to outgoing data:

socket <-> optional TLS <-> WebSocket filter <-> MQTT or another byte protocol

The protocol receives bytes from binary messages and their continuation fragments. The filter rejects text messages. Unlike the frame services above, it answers pings itself and ignores pongs. It also takes part in the closing handshake when the I/O stream shuts down.

The filter exposes a byte stream without WebSocket message boundaries. One protocol packet may span several WebSocket messages. A single WebSocket message may contain several protocol packets. The protocol’s decoder handles these boundaries. Use the frame API if you need to track where each WebSocket message begins and ends.

Installing the filter

On the server, send the 101 upgrade response first, then add the filter:

let io = ntex::ws::WsTransport::create(io, ntex::ws::Codec::new());

Pass the resulting Io<Layer<WsTransport, F>> to a protocol service. Call io.boxed() if that service expects type-erased I/O.

To add the filter as a service-pipeline step, use WsTransportService. Construct it with WsTransportService::new(ws::Codec::new()). Both approaches require a completed HTTP handshake.

On the client, convert the connection returned by WsClient:

let io = connection.into_transport();

This keeps the client’s codec, including its masking mode and any compression negotiated during the handshake. A fresh codec would use the server’s masking rules, which are wrong for a client.

The inner protocol now owns the stream. Stop using any old frame sink so you do not mix two ways of writing to the same connection.

The I/O filter subsystem section explains how filters work with TLS, buffering, backpressure, and shutdown.

MQTT over WebSocket

The MQTT-over-WebSocket example connects an ntex-mqtt server and client over WebSocket. Both speak MQTT 3.1.1 and use the standard mqtt subprotocol.

The server accepts upgrades only on /mqtt from clients that offer mqtt. It checks the handshake and selects mqtt in its 101 response. It then stops the HTTP timer, installs WsTransport, and passes the boxed stream to v3::MqttServer. The MQTT server handles CONNECT and incoming publishes.

This minimal server receives publishes. It does not act as a broker, manage subscriptions, or forward messages.

The client offers mqtt with WsClientConfig::set_protocols() and checks the server’s choice. A custom connector for v3::client::MqttConnector returns connection.into_transport() instead of a plain socket. The MQTT client uses that stream to send CONNECT and exchange packets as usual.

The client sends a QoS 1 publish and waits for its acknowledgement. The MQTT dispatcher runs at the same time to read that acknowledgement. The client then closes the MQTT session and lets the transport shut down.

MQTT still handles keepalive. WebSocket ping/pong is a separate mechanism and does not replace it.

AMQP over WebSocket

You can also carry AMQP 1.0 over WebSocket with an ntex-amqp server or a custom client connector. The AMQP WebSocket binding uses the amqp subprotocol. Both sides should check that it was negotiated.

Each opening AMQP protocol header must travel in its own WebSocket message. This includes the SASL header when SASL is used. After the headers, AMQP frames may span messages or share them freely.

WsTransport does not track AMQP protocol-header boundaries and cannot enforce this rule alone. Your AMQP handshake integration must keep each header in its own message. Test it against the peer you plan to use.

The AMQP implementation handles the rest, including authentication, sessions, links, idle timeouts, and orderly shutdown. For TLS, use wss:// for the WebSocket connection rather than negotiating TLS inside the AMQP stream.