OffBroadwayWebSocket

https://hexdocs.pm/off_broadway_websocket/
Admin
An Elixir library providing an Off-Broadway producer for resilient WebSocket ingestion with...
Tokens:8,466
Snippets:92
Trust Score:9
Update:6 months ago
Show doc for...
Tokens:
Raw
### Minimal Broadway Setup with OffBroadwayWebSocket Producer

Source: https://hexdocs.pm/off_broadway_websocket/getting_started.md

This example demonstrates a minimal Broadway setup using `OffBroadwayWebSocket.Producer`. Configure the `url`, `path`, and `telemetry_id` for your websocket stream. The `transform` function is used to wrap the payload in a Broadway `Message` with a `NoopAcknowledger`.

```elixir
defmodule MyApp.Broadway do
  use Broadway

  alias Broadway.Message
  alias Broadway.NoopAcknowledger

  def start_link(_opts) do
    Broadway.start_link(__MODULE__,
      name: __MODULE__,
      producer: [
        module: {
          OffBroadwayWebSocket.Producer,
          url: "wss://example.com",
          path: "/stream",
          telemetry_id: :example_ws
        },
        transformer: {__MODULE__, :transform, []},
        concurrency: 1
      ],
      processors: [
        default: [min_demand: 0, max_demand: 100, concurrency: 4]
      ]
    )
  end

  def handle_message(_stage, message, _context), do: message

  def transform(payload, _opts) do
    %Message{
      data: payload,
      acknowledger: NoopAcknowledger.init()
    }
  end
end
```

--------------------------------

### Start the WebSocket producer

Source: https://hexdocs.pm/off_broadway_websocket/OffBroadwayWebSocket.Producer.md

Starts the producer process under a GenStage supervisor using the provided configuration options.

```elixir
@spec start_link(keyword()) :: GenServer.on_start()
```

--------------------------------

### Quickstart Broadway Producer Configuration

Source: https://hexdocs.pm/off_broadway_websocket/index.html

Configure a Broadway producer using OffBroadwayWebSocket for resilient WebSocket ingestion. This example includes URL, path, timeouts, telemetry, TLS options, and frame handling.

```elixir
defmodule MyApp.Broadway do
  use Broadway

  alias Broadway.Message
  alias Broadway.NoopAcknowledger

  def start_link(_args) do
    Broadway.start_link(__MODULE__,
      name: __MODULE__,
      producer: [
        module: {
          OffBroadwayWebSocket.Producer,
          url: "wss://example.com:443",
          path: "/stream/trades",
          ws_timeout: 15_000,
          telemetry_id: :my_app_ws,
          gun_opts: %{
            transport: :tls,
            protocols: [:http],
            tls_opts: [
              verify: :verify_peer,
              cacertfile: CAStore.file_path(),
              verify_fun: {
                 പാssl_verify_hostname.verify_fun/3,
                [check_hostname: String.to_charlist("example.com")]
              }
            ],
            http_opts: %{version: :"HTTP/1.1"},
            ws_opts: %{keepalive: 10_000, silence_pings: false}
          },
          on_upgrade: {MyApp.WebSocket, :subscription_frames, [[]]},
          frame_handler: {MyApp.WebSocket, :handle_frame, []},
          frame_handler_state: %{subscriptions: %{}}
        },
        transformer: {__MODULE__, :transform, []},
        concurrency: 1
      ],
      processors: [default: [min_demand: 0, max_demand: 100, concurrency: 4]]
    )
  end

  def handle_message(_stage, %Message{data: payload} = message, _context) do
    message
    |> Map.put(:data, payload)
  end

  def transform(payload, _opts) do
    %Broadway.Message{
      data: payload,
      acknowledger: NoopAcknowledger.init()
    }
  end
end
```

--------------------------------

### Configuring Gun Options for OffBroadwayWebSocket

Source: https://hexdocs.pm/off_broadway_websocket/configuration.md

Example of how to configure `gun_opts` to customize transport, TLS verification, HTTP version, and websocket keepalive behavior. Use this to fine-tune connection and communication settings.

