DpExchange.Webull.Vendor.WebSockex behaviour (DpExchangeWebull v0.4.52)

Copy Markdown View Source

A vendored, patched fork of websockex 0.5.1, private to this package — never aliased or referenced as WebSockex from outside DpExchange.Webull, and not part of this package's own public API.

Why this exists — dp-exchange-core issue #27

Webull closes its MQTT-over-WebSocket connection with a close frame carrying prose where RFC 6455 §5.5.1 requires a 2-byte status code:

<<136, 10, 98, 121, 101, 45, 98, 121, 101, 33, 33, 33>>   # FIN+close, len 10, "bye-bye!!!"

The first two bytes parse as "by" = 25209, outside every valid close-code range (1000–1015, 3000–4999). WebSockex.Frame.parse_frame/1 correctly returns {:error, %WebSockex.FrameError{reason: :invalid_close_code, ...}} for this — see deps/websockex/lib/websockex/frame.ex:118, unmodified and still depended on directly (see below).

The defect is one level up: websocket_loop/3 below matched only {:ok, frame, buffer} and :incomplete, even though WebSockex.Frame.parse_frame/1's own @spec names the error tuple as a real return value. A frame that hits it raised CaseClauseError before any callback ranDpExchange.Webull.Socket's handle_disconnect/2 never fired, so the process died a hard, uncaught crash instead of the ordinary, supervised reconnect every other disconnect reason gets.

Measured impact, live: 117 crashes in 7 minutes across three shards — about one close every 10 seconds per shard, sustained. Supervision (DpExchange.Webull.Feed's isolate_crashed_shard/3) reopened each shard, so nothing alarmed; the venue simply never reached full coverage.

Confirmed against dominicletz/websockex (the current upstream; Azolo/websockex is abandoned) at its unreleased master on 2026-09-08: the same unguarded case is still there. websockex 0.5.1 is the latest Hex release. There is no upstream fix to wait for, and an upstream PR (opened alongside this fix) cannot be the fix that ships today — a malformed close frame is a protocol violation by the peer; RFC 6455 §7.1.7's specified response is to close the connection, not to crash the process reading it.

What changed from upstream 0.5.1

Exactly two things, both marked ## VENDORED FIX at their call site below:

  1. websocket_loop/3 gains a clause for {:error, %WebSockex.FrameError{}}, routed through the already-existing handle_close({:error, reason}, ...) path that sync_send/5 already used for a send-time framing error. That path (handle_error_close/4) sends this side's own close frame, waits for the socket to actually close, and then runs the ordinary disconnect flow — handle_disconnect/2 fires, {:reconnect, state} reopens the connection, exactly as any other {:remote, ...} close already does. No new recovery logic was written; the crash was the only thing standing between this frame shape and infrastructure that already existed.
  2. do_spawn/2's two clauses (inlined here from the real dependency's WebSockex.Utils.spawn/5, hidden from its own docs, so not linked here — which hardcodes the literal atom WebSockex as the :proc_lib entry module — see "What stayed a dependency, and why" below) use __MODULE__ instead, so a process started via this vendored copy's start_link/4 actually enters this module's init/5,6, not the real, unpatched one.

Every function body and every @spec is copied verbatim from deps/websockex/lib/websockex.ex at 0.5.1 — diff against that file to see the fix's true size: two clauses, plus the spawn_process/5/do_spawn/2 glue above. A handful of doc comments were additionally edited, never behaviour: the two self-referential examples in this moduledoc's closing section (use WebSockexuse DpExchange.Webull.Vendor.WebSockex), the two @doc headers for start/4 and start_link/4, and a few backtick spans de-linked to plain text where this package's own mix docs flagged a reference into the real dependency's own hidden/undocumented internals (WebSockex.Utils, WebSockex.ConnError — both carry @moduledoc false upstream, plain text here rather than another broken link) as an unresolvable warning. None of those changes the meaning of the original text, only whether ExDoc treats it as a link.

What stayed a dependency, and why

mix.exs still pins {:websockex, "== 0.5.1"} exactly, not ~>. This module still calls WebSockex.Frame, WebSockex.Conn, WebSockex.Utils (hidden from its own docs, so plain text here rather than a broken link), WebSockex.Application (started automatically as part of the :websockex OTP app) and every WebSockex.*Error struct from that real, unmodified dependency — none of those had the bug, vendoring them would have tripled this diff for no safety gained, and WebSockex.Frame/WebSockex.Conn are pure parsing/connection-building code with no process identity to collide with anything else in a consumer's dependency tree.

