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).
Inbound: CONNACK (2), PUBLISH (3), PINGRESP (13).
QoS 0 throughout, which is why PUBACK/PUBREC/PUBREL/PUBCOMP are absent: a
dropped price tick is replaced by the next one. Higher QoS would buy redelivery of stale
prices at the cost of broker-side queueing — the wrong trade for a feed where only the
newest value matters.
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
CONNECT with a clean session.
Decodes one inbound packet from the front of a buffer.
DISCONNECT — a clean, client-initiated end to the session.
Functions
@spec connect(String.t(), String.t(), String.t(), pos_integer()) :: binary()
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, term(), binary()} | {:error, :incomplete} | {:error, :malformed_length}
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} when the stream cannot be resynchronised.
@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()