```elixir
gun_opts: %{
  connect_timeout: 5_000,
  protocols: [:http],
  transport: :tls,
  tls_opts: [
    verify: :verify_peer,
    cacertfile: CAStore.file_path()
  ],
  ws_opts: %{
    keepalive: 10_000,
    silence_pings: false
  },
  http_opts: %{
    version: :"HTTP/1.1"
  }
}
```

--------------------------------

### POST /connect

Source: https://hexdocs.pm/off_broadway_websocket/OffBroadwayWebSocket.Client.md

Establishes a WebSocket connection using the :gun library with customizable options.

```APIDOC
## POST /connect

### Description
Establishes a WebSocket connection using `:gun` with configurable options and timeouts.

### Method
POST

### Endpoint
/connect

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
- **url** (string) - Required - The base URL for the WebSocket connection (e.g., "wss://example.com").
- **path** (string) - Required - The WebSocket path to upgrade to (e.g., "/ws").
- **gun_opts** (map) - Required - Options passed to `:gun.open/3` for TLS, HTTP, and WebSocket configuration.
- **await_timeout** (non_neg_integer) - Required - How long to wait for the connection to become ready in milliseconds.
- **headers** (list of tuples) - Optional - Additional headers for the WebSocket upgrade.

### Request Example
```json
{
  "url": "wss://example.com",
  "path": "/ws",
  "gun_opts": {}, 
  "await_timeout": 5000,
  "headers": [["Authorization", "Bearer YOUR_TOKEN"]]
}
```

### Response
#### Success Response (200)
- **conn_pid** (pid) - The process ID of the established connection.
- **stream_ref** (reference) - The reference to the WebSocket stream.

#### Response Example
```json
{
  "conn_pid": "<0.123.456>",
  "stream_ref": "#Ref<0.0.0.123>"
}
```

#### Error Response
- **error** (term) - The reason for the connection or upgrade failure.
```

--------------------------------

### OffBroadwayWebSocket.Producer Configuration Options

Source: https://hexdocs.pm/off_broadway_websocket/configuration.md

Configuration options for initializing the OffBroadwayWebSocket producer.

```APIDOC
## OffBroadwayWebSocket.Producer Configuration

### Description
Configuration options for the OffBroadwayWebSocket producer, used to define connection parameters, retry logic, and lifecycle hooks.

### Required Parameters
- **:url** (string) - Required - Websocket base URL
- **:path** (string) - Required - Request path and query string

### Optional Parameters
- **:ws_timeout** (integer) - Optional - Idle timeout in milliseconds
- **:await_timeout** (integer) - Optional - Timeout for :gun.await_up/2
- **:headers** (list) - Optional - Websocket upgrade headers
- **:headers_fn** (function) - Optional - Zero-arity callback for per-connect header refresh
- **:telemetry_id** (string) - Optional - Prefix for telemetry events
- **:gun_opts** (map) - Optional - Options forwarded to :gun
- **:min_demand** (integer) - Optional - Broadway producer min demand
- **:max_demand** (integer) - Optional - Broadway producer max demand
- **:ws_retry_opts** (map) - Optional - Retry state map
- **:ws_retry_fun** (function) - Optional - Function to compute next retry state
- **:on_upgrade** (MFA) - Optional - MFA run after websocket upgrade
- **:frame_handler** (MFA) - Optional - MFA run for normalized inbound data frames
- **:frame_handler_state** (any) - Optional - Initial connection-local state for :frame_handler
```

--------------------------------

### On Upgrade Bootstrap Configuration

Source: https://hexdocs.pm/off_broadway_websocket/on_upgrade_bootstrap.md

Configuration and contract details for the :on_upgrade callback.

```APIDOC
## :on_upgrade Callback

### Description
Defines a callback to be executed after the websocket upgrade completes, allowing for outbound subscribe or bootstrap frames.

### Contract
Pass an MFA tuple:
`on_upgrade: {Module, :function, [args]}`

### Return Values
- `{:ok, []}`: Success with no frames to send.
- `{:ok, [{:text | :binary, iodata()}, ...]}`: Success with a list of frames to send.
- `{:error, reason}`: Failure to bootstrap.

### Semantics
- The callback runs after the websocket upgrade.
- Returned frames are sent in order using `:gun.ws_send/3`.
- The producer is not considered ready until bootstrap succeeds.
- Connection-success telemetry is emitted only after bootstrap succeeds.

### Failure Behavior
Bootstrap failures occur if:
- The callback returns `{:error, reason}`.
- The callback returns an invalid result.
- An immediate `:gun.ws_send/3` failure occurs for any returned frame.
- The callback raises an exception.

Bootstrap failure triggers the standard reconnect/backoff path.
```

