nhttp_ws (nhttp_lib v1.3.0)

View Source

WebSocket message-level codec, handshake, and session API (RFC 6455).

Sits on top of nhttp_ws_frame (per-frame codec) and adds:

  • The handshake half: server validation and client/server response builders, Sec-WebSocket-Accept derivation.
  • A framing-aware buffer (ws_decoder/0) that reassembles continuation frames into messages and surfaces interleaved control frames inline.
  • A process-coupled session API used by the H1 / H2 / H3 connection processes after upgrade.

The pure per-frame primitives (encode/1,2, decode/1, decode_unmasked/1, encode_masked/1) are exported here as thin shims over nhttp_ws_frame for backwards compatibility.

Summary

Functions

Generate the Sec-WebSocket-Accept header value.

Send a message to every WebSocket session multiplexed on the connection that owns Session (or directly to the connection pid). Equivalent to Pid ! Msg. Handlers receive it via handle_ws_info/3 with broadcast semantics.

Initiate the §5.5.1 close handshake with status 1000 and no reason.

Initiate the §5.5.1 close handshake with explicit code and reason.

Initiate the §5.5.1 close handshake. Pending sends flush before the reciprocal CLOSE is written unless force => true is set in Opts.

Decode a masked WebSocket frame (client-to-server). Returns {ok, Message, Rest} on success, {more, MinBytes} if more data needed.

Decode an unmasked WebSocket frame (server-to-client). RFC 6455 Section 5.1: a server MUST NOT mask frames sent to clients.

decode_with_state/2 deprecated

Decode one frame, reassembling fragments (RFC 6455 §5.4). Control frames between fragments are returned at once.

