DpExchange.Webull.Socket (DpExchangeWebull v0.2.15)

Copy Markdown View Source

The MQTT-over-WebSocket connection — internal, never named above the facade.

websockex carries the frames; MqttPacket provides the framing; QuoteProto decodes the payloads. None of those three names can appear in anything a consumer receives.

The buffer is the point

A WebSocket frame is not an MQTT packet. One frame may carry several packets, or half of one, and the broker is under no obligation to align them. So every inbound frame is appended to a buffer and the buffer is drained packet by packet until it returns :incomplete.

Assuming frame boundaries are packet boundaries loses every packet after the first coalesced frame — quietly, because the socket stays up and the first message parses fine.

A :malformed_length means the stream can no longer be resynchronised. The buffer is dropped and a notice raised rather than retaining bytes that can never parse: keeping them would leave a live socket delivering nothing, which is the failure mode this family ranks worst.

Keep-alive

The venue's CONNECT carries a keep-alive interval, and MQTT requires the client to send something within it. PINGREQ goes out at half that interval — early enough that one lost ping is not a disconnect.

What it does not do

It does not subscribe. Subscriptions on this venue are HTTP calls, made by Feed with the same session_id this connection registered as its MQTT client id. The socket's only job is to connect, stay connected, and turn payloads into Core.Types.Quote (the snapshot topic), Core.Types.TopOfBook (the quote topic) or Core.Types.Trade (the tick topic) — see emit/3 below.

tick carries no trade id, and Trade.id is nil rather than invented

The venue's Tick message (docs/reference/webull/streaming-api.md) is Basic, time, price, volume, side — nothing identifies one print from the next. Core.Types.Trade requires :id, so this builds the struct literally (%Trade{id: nil, ...}) rather than through Trade.new/1, the same way Rest.get_trades/3's to_trade/2 already does for this venue's REST tape, which has the identical gap and says so in its own comment. Trade.new/1's validation exists to catch an accidentally-absent required field; this absence is not accidental, so going around it here is not going around the check — it is the one place a real, checked absence is allowed to be nil instead of failing closed.

Ending a session cleanly

It does not close itself, either — Feed decides when a shard's session ends, and disconnect/2 is how it says so on the wire before that shard's process goes down. See MqttPacket.disconnect/0 for why a clean DISCONNECT matters and Feed's own terminate/2 for where this is actually called.

Summary

Functions

connection_opts(opts)

The connection options handed to WebSockex.start_link/4.

disconnect(pid, timeout \\ 500)

Sends a clean MQTT DISCONNECT on an already-open socket, synchronously.

start_link(opts)

Functions

connection_opts(opts)

@spec connection_opts(keyword()) :: keyword()

The connection options handed to WebSockex.start_link/4.

Exposed so the deliberate timeouts can be asserted without opening a real socket — a later refactor must not be able to drop them back to the dependency's defaults unnoticed.

disconnect(pid, timeout \\ 500)

@spec disconnect(pid(), timeout()) :: :ok | {:error, term()}

Sends a clean MQTT DISCONNECT on an already-open socket, synchronously.

WebSockex.send_frame/3 is a :gen.call against the socket process itself — the only way to put a frame on an already-running Socket from outside its own callbacks, since MqttPacket's framing stays private to this module (Feed must not learn it — see the moduledoc's boundary). Called by Feed's own terminate/2, once per shard still connected when this package is shutting down cleanly.

Best-effort and never raises: a shard whose socket has already gone — crashed, already reconnecting, already torn down by the time shutdown reaches it — must not block or crash the shutdown asking for this. {:error, reason} says so; there is nothing a caller mid-shutdown can usefully do with it beyond logging, which Feed does.

The pid == self() guard exists because this runs from inside Feed's own terminate/2: a call this deep can legitimately end up with pid being the calling process itself only through a test fixture, never in production (a shard's socket is always a distinct Socket.start_link/1 process) — but WebSockex.send_frame/3 answers that specific case by raising WebSockex.CallingSelfError rather than returning an error, which the catch below alone would not stop. Checked first so this function's own "never raises" holds regardless.

start_link(opts)

@spec start_link(keyword()) :: {:ok, pid()} | {:error, term()}