--------------------------------

### Fetch Dependencies

Source: https://hexdocs.pm/off_broadway_websocket/getting_started.md

Run this command in your terminal to fetch the project dependencies after updating `mix.exs`.

```bash
mix deps.get
```

--------------------------------

### Run Tests

Source: https://hexdocs.pm/off_broadway_websocket/index.html

Execute the project's tests using the mix test command.

```bash
mix test
```

--------------------------------

### Connect to WebSocket with Gun Options

Source: https://hexdocs.pm/off_broadway_websocket/OffBroadwayWebSocket.Client.md

Use this function to establish a WebSocket connection. It forwards gun options for TLS, HTTP, and WebSocket configuration. The connection attempt waits for a specified timeout before returning.

```elixir
def connect(
  url :: String.t(),
  path :: String.t(),
  gun_opts :: map(),
  await_timeout :: non_neg_integer(),
  headers :: [{String.t(), String.t()}]
) :: {:ok, %{conn_pid: pid(), stream_ref: reference()}} | {:error, term()}
```

--------------------------------

### Callback Return Values for :on_upgrade

Source: https://hexdocs.pm/off_broadway_websocket/on_upgrade_bootstrap.md

The callback function for :on_upgrade must return one of these valid structures: {:ok, []}, {:ok, [{:text | :binary, iodata()}, ...]}, or {:error, reason}.

```elixir
 {:ok, []}
{:ok, [{:text | :binary, iodata()}, ...]},
{:error, reason}
```

--------------------------------

### Define :on_upgrade Callback Contract

Source: https://hexdocs.pm/off_broadway_websocket/on_upgrade_bootstrap.md

Specify the contract for the :on_upgrade callback. It must return an :ok tuple with either an empty list, a list of frames, or an error reason.

```elixir
on_upgrade: {MyApp.Subscriptions, :bootstrap_frames, [args]}
```

--------------------------------

### POST /connect_once

Source: https://hexdocs.pm/off_broadway_websocket/OffBroadwayWebSocket.Client.md

Connects to a WebSocket endpoint once and returns the state.

```APIDOC
## POST /connect_once

### Description
Establishes a WebSocket connection once and returns the resulting state.

### Method
POST

### Endpoint
/connect_once

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
- **state** (OffBroadwayWebSocket.State.t()) - Required - The current state of the WebSocket connection.

### Request Example
```json
{
  "state": { ... } 
}
```

### Response
#### Success Response (200)
- **map** - The resulting state after connecting.

#### Response Example
```json
{
  "connected": true, 
  "conn_pid": "<0.123.456>",
  "stream_ref": "#Ref<0.0.0.123>"
}
```

#### Error Response
- **error** (term) - The reason for the connection failure.
```

--------------------------------

### OffBroadwayWebSocket.State.new/1

Source: https://hexdocs.pm/off_broadway_websocket/OffBroadwayWebSocket.State.md

Initializes a new state struct for the WebSocket producer with the provided configuration options.

```APIDOC
## new(options)

### Description
Builds a new state struct from the given options to configure the WebSocket producer.

### Parameters
#### Request Body
- **url** (String) - Required - WebSocket endpoint
- **path** (String) - Required - WebSocket path
- **gun_opts** (map) - Optional - Options passed directly to Gun (default: %{})
- **ws_timeout** (integer) - Optional - Idle timeout for WebSocket frames (default: nil)
- **await_timeout** (integer) - Optional - Timeout for synchronous Gun calls (default: 10000)
- **headers** (list) - Optional - HTTP headers for the upgrade (default: [])
- **headers_fn** (function) - Optional - Callback to refresh upgrade headers (default: nil)
- **on_upgrade** (MFA) - Optional - Callback invoked after upgrade (default: nil)
- **frame_handler** (MFA) - Optional - Callback for inbound data frames (default: nil)
- **frame_handler_state** (term) - Optional - Initial connection-local state (default: nil)
- **telemetry_id** (atom) - Optional - ID for telemetry events (default: :websocket_producer)
- **broadway** (keyword) - Optional - Customizes min_demand/max_demand
- **ws_retry_opts** (map) - Optional - Retry parameters (default: default_ws_retry_opts/0)
- **ws_retry_fun** (function) - Optional - Callback to calculate next retry delay

### Response
#### Success Response (200)
- **t** (struct) - Returns a new OffBroadwayWebSocket.State struct
```

