MQTT 3.1.1 packet encode/decode — only the packets this venue needs.
Why not a full MQTT client library
The reachable broker is wss://data-api.webull.com:8883/mqtt — MQTT over WebSocket.
Tortoise311 ships Transport.Tcp and Transport.SSL only, and its transport behaviour
is sixteen callbacks modelling an active-mode byte stream: setopts,
controlling_process, {:tcp, socket, data} messages to an owner process. Shimming a
frame-oriented WebSocket into that faithfully is more moving parts, and more ways to be
subtly wrong, than encoding the six packet types actually used here.
The connection lifecycle is already solved by websockex, which this family runs for
several venues. Only the framing was missing.
Packets implemented
Outbound: CONNECT (1), PINGREQ (12), DISCONNECT (14), and the acknowledgements
PUBACK (4), PUBREC (5) and PUBCOMP (7) described below.
Inbound: CONNACK (2), PUBLISH (3), PUBREL (6), PINGRESP (13).
QoS 0 is what this package expects, but it does not choose it. The client never
sends a SUBSCRIBE (see below), so the QoS a session is delivered at is set by the
venue's HTTP subscribe, not here, and the documentation states none. This module used to
say "QoS 0 throughout" and leave PUBACK/PUBREC/PUBREL/PUBCOMP unimplemented on
that footing. A broker that publishes at QoS 1 or 2 keeps an in-flight window per session
and stops delivering when it fills with messages nobody acknowledged (MQTT 3.1.1 §4.3):
a connection that stays up, passes keep-alive, and delivers less and then nothing. So a
QoS 1 PUBLISH is acknowledged with PUBACK, a QoS 2 one with PUBREC, and the PUBREL
that follows with PUBCOMP (ack/2). What is delivered, and how often, is the socket's
decision and not this module's, which neither tracks packet ids nor exposes the DUP flag:
a QoS 1 PUBLISH is at-least-once (§4.3.2, a repeated id after the PUBACK is a new
publication whatever DUP says), and a QoS 2 one is delivered once per packet id until its
PUBREL (§4.3.3) — see DpExchange.Webull.Socket's "QoS 2 is delivered once".
No SUBSCRIBE, and this was checked rather than assumed
An earlier version of this module encoded SUBSCRIBE (8) and decoded SUBACK (9),
because a hand-rolled MQTT client reaches for them by reflex — that is how the protocol
is usually used. This venue's own documentation says otherwise:
docs/reference/webull/streaming-api.md states plainly that "subscriptions are not
managed over MQTT" and are HTTP calls instead — see DpExchange.Webull.Subscription.
Confirmed by tracing every call site: nothing in this package ever sent a SUBSCRIBE,
the venue delivers on Socket's connection regardless, and the socket carries live
quotes in the fake and in every test that exercises it without one ever going out.
So the encoder was dead code for a mechanism this venue does not use, not a mechanism
this package forgot to call — deleting it removes nothing a consumer could reach, and
keeping it would have kept inviting a future reader to wire it in by reflex, against a
venue that would answer a SUBSCRIBE with nothing at all (this venue publishes no
documented SUBACK failure mode for it either, because the endpoint that matters here
is the HTTP one).
Frame boundaries are not packet boundaries
decode/1 returns {:ok, packet, rest} so a caller holding a byte buffer can loop. A
WebSocket frame may carry several MQTT packets or half of one, and assuming otherwise is
a classic way to lose every packet after the first coalesced frame.
{:error, :incomplete} means "need more bytes" and is an ordinary state, not a failure.
Remaining length
MQTT encodes variable-header-plus-payload size as a base-128 varint of at most four bytes. The encoder handles the full range because a truncated length silently corrupts every subsequent packet in the stream rather than failing where the mistake was made.
Summary
Functions
The acknowledgement for an inbound QoS 1 or 2 packet: :puback (QoS 1 PUBLISH),
:pubrec (QoS 2 PUBLISH) or :pubcomp (the PUBREL that follows a PUBREC).
CONNECT with a clean session.
Decodes one inbound packet from the front of a buffer.
DISCONNECT — a clean, client-initiated end to the session.
PINGREQ — the keep-alive probe; the broker answers with PINGRESP.
Types
@type packet() :: {:connack, non_neg_integer()} | {:publish, binary(), binary()} | {:publish, binary(), binary(), {:puback | :pubrec, 0..65535}} | {:pubrel, 0..65535} | :pingresp | {:unhandled, non_neg_integer(), binary()}
A decoded inbound packet. A QoS 1 or 2 PUBLISH carries the acknowledgement it is owed as
a fourth element, to be turned into bytes with ack/2.
Functions
@spec ack(:puback | :pubrec | :pubcomp, 0..65535) :: binary()
The acknowledgement for an inbound QoS 1 or 2 packet: :puback (QoS 1 PUBLISH),
:pubrec (QoS 2 PUBLISH) or :pubcomp (the PUBREL that follows a PUBREC).
MQTT 3.1.1 §3.4, §3.5, §3.7: four bytes, the fixed header with zero flags, a remaining length of 2, and the packet identifier being acknowledged. See the moduledoc for why a package that never chose a QoS must still be able to answer one.
CONNECT with a clean session.
client_id is Webull's session_id — the same value the HTTP subscribe call names,
which is the only thing joining the two transports. username is the App Key. The
password is documented as "any value" and is not validated by this broker, so it is sent
to satisfy the flag rather than to authenticate; authorisation happens on the signed
HTTP subscribe.
clean_session is set: there is no durable subscription state worth resuming, and a
resumed session would replay stale queued prices on reconnect. The venue does not restore
subscriptions across a reconnect anyway.
@spec decode(binary()) :: {:ok, packet(), binary()} | {:error, :incomplete} | {:error, :malformed_length} | {:error, :malformed_header}
Decodes one inbound packet from the front of a buffer.
Returns {:ok, packet, rest}, {:error, :incomplete} when more bytes are needed, or
{:error, :malformed_length} / {:error, :malformed_header} when the stream cannot be
resynchronised.
A fixed header is checked the moment its length is known, before waiting for its
body. MQTT 3.1.1 §2.2 fixes the flags, and for most types the exact length, of every
packet a server may send, and says a receiver that gets invalid ones MUST close the
connection. This decoder used to accept any header. So a corrupted byte that turned the
front of the stream into, say, a PINGRESP declaring a 26-byte body made it swallow the
next 26 bytes of real packets as that body. The stream was then out of step with nothing
to say so. A byte-level fuzz (2026-09-26) found 99 of 1,045 mutated packets left it
silently wedged that way. See valid_header?/3.
@spec disconnect() :: binary()
DISCONNECT — a clean, client-initiated end to the session.
Sent by DpExchange.Webull.Socket.disconnect/2, which Feed calls on every connected
shard while it is shutting down cleanly (see Feed's own terminate/2). MQTT 3.1.1
§3.14 makes DISCONNECT the protocol's normal-close signal: a client that sends it and
then closes the network connection told the broker this was intentional; a client that
just drops the TCP connection did not, and a compliant broker treats the two
differently. Whichever timing rule this venue applies to the ~1-minute session
retention it documents is not stated for either case, so nothing here claims the
retention window is shorter for one than the other — the reason to send DISCONNECT is
that it is what a clean shutdown means on the wire, not a measured venue effect.
@spec pingreq() :: binary()
PINGREQ — the keep-alive probe; the broker answers with PINGRESP.