Create a new stateful decoder for the given role, with the default max_message_size of 16 MiB. Equivalent to decoder_new(Role, #{}).

Create a new stateful decoder honouring max_message_size from the runtime opts. The option defaults to 16 MiB and bounds two different things. The cumulative payload of a fragmented message, and the payload of a single non-control frame, are measured after the bytes arrive. feed/2 measures the cumulative payload at the header of each continuation frame. Exceeding the option returns {error, message_too_large}. The payload length that a frame header declares is refused before any payload is buffered (RFC 6455 §10.4). Exceeding the option returns {error, {frame_too_large, DeclaredLength}}, and the caller closes with status 1009 (§7.4.1). This bound never drops below 125 bytes, the largest control frame that §5.5 permits. Pass max_message_size => infinity to remove both bounds.

Encode a WebSocket message as an unmasked frame (server-to-client).

Encode a WebSocket message with options. The only option is mask => boolean(); defaults to false.

Encode a binary message (server, unmasked).

Encode a close frame with no status code (server, unmasked).

Encode a close frame with status code and reason (server, unmasked).

Encode a masked WebSocket message (client, RFC 6455 Section 5.1).

Encode a masked binary message (client).

Encode a masked close frame with no status code (client).

Encode a masked close frame with status code and reason (client).

Encode a masked ping frame with empty payload (client).

Encode a masked ping frame with payload (client).

Encode a masked pong frame (client).

Encode a masked text message (client).

Encode a ping frame with empty payload (server, unmasked).

Encode a ping frame with payload (server, unmasked).

Encode a pong frame, must echo ping payload (server, unmasked).

Encode a text message (server, unmasked).

Decode the bytes of one read and return the messages they complete, in wire order. The decoder keeps any unfinished frame, so the caller keeps no buffer. An empty list means more bytes are needed.

Return true while the decoder holds the bytes of a frame that is not complete. A fragmented message between two complete frames is not pending.

Generate a random Sec-WebSocket-Key (RFC 6455 Section 4.1).

Generate the complete handshake response.

Deliver an arbitrary Erlang term to a specific session's handle_ws_info/3 callback. Use this instead of a bare Pid ! Msg when the connection multiplexes more than one session and you want to address one in particular.

Liveness check for the connection process owning the session.

Build a fresh session handle owned by the calling process. Used by the nhttp connection processes (H1/H2/H3) at upgrade time. The constructor is the only sanctioned way to mint a session. The record itself is opaque to callers outside nhttp_lib.

Same as new_session/3 but lets the caller supply an existing reference (useful when the conn process needs to keep the same ref across two different per-stream records, e.g. during in-place migrations).

Same as new_session/4 for HTTP/1.1. Sends write to Socket from the caller.

Return the connection process pid for the session.

Send an empty WebSocket PING (§5.5.2) on the session.

Send a WebSocket PING with the given payload (≤ 125 bytes per §5.5).

Send a message and wait for the write. On a new_session/5 session the caller writes to the socket and Opts is ignored. Otherwise the connection process writes it within Opts#timeout ms. Returns {error, gone} once the session is closed.

Enqueue a message on the connection process, or {error, would_block} when its outbox is full. On a new_session/5 session it is a blocking socket write from the caller and never returns would_block.

Send a binary message.

Send a UTF-8 text message (caller is responsible for valid UTF-8).

Return the unique reference stamped on the session at creation. The conn process compares it against the ref it remembers per active stream and rejects sends whose ref no longer matches (stale handle after the stream ended but before the conn pid died).

Return the transport-level stream id, or undefined for HTTP/1.1.

Return the transport carrying the session.

Validate the server's Sec-WebSocket-Accept value against the client key.

Validate WebSocket upgrade request headers. Returns {ok, Key} if valid, {error, Reason} otherwise.

Types

close_code()

-type close_code() :: nhttp_ws_frame:close_code().

decode_error()

-type decode_error() ::
          nhttp_ws_frame:header_error() |
          control_frame_too_large | fragmented_control_frame | expected_continuation |
          message_too_large | invalid_utf8 | invalid_close_payload | invalid_close_code |
          invalid_close_reason |
          {unknown_opcode, 0..15}.

decode_result()

-type decode_result() :: nhttp_ws_frame:decode_result().

feed_result()

-type feed_result() :: {ok, [ws_message()], ws_decoder()} | {error, decode_error(), [ws_message()]}.

session()

-opaque session()

stateful_decode_result()

-type stateful_decode_result() ::
          {ok, ws_message(), Rest :: binary(), ws_decoder()} |
          {continue, Rest :: binary(), ws_decoder()} |
          {more, MinBytes :: pos_integer(), ws_decoder()} |
          {error, term()}.

ws_close_opts()

-type ws_close_opts() :: #{force => boolean(), timeout => timeout()}.

ws_decoder()

-opaque ws_decoder()

ws_frame()

-type ws_frame() :: {text, binary()} | {binary, binary()} | {ping, binary()} | {pong, binary()}.

ws_message()

-type ws_message() :: nhttp_ws_frame:ws_message().

ws_opcode()

-type ws_opcode() :: nhttp_ws_frame:ws_opcode().

ws_runtime_opts()

-type ws_runtime_opts() ::
          #{deliver_ping => boolean(),
            deliver_pong => boolean(),
            max_message_size => pos_integer() | infinity,
            outbound_high_water => pos_integer(),
            outbound_low_water => pos_integer(),
            idle_timeout => timeout()}.

ws_send_opts()

-type ws_send_opts() :: #{timeout => timeout(), priority => low | normal}.

ws_session_opts()

-type ws_session_opts() :: #{subprotocol => binary() | undefined, extensions => [binary()]}.

Functions

accept_key(Key)

-spec accept_key(Key :: binary()) -> binary().

Generate the Sec-WebSocket-Accept header value.

broadcast/2

-spec broadcast(session() | pid(), term()) -> ok.

Send a message to every WebSocket session multiplexed on the connection that owns Session (or directly to the connection pid). Equivalent to Pid ! Msg. Handlers receive it via handle_ws_info/3 with broadcast semantics.

close(Session)

-spec close(session()) -> ok.

Initiate the §5.5.1 close handshake with status 1000 and no reason.

close(Session, Code, Reason)

-spec close(session(), close_code(), binary()) -> ok.

Initiate the §5.5.1 close handshake with explicit code and reason.

close/4

-spec close(session(), close_code(), binary(), ws_close_opts()) -> ok.

Initiate the §5.5.1 close handshake. Pending sends flush before the reciprocal CLOSE is written unless force => true is set in Opts.

decode(Data)

-spec decode(binary()) -> decode_result().

Decode a masked WebSocket frame (client-to-server). Returns {ok, Message, Rest} on success, {more, MinBytes} if more data needed.

decode_unmasked(Data)

-spec decode_unmasked(binary()) -> decode_result().