--------------------------------

### OffBroadwayWebSocket.ClientBehaviour - connect callback

Source: https://hexdocs.pm/off_broadway_websocket/OffBroadwayWebSocket.ClientBehaviour.md

The `connect` callback is used to establish a WebSocket connection. It takes the URL, path, Gun options, a timeout, and headers as arguments, returning either a successful connection PID and stream reference or an error.

```APIDOC
## `connect` Callback

### Description
Establishes a WebSocket connection.

### Method
`@callback connect/5`

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Parameters
- **url** (String.t()) - The WebSocket server URL.
- **path** (String.t()) - The path for the WebSocket connection.
- **gun_opts** (map()) - Options for the underlying Gun connection.
- **await_timeout** (non_neg_integer()) - The timeout in milliseconds to wait for the connection.
- **headers** (list of {String.t(), String.t()}) - Custom headers to send with the connection request.

### Request Example
None provided.

### Response
#### Success Response (200)
- **conn_pid** (pid()) - The process ID of the established connection.
- **stream_ref** (reference()) - A reference to the WebSocket stream.

#### Response Example
```json
{
  "conn_pid": "<0.123.0>",
  "stream_ref": "#Ref<0.1234.5.6789>"
}
```

#### Error Response
- **error** (term()) - An error term if the connection fails.
```

--------------------------------

### On-Upgrade Bootstrap Callback

Source: https://hexdocs.pm/off_broadway_websocket/index.html

Defines the return values for an `:on_upgrade` callback, used for sending frames immediately after a WebSocket upgrade.

```elixir
 {:ok, []}
 {:ok, [{:text | :binary, iodata()}, ...]}}
 {:error, reason}
```

--------------------------------

### Frame Handler Configuration and Contract

Source: https://hexdocs.pm/off_broadway_websocket/frame_handler.md

Defines how to configure the frame handler and the expected return values for the callback function.

```APIDOC
## Frame Handler Configuration

### Description
Configures the websocket transport to use a custom frame handler for session-aware protocols.

### Parameters
- **frame_handler** (MFA) - Required - The module, function, and arguments to process frames.
- **frame_handler_state** (Map) - Optional - The initial state for the connection-local handler.

### Request Example
```elixir
frame_handler: {MyApp.Transport, :handle_frame, []},
frame_handler_state: %{subscriptions: %{}}
```

### Callback Return Values
- **{:emit, [payload, ...], new_state}** - Queues payloads downstream and updates state.
- **{:skip, new_state}** - Emits nothing downstream but updates state.
- **{:error, reason, new_state}** - Updates state and triggers connection failure/reconnect logic.
```

--------------------------------

### OffBroadwayWebSocket.Client

Source: https://hexdocs.pm/off_broadway_websocket/api-reference.md

Provides functions to establish and configure WebSocket connections using the gun library.

```APIDOC
## OffBroadwayWebSocket.Client

### Description
Provides functions to establish and configure WebSocket connections using the **gun** library, with customizable timeouts and connection options.

### Method
N/A (Module Documentation)

### Endpoint
N/A (Module Documentation)

### Parameters
N/A (Module Documentation)

### Request Example
N/A (Module Documentation)

### Response
N/A (Module Documentation)

### Response Example
N/A (Module Documentation)
```

--------------------------------

### Refreshing Headers on Each Connect

Source: https://hexdocs.pm/off_broadway_websocket/auth_refresh_and_handshake_failures.md

Use `:headers_fn` to provide authentication headers that are refreshed before each connection attempt, including reconnects. Ensure the function is fast and side-effect free.