The exact pin (not ~>) is deliberate: this module calls into WebSockex.Conn's and WebSockex.Frame's functions the same way the original websockex.ex did — a private-in-spirit internal API those modules never promised to keep stable across versions the way their own public behaviour is. A ~> floor invites exactly the failure this family already had once with this same dependency (see mix.exs's own long comment on the send_frame/3 arity floor) — a version the range permits silently changing an internal shape this file was written against.

Only websockex.ex itself needed vendoring. WebSockex.Utils.spawn/5 (hidden from its own docs, so plain text here) is the one place outside that file that hardcodes the literal atom WebSockex (:proc_lib.start_link( WebSockex, :init, args)), which would have booted every vendored socket into the original, unpatched module's init/5,6 regardless of this rename. Rather than vendoring Utils too, spawn_process/5 and do_spawn/2 below reimplement that one small function locally, using __MODULE__ — the only change needed to close that gap.

No name collision, by construction

This module is named DpExchange.Webull.Vendor.WebSockex, never bare WebSockex — a consumer that also depends on the real :websockex package (directly, or through another venue in this family) gets that package's actual WebSockex, WebSockex.Conn, WebSockex.Frame, etc. unchanged. This module is compiled alongside them, under a private namespace, calling into their real, un-shadowed code. Nothing here is WebSockex. from the outside, and nothing outside DpExchange.Webull may reference this module by any other name than the fully-qualified one.

Re-syncing with upstream

If websockex publishes 0.5.2+ (or dominicletz/websockex merges this family's upstream PR), diff that release's lib/websockex.ex against this file. If it fixes this exact defect, drop this vendor module, restore use WebSockex / WebSockex.start_link / WebSockex.send_frame in DpExchange.Webull.Socket, and relax mix.exs's pin back to ~>. Until then, every future upstream release should be diffed here for anything beyond this file's own two marked changes.


The following is the original websockex moduledoc, unmodified except for the two self-referential examples below, which now name this vendored module.

A client handles negotiating the connection, then sending frames, receiving frames, closing, and reconnecting that connection.

A simple client implementation would be:

defmodule WsClient do
  use DpExchange.Webull.Vendor.WebSockex

  def start_link(url, state) do
    DpExchange.Webull.Vendor.WebSockex.start_link(url, __MODULE__, state)
  end

  def handle_frame({:text, msg}, state) do
    IO.puts "Received a message: #{msg}"
    {:ok, state}
  end

  def handle_cast({:send, {type, msg} = frame}, state) do
    IO.puts "Sending #{type} frame with payload: #{msg}"
    {:reply, frame, state}
  end
end

Closing Connections

WebSockex connections can be closed gracefully by returning close tuples from callback functions. The following callbacks support close returns:

Basic Close

Return {:close, state} to close with the default close code (1000):

def handle_frame({:text, "quit"}, state) do
  {:close, state}
end

Close with Custom Code

Return {:close, {close_code, message}, state} to specify a close code and reason:

def handle_frame({:text, "error"}, state) do
  {:close, {4000, "Application error"}, state}
end

Close codes are integers in specific ranges:

  • 1000-1015 - Standard protocol codes (e.g., 1000 = normal, 1001 = going away, 1002 = protocol error, 1003 = unsupported data)
  • 3000-3999 - Reserved for use by libraries, frameworks, and applications (registered with IANA)
  • 4000-4999 - Private use for applications

Common standard codes include:

  • 1000 - Normal closure
  • 1001 - Going away
  • 1002 - Protocol error
  • 1003 - Unsupported data

Supervision

WebSockex is implemented as an OTP Special Process and as a result will fit into supervision trees.

WebSockex also supports the Supervisor children format introduced in Elixir 1.5. Meaning that a child specification could be {ClientModule, [state]}.

However, since there is a possibility that you would like to provide a WebSockex.Conn (plain text: the original upstream doc names this without the .t suffix its own type actually needs, which is why it does not resolve as a link either way) or a url as well as the state, there are two versions of the child_spec function. If you need functionality beyond that it is recommended that you override the function or define your own.

Just remember to use the version that corresponds with your start_link's arity.

Summary

Types

An integer between 1000 and 4999 that specifies the reason for closing the connection.

The error returned when a connection fails to be established.

The frame sent when the negotiating a connection closure.

The reason a connection was closed.

A map that contains information about the failure to connect.

Debug options to be parsed by :sys.debug_options/1.

Options values for start_link.

Callbacks

Invoked when a new version the module is loaded during runtime.

Invoked to retrieve a formatted status of the state in a WebSockex process.

Invoked to handle asynchronous cast/2 messages.

Invoked after a connection is established.

Invoked when the WebSocket disconnects from the server.

Invoked on the reception of a frame on the socket.

Invoked to handle all other non-WebSocket messages.

Invoked when the Websocket receives a ping frame

Invoked when the Websocket receives a pong frame.

Invoked when the process is terminating.

Types

client()

@type client() :: pid() | atom() | {:via, module(), term()} | {:global, term()}

close_code()

@type close_code() :: integer()

An integer between 1000 and 4999 that specifies the reason for closing the connection.

close_error()

@type close_error() ::
  %WebSockex.RequestError{__exception__: true, code: term(), message: term()}
  | %WebSockex.ConnError{__exception__: true, original: term()}
  | %WebSockex.InvalidFrameError{__exception__: true, frame: term()}
  | %WebSockex.FrameEncodeError{
      __exception__: true,
      close_code: term(),
      frame_payload: term(),
      frame_type: term(),
      reason: term()
    }

The error returned when a connection fails to be established.

close_frame()

@type close_frame() :: {close_code(), message :: binary()}

The frame sent when the negotiating a connection closure.

close_reason()

@type close_reason() ::
  {:remote | :local, :normal}
  | {:remote | :local, close_code(), message :: binary()}
  | {:remote, :closed}
  | {:error, term()}

The reason a connection was closed.

A :normal reason is the same as a 1000 reason with no payload.

If the peer closes the connection abruptly without a close frame then the close reason is {:remote, :closed}.

connection_status_map()

@type connection_status_map() :: %{
  reason: close_reason() | close_error(),
  attempt_number: integer(),
  conn: WebSockex.Conn.t()
}

A map that contains information about the failure to connect.

This map contains the error, attempt number, and the WebSockex.Conn.t/0 that was used to attempt the connection.

debug_opts()

@type debug_opts() :: [
  :trace
  | :log
  | {:log, log_depth :: pos_integer()}
  | :statistics
  | {:log_to_file, Path.t()}
]

Debug options to be parsed by :sys.debug_options/1.

These options can also be set after the process is running using the functions in the Erlang :sys module.

frame()

@type frame() ::
  :ping
  | :pong
  | {:ping | :pong, nil | (message :: binary())}
  | {:text | :binary, message :: binary()}

option()

@type option() ::
  WebSockex.Conn.connection_option()
  | {:async, boolean()}
  | {:debug, debug_opts()}
  | {:name, atom() | {:global, term()} | {:via, module(), term()}}
  | {:handle_initial_conn_failure, boolean()}

Options values for start_link.

  • :async - Replies with {:ok, pid} before establishing the connection. This is useful for when attempting to connect indefinitely, this way the process doesn't block trying to establish a connection.
  • :handle_initial_conn_failure - When set to true a connection failure while establishing the initial connection won't immediately return an error and instead will invoke the handle_disconnect/2 callback. This option only matters during process initialization. The handle_disconnect callback is always invoked if an established connection is lost.
  • :debug - Options to set the debug options for :sys.handle_debug.
  • :name - An atom that the registers the process with name locally. Can also be a {:via, module, term} or {:global, term} tuple.

Other possible option values include: WebSockex.Conn.connection_option/0

options()

@type options() :: [option()]

Callbacks

code_change(old_vsn, state, extra)

@callback code_change(
  old_vsn :: term() | {:down, term()},
  state :: term(),
  extra :: term()
) ::
  {:ok, new_state :: term()} | {:error, reason :: term()}

Invoked when a new version the module is loaded during runtime.

format_status(atom, list)

(optional)
@callback format_status(:normal, [process_dictionary | state]) :: status :: term()
when process_dictionary: [{key :: term(), val :: term()}], state: term()

Invoked to retrieve a formatted status of the state in a WebSockex process.

This optional callback is used when you want to edit the values returned when invoking :sys.get_status.

The second argument is a two-element list with the order of [pdict, state].

handle_cast(msg, state)

@callback handle_cast(msg :: term(), state :: term()) ::
  {:ok, new_state}
  | {:reply, frame(), new_state}
  | {:close, new_state}
  | {:close, close_frame(), new_state}
when new_state: term()

Invoked to handle asynchronous cast/2 messages.

handle_connect(conn, state)

@callback handle_connect(conn :: WebSockex.Conn.t(), state :: term()) ::
  {:ok, new_state :: term()}

Invoked after a connection is established.

This is invoked after both the initial connection and a reconnect.

handle_disconnect(connection_status_map, state)

@callback handle_disconnect(connection_status_map(), state :: term()) ::
  {:ok, new_state}
  | {:reconnect, new_state}
  | {:reconnect, new_conn :: WebSockex.Conn.t(), new_state}
when new_state: term()

Invoked when the WebSocket disconnects from the server.

This callback is only invoked in the event of a connection failure. In cases of crashes or other errors the process will terminate immediately skipping this callback.

If the handle_initial_conn_failure: true option is provided during process startup, then this callback will be invoked if the process fails to establish an initial connection.

If a connection is established by reconnecting, the handle_connect/2 callback will be invoked.

The possible returns for this callback are:

  • {:ok, state} will continue the process termination.
  • {:reconnect, state} will attempt to reconnect instead of terminating.
  • {:reconnect, conn, state} will attempt to reconnect with the connection data in conn. conn is expected to be a WebSockex.Conn.t/0.

handle_frame(frame, state)

@callback handle_frame(frame(), state :: term()) ::
  {:ok, new_state}
  | {:reply, frame(), new_state}
  | {:close, new_state}
  | {:close, close_frame(), new_state}
when new_state: term()

Invoked on the reception of a frame on the socket.

The control frames have possible payloads, when they don't have a payload then the frame will have nil as the payload. e.g. {:ping, nil}

handle_info(msg, state)

@callback handle_info(msg :: term(), state :: term()) ::
  {:ok, new_state}
  | {:reply, frame(), new_state}
  | {:close, new_state}
  | {:close, close_frame(), new_state}
when new_state: term()

Invoked to handle all other non-WebSocket messages.

handle_ping(ping_frame, state)

@callback handle_ping(ping_frame :: :ping | {:ping, binary()}, state :: term()) ::
  {:ok, new_state}
  | {:reply, frame(), new_state}
  | {:close, new_state}
  | {:close, close_frame(), new_state}
when new_state: term()

Invoked when the Websocket receives a ping frame

handle_pong(pong_frame, state)

@callback handle_pong(pong_frame :: :pong | {:pong, binary()}, state :: term()) ::
  {:ok, new_state}
  | {:reply, frame(), new_state}
  | {:close, new_state}
  | {:close, close_frame(), new_state}
when new_state: term()

Invoked when the Websocket receives a pong frame.

terminate(close_reason, state)

@callback terminate(close_reason(), state :: term()) :: any()

Invoked when the process is terminating.

Functions

cast(client, message)

@spec cast(client(), term()) :: :ok

Asynchronously sends a message to a client that is handled by handle_cast/2.

handle_terminate_close(reason, parent, debug, state)

@spec handle_terminate_close(any(), pid(), any(), any()) :: no_return()

init(parent, name, conn, module, module_state, opts)

@spec init(pid(), atom(), WebSockex.Conn.t(), module(), term(), options()) ::
  {:ok, pid()} | {:error, term()}

send_frame(client, frame, timeout \\ 5000)

@spec send_frame(client(), frame(), timeout()) ::
  :ok
  | {:error,
     %WebSockex.FrameEncodeError{
       __exception__: true,
       close_code: term(),
       frame_payload: term(),
       frame_type: term(),
       reason: term()
     }
     | %WebSockex.ConnError{__exception__: true, original: term()}
     | %WebSockex.NotConnectedError{
         __exception__: true,
         connection_state: term()
       }
     | %WebSockex.InvalidFrameError{__exception__: true, frame: term()}}
  | none()

Sends a frame through the WebSocket.

If the connection is either connecting or closing then this will return an error tuple with a WebSockex.NotConnectedError exception struct as the second element.

If a connection failure is discovered while sending then it will return an error tuple with a WebSockex.ConnError (hidden from its own docs, so plain text here) exception struct as the second element.

start(conn_info, module, state, opts \\ [])

@spec start(url :: String.t() | WebSockex.Conn.t(), module(), term(), options()) ::
  {:ok, pid()} | {:error, term()}

Starts a DpExchange.Webull.Vendor.WebSockex process.

Acts like start_link/4, except doesn't link the current process.

See start_link/4 for more information.

start_link(conn_info, module, state, opts \\ [])

@spec start_link(url :: String.t() | WebSockex.Conn.t(), module(), term(), options()) ::
  {:ok, pid()} | {:error, term()}

Starts a DpExchange.Webull.Vendor.WebSockex process linked to the current process.

For available option values see option/0.

If a WebSockex.Conn.t is used in place of a url string, then the options available in WebSockex.Conn.connection_option/0 have effect.

The callback handle_connect/2 is invoked after the connection is established.