Decode an unmasked WebSocket frame (server-to-client). RFC 6455 Section 5.1: a server MUST NOT mask frames sent to clients.

decode_with_state/2

This function is deprecated. Use feed/2 instead.
-spec decode_with_state(binary(), ws_decoder()) -> stateful_decode_result().

Decode one frame, reassembling fragments (RFC 6455 §5.4). Control frames between fragments are returned at once.

  • {ok, Message, Rest, Decoder}: message complete. Keep Rest.
  • {continue, Rest, Decoder}: fragment buffered. Keep Rest.
  • {more, MinBytes, Decoder}: keep the input and the decoder, and wait.

Text is scanned as UTF-8 as it arrives (§8.1). Each call parses the buffer from its first byte, so a frame split over many reads costs quadratic time. Prefer feed/2.

decoder_new(Role)

-spec decoder_new(client | server) -> ws_decoder().

Create a new stateful decoder for the given role, with the default max_message_size of 16 MiB. Equivalent to decoder_new(Role, #{}).

decoder_new(Role, Opts)

-spec decoder_new(client | server, ws_runtime_opts()) -> ws_decoder().

Create a new stateful decoder honouring max_message_size from the runtime opts. The option defaults to 16 MiB and bounds two different things. The cumulative payload of a fragmented message, and the payload of a single non-control frame, are measured after the bytes arrive. feed/2 measures the cumulative payload at the header of each continuation frame. Exceeding the option returns {error, message_too_large}. The payload length that a frame header declares is refused before any payload is buffered (RFC 6455 §10.4). Exceeding the option returns {error, {frame_too_large, DeclaredLength}}, and the caller closes with status 1009 (§7.4.1). This bound never drops below 125 bytes, the largest control frame that §5.5 permits. Pass max_message_size => infinity to remove both bounds.

encode(Message)

-spec encode(ws_message()) -> iodata().

Encode a WebSocket message as an unmasked frame (server-to-client).

encode(Message, Opts)

-spec encode(ws_message(), nhttp_ws_frame:encode_opts()) -> iodata().

Encode a WebSocket message with options. The only option is mask => boolean(); defaults to false.

encode_binary(Data)

-spec encode_binary(iodata()) -> iodata().

Encode a binary message (server, unmasked).

encode_close()

-spec encode_close() -> iodata().

Encode a close frame with no status code (server, unmasked).

encode_close(Code, Reason)

-spec encode_close(Code :: close_code(), Reason :: binary()) -> iodata().

Encode a close frame with status code and reason (server, unmasked).

encode_masked(Message)

-spec encode_masked(ws_message()) -> iodata().

Encode a masked WebSocket message (client, RFC 6455 Section 5.1).

encode_masked_binary(Data)

-spec encode_masked_binary(iodata()) -> iodata().

Encode a masked binary message (client).

encode_masked_close()

-spec encode_masked_close() -> iodata().

Encode a masked close frame with no status code (client).

encode_masked_close(Code, Reason)

-spec encode_masked_close(Code :: close_code(), Reason :: binary()) -> iodata().

Encode a masked close frame with status code and reason (client).

encode_masked_ping()

-spec encode_masked_ping() -> iodata().

Encode a masked ping frame with empty payload (client).

encode_masked_ping(Data)

-spec encode_masked_ping(binary()) -> iodata().

Encode a masked ping frame with payload (client).

encode_masked_pong(Data)

-spec encode_masked_pong(binary()) -> iodata().

Encode a masked pong frame (client).

encode_masked_text(Data)

-spec encode_masked_text(iodata()) -> iodata().

Encode a masked text message (client).

encode_ping()

-spec encode_ping() -> iodata().

Encode a ping frame with empty payload (server, unmasked).

encode_ping(Data)

-spec encode_ping(binary()) -> iodata().

Encode a ping frame with payload (server, unmasked).

encode_pong(Data)

-spec encode_pong(binary()) -> iodata().

Encode a pong frame, must echo ping payload (server, unmasked).

encode_text(Data)

-spec encode_text(iodata()) -> iodata().

Encode a text message (server, unmasked).

feed(Data, Dec)

-spec feed(binary(), ws_decoder()) -> feed_result().

Decode the bytes of one read and return the messages they complete, in wire order. The decoder keeps any unfinished frame, so the caller keeps no buffer. An empty list means more bytes are needed.

Each byte is unmasked and UTF-8 scanned once, on arrival. The limits of decoder_new/2 apply. On error, Messages holds the messages completed before the violation. Do not mix with decode_with_state/2 on one decoder.

frame_pending/1

-spec frame_pending(ws_decoder()) -> boolean().

Return true while the decoder holds the bytes of a frame that is not complete. A fragmented message between two complete frames is not pending.

generate_key()

-spec generate_key() -> binary().

Generate a random Sec-WebSocket-Key (RFC 6455 Section 4.1).

handshake_response(Key)

-spec handshake_response(Key :: binary()) -> iodata().

Generate the complete handshake response.

handshake_response(Key, SessionOpts)

-spec handshake_response(Key :: binary(), SessionOpts :: ws_session_opts()) -> iodata().

info/2

-spec info(session(), term()) -> ok.

Deliver an arbitrary Erlang term to a specific session's handle_ws_info/3 callback. Use this instead of a bare Pid ! Msg when the connection multiplexes more than one session and you want to address one in particular.

is_alive/1

-spec is_alive(session()) -> boolean().

Liveness check for the connection process owning the session.

new_session(Transport, ConnPid, StreamId)

-spec new_session(h1 | h2 | h3, pid(), undefined | nhttp_lib:stream_id()) -> session().

Build a fresh session handle owned by the calling process. Used by the nhttp connection processes (H1/H2/H3) at upgrade time. The constructor is the only sanctioned way to mint a session. The record itself is opaque to callers outside nhttp_lib.

new_session(Transport, ConnPid, StreamId, Ref)

-spec new_session(h1 | h2 | h3, pid(), undefined | nhttp_lib:stream_id(), reference()) -> session().

Same as new_session/3 but lets the caller supply an existing reference (useful when the conn process needs to keep the same ref across two different per-stream records, e.g. during in-place migrations).

new_session/5

-spec new_session(h1, pid(), undefined, reference(), nhttp_sock:t()) -> session().

Same as new_session/4 for HTTP/1.1. Sends write to Socket from the caller.

owner/1

-spec owner(session()) -> pid().

Return the connection process pid for the session.

ping(Session)

-spec ping(session()) -> ok | {error, term()}.

Send an empty WebSocket PING (§5.5.2) on the session.

ping(Session, Payload)

-spec ping(session(), binary()) -> ok | {error, term()}.

Send a WebSocket PING with the given payload (≤ 125 bytes per §5.5).

send(Session, Msg)

-spec send(session(), ws_message()) -> ok | {error, term()}.

Send a message and wait for the write. On a new_session/5 session the caller writes to the socket and Opts is ignored. Otherwise the connection process writes it within Opts#timeout ms. Returns {error, gone} once the session is closed.

send/3

-spec send(session(), ws_message(), ws_send_opts()) -> ok | {error, term()}.

send_async/2

-spec send_async(session(), ws_message()) -> ok | {error, would_block | gone | term()}.

Enqueue a message on the connection process, or {error, would_block} when its outbox is full. On a new_session/5 session it is a blocking socket write from the caller and never returns would_block.

send_binary(Session, Data)

-spec send_binary(session(), iodata()) -> ok | {error, term()}.

Send a binary message.

send_text(Session, Data)

-spec send_text(session(), iodata()) -> ok | {error, term()}.

Send a UTF-8 text message (caller is responsible for valid UTF-8).

session_ref/1

-spec session_ref(session()) -> reference().

Return the unique reference stamped on the session at creation. The conn process compares it against the ref it remembers per active stream and rejects sends whose ref no longer matches (stale handle after the stream ended but before the conn pid died).

stream_id/1

-spec stream_id(session()) -> undefined | nhttp_lib:stream_id().

Return the transport-level stream id, or undefined for HTTP/1.1.

transport/1

-spec transport(session()) -> h1 | h2 | h3.

Return the transport carrying the session.

validate_accept(ClientKey, ServerAccept)

-spec validate_accept(ClientKey :: binary(), ServerAccept :: binary()) -> ok | {error, invalid_accept}.

Validate the server's Sec-WebSocket-Accept value against the client key.

validate_upgrade/1

-spec validate_upgrade(nhttp_h1:req()) -> {ok, binary()} | {error, term()}.

Validate WebSocket upgrade request headers. Returns {ok, Key} if valid, {error, Reason} otherwise.