```elixir
producer: [
  module: {
    OffBroadwayWebSocket.Producer,
    url: "wss://api.exchange.example",
    path: "/v1/stream",
    headers_fn: fn ->
      token = MyApp.Auth.refresh_ws_token!()
      [{"authorization", "Bearer " <> token}]
    end
  }
]
```

--------------------------------

### Client Behaviour Connect Callback

Source: https://hexdocs.pm/off_broadway_websocket/OffBroadwayWebSocket.ClientBehaviour.md

Defines the callback signature for the `connect` function within the client behaviour. It specifies the parameters required to establish a WebSocket connection and the expected return types for success or failure.

```elixir
@callback connect(
  url :: String.t(),
  path :: String.t(),
  gun_opts :: map(),
  await_timeout :: non_neg_integer(),
  headers :: [{String.t(), String.t()}]
) :: {:ok, %{conn_pid: pid(), stream_ref: reference()}} | {:error, term()}
```

--------------------------------

### Configure Frame Handler MFA and State

Source: https://hexdocs.pm/off_broadway_websocket/frame_handler.md

Define the module, function, and arguments for the frame handler along with the initial connection-local state.

```elixir
frame_handler: {MyApp.Transport, :handle_frame, []},
frame_handler_state: %{subscriptions: %{}}
```

--------------------------------

### Add off_broadway_websocket Dependency

Source: https://hexdocs.pm/off_broadway_websocket/getting_started.md

Add the `off_broadway_websocket` dependency to your `mix.exs` file to include it in your project.

```elixir
def deps do
  [
    {:off_broadway_websocket, "~> 1.2.1"}
  ]
end
```

--------------------------------

### OffBroadwayWebSocket.ClientBehaviour

Source: https://hexdocs.pm/off_broadway_websocket/api-reference.md

Behaviour describing the minimal client API used by the producer.

```APIDOC
## OffBroadwayWebSocket.ClientBehaviour

### Description
Behaviour describing the minimal client API used by the producer.

### Method
N/A (Module Documentation)

### Endpoint
N/A (Module Documentation)

### Parameters
N/A (Module Documentation)

### Request Example
N/A (Module Documentation)

### Response
N/A (Module Documentation)

### Response Example
N/A (Module Documentation)
```

--------------------------------

### Connect Once with State

Source: https://hexdocs.pm/off_broadway_websocket/OffBroadwayWebSocket.Client.md

This function establishes a single WebSocket connection based on the provided OffBroadwayWebSocket.State.t() configuration.

```elixir
def connect_once(OffBroadwayWebSocket.State.t()) :: {:ok, map()} | {:error, term()}
```

--------------------------------

### Define new state constructor specification

Source: https://hexdocs.pm/off_broadway_websocket/OffBroadwayWebSocket.State.md

Specifies the function signature for creating a new state struct from options.

```elixir
@spec new(keyword()) :: t()
```

--------------------------------

### Retry Configuration Functions

Source: https://hexdocs.pm/off_broadway_websocket/OffBroadwayWebSocket.State.md

Functions for managing WebSocket reconnection logic and backoff strategies.

```APIDOC
## default_ws_retry_fun(retry_opts)

### Description
Default function used for reconnect backoff. It decrements retries_left while keeping the delay constant.

## default_ws_retry_opts()

### Description
Returns the default retry options map used for WebSocket reconnects.

### Response
#### Success Response (200)
- **retry_opts** (map) - Returns a map containing retries_left, max_retries, and delay.
```

--------------------------------

### Implement Broadway pipeline with WebSocket producer

Source: https://hexdocs.pm/off_broadway_websocket/readme.md

Define a Broadway module using OffBroadwayWebSocket.Producer to handle WebSocket connections and message processing.

```elixir
defmodule MyApp.Broadway do
  use Broadway

  alias Broadway.Message
  alias Broadway.NoopAcknowledger

  def start_link(_args) do
    Broadway.start_link(__MODULE__,
      name: __MODULE__,
      producer: [
        module: {
          OffBroadwayWebSocket.Producer,
          url: "wss://example.com:443",
          path: "/stream/trades",
          ws_timeout: 15_000,
          telemetry_id: :my_app_ws,
          gun_opts: %{
            transport: :tls,
            protocols: [:http],
            tls_opts: [
              verify: :verify_peer,
              cacertfile: CAStore.file_path(),
              verify_fun: {
                &:ssl_verify_hostname.verify_fun/3,
                [check_hostname: String.to_charlist("example.com")]
              }
            ],
            http_opts: %{version: :"HTTP/1.1"},
            ws_opts: %{keepalive: 10_000, silence_pings: false}
          },
          on_upgrade: {MyApp.WebSocket, :subscription_frames, [[]]},
          frame_handler: {MyApp.WebSocket, :handle_frame, []},
          frame_handler_state: %{subscriptions: %{}}
        },
        transformer: {__MODULE__, :transform, []},
        concurrency: 1
      ],
      processors: [default: [min_demand: 0, max_demand: 100, concurrency: 4]]
    )
  end

  def handle_message(_stage, %Message{data: payload} = message, _context) do
    message
    |> Map.put(:data, payload)
  end

  def transform(payload, _opts) do
    %Broadway.Message{
      data: payload,
      acknowledger: NoopAcknowledger.init()
    }
  end
end
```

--------------------------------

### Define default_ws_retry_opts specification

Source: https://hexdocs.pm/off_broadway_websocket/OffBroadwayWebSocket.State.md

Specifies the function signature for retrieving default retry options.

```elixir
@spec default_ws_retry_opts() :: retry_opts()
```

--------------------------------

### Define retry_opts type

Source: https://hexdocs.pm/off_broadway_websocket/OffBroadwayWebSocket.State.md

Defines the structure for WebSocket reconnection retry parameters.

```elixir
@type retry_opts() :: %{
  retries_left: non_neg_integer(),
  max_retries: non_neg_integer(),
  delay: non_neg_integer()
}
```

--------------------------------

### Define on_upgrade_result type

Source: https://hexdocs.pm/off_broadway_websocket/OffBroadwayWebSocket.Producer.md

Defines the result type for WebSocket upgrade operations.

```elixir
@type on_upgrade_result() :: {:ok, [outbound_frame()]} | {:error, term()}
```

--------------------------------

### OffBroadwayWebSocket.Producer

Source: https://hexdocs.pm/off_broadway_websocket/OffBroadwayWebSocket.Producer.md

A GenStage producer that streams data from a WebSocket connection. It uses the gun library for connection management and includes features like buffering, demand-based dispatching, connection supervision with ping/pong, and automatic reconnection with a backoff strategy.

```APIDOC
## OffBroadwayWebSocket.Producer

### Description
A `GenStage` producer that streams data from a WebSocket connection.

Under the hood it relies on the **gun** library to establish and manage the
connection. Incoming frames are buffered and dispatched based on the demand
from Broadway consumers. Idle connections are supervised with ping/pong logic
and automatic reconnection using a user supplied backoff strategy.

### Types

#### `frame_handler_result()`

```elixir
@type frame_handler_result() ::
  {:emit, [term()], term()} | {:skip, term()} | {:error, term(), term()}
```

#### `on_upgrade_result()`

```elixir
@type on_upgrade_result() :: {:ok, [outbound_frame()]} | {:error, term()}
```

#### `outbound_frame()`

```elixir
@type outbound_frame() :: {:text | :binary, iodata()}
```

### Functions

#### `start_link(opts)`

```elixir
@spec start_link(keyword()) :: GenServer.on_start()
```

Starts the WebSocket producer under a GenStage supervisor.

### See Also

* Consult [api-reference.md](api-reference.md) for complete listing.
```

--------------------------------

### Attach Telemetry Handler for WebSocket Success

Source: https://hexdocs.pm/off_broadway_websocket/telemetry.md

Use this to attach a handler that will be called when a WebSocket connection succeeds. The handler receives the event, measurements, and metadata.

```elixir
:telemetry.attach(
  "off-broadway-websocket-success",
  [:websocket_producer, :connection, :success],
  fn event, measurements, metadata, _config ->
    IO.inspect({event, measurements, metadata}, label: "telemetry")
  end,
  nil
)
```

--------------------------------

### Stateful Inbound Frame Handler Callback

Source: https://hexdocs.pm/off_broadway_websocket/index.html

Defines the return values for a `:frame_handler` callback, used for processing raw WebSocket frames with connection-local state.

