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 eight 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), SUBSCRIBE (8), PINGREQ (12), DISCONNECT (14).
Inbound: CONNACK (2), SUBACK (9), 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.
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.
SUBSCRIBE to topics at QoS 0.
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()
@spec pingreq() :: binary()
SUBSCRIBE to topics at QoS 0.
packet_id must be non-zero — MQTT reserves 0 — and is echoed in the SUBACK so a
caller can match them.