```elixir
 {:emit, [payload, ...], new_state}
 {:skip, new_state}
 {:error, reason, new_state}
```

--------------------------------

### Default Retry State for OffBroadwayWebSocket

Source: https://hexdocs.pm/off_broadway_websocket/configuration.md

This map defines the default retry behavior for websocket connections, specifying the maximum number of retries, remaining retries, and the delay between retries in milliseconds.

```elixir
%{
  max_retries: 5,
  retries_left: 5,
  delay: 10_000
}
```

--------------------------------

### Define State struct type

Source: https://hexdocs.pm/off_broadway_websocket/OffBroadwayWebSocket.State.md

Defines the primary state struct for the WebSocket producer.

```elixir
@type t() :: %OffBroadwayWebSocket.State{
  await_timeout: non_neg_integer(),
  conn_pid: pid() | nil,
  frame_handler: {module(), atom(), [term()]} | nil,
  frame_handler_init_state: term(),
  frame_handler_state: term(),
  gun_opts: map(),
  headers: [{String.t(), String.t()}],
  headers_fn:
    (-&gt; [{String.t(), String.t()}]
        | {:ok, [{String.t(), String.t()}]}
        | {:error, term()})
    | nil,
  last_msg_dt: DateTime.t() | nil,
  max_demand: non_neg_integer(),
  message_queue: :queue.queue(any()),
  min_demand: non_neg_integer(),
  on_upgrade: {module(), atom(), [term()]} | nil,
  path: String.t() | nil,
  pid: pid() | nil,
  queue_size: non_neg_integer(),
  stream_ref: reference() | nil,
  telemetry_id: atom(),
  total_demand: non_neg_integer(),
  url: String.t() | nil,
  ws_init_retry_opts: retry_opts() | nil,
  ws_retry_fun: (retry_opts() -&gt; retry_opts()),
  ws_retry_opts: retry_opts() | nil,
  ws_timeout: non_neg_integer() | nil
}
```

--------------------------------

### Define default_ws_retry_fun specification

Source: https://hexdocs.pm/off_broadway_websocket/OffBroadwayWebSocket.State.md

Specifies the function signature for the default reconnection backoff logic.

```elixir
@spec default_ws_retry_fun(retry_opts()) :: retry_opts()
```

--------------------------------

### Define outbound_frame type

Source: https://hexdocs.pm/off_broadway_websocket/OffBroadwayWebSocket.Producer.md

Defines the structure for frames sent over the WebSocket connection.

```elixir
@type outbound_frame() :: {:text | :binary, iodata()}
```

--------------------------------

### OffBroadwayWebSocket.State

Source: https://hexdocs.pm/off_broadway_websocket/api-reference.md

Holds the connection and retry state for a WebSocket producer.

```APIDOC
## OffBroadwayWebSocket.State

### Description
Holds the connection and retry state for a WebSocket producer in an Off-Broadway pipeline.

### Method
N/A (Module Documentation)

### Endpoint
N/A (Module Documentation)

### Parameters
N/A (Module Documentation)

### Request Example
N/A (Module Documentation)

### Response
N/A (Module Documentation)

### Response Example
N/A (Module Documentation)
```

--------------------------------

### OffBroadwayWebSocket.Producer

Source: https://hexdocs.pm/off_broadway_websocket/api-reference.md

A GenStage producer that streams data from a WebSocket connection.

```APIDOC
## OffBroadwayWebSocket.Producer

### Description
A `GenStage` producer that streams data from a WebSocket connection.

### Method
N/A (Module Documentation)

### Endpoint
N/A (Module Documentation)

### Parameters
N/A (Module Documentation)

### Request Example
N/A (Module Documentation)

### Response
N/A (Module Documentation)

### Response Example
N/A (Module Documentation)
```

--------------------------------

### Define frame_handler_result type

Source: https://hexdocs.pm/off_broadway_websocket/OffBroadwayWebSocket.Producer.md

Defines the return structure for frame handlers, supporting emission of data, skipping frames, or error reporting.

```elixir
@type frame_handler_result() ::
  {:emit, [term()], term()} | {:skip, term()} | {:error, term(), term()}
```
Tokens:
Raw