### Complete Production WebSocket Example

Source: https://context7.com/mpol1t/off_broadway_websocket/llms.txt

A full production setup for OffBroadwayWebSocket, including TLS, authentication, subscriptions, stateful frame handling, custom retry logic, and telemetry.

```elixir
defmodule MyApp.Exchange.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://ws.exchange.example:443",
          path: "/v2/realtime",
          ws_timeout: 30_000,
          await_timeout: 15_000,
          telemetry_id: :exchange_feed,
          headers_fn: &MyApp.Exchange.Auth.ws_headers/0,
          on_upgrade: {MyApp.Exchange.Subscriptions, :bootstrap, [["BTC-USD", "ETH-USD"]}},
          frame_handler: {MyApp.Exchange.Protocol, :handle_frame, []},
          frame_handler_state: %{channels: %{}, sequence: 0},
          ws_retry_opts: %{max_retries: 20, retries_left: 20, delay: 1_000},
          ws_retry_fun: &MyApp.RetryStrategy.exponential_backoff/1,
          gun_opts: %{
            connect_timeout: 10_000,
            protocols: [:http],
            transport: :tls,
            tls_opts: [
              verify: :verify_peer,
              cacertfile: CAStore.file_path(),
              depth: 3,
              verify_fun: [
                 பகிர்:ssl_verify_hostname.verify_fun/3,
                [check_hostname: ~c"ws.exchange.example"]
              ]
            ],
            http_opts: %{version: :"HTTP/1.1"},
            ws_opts: %{keepalive: 15_000, silence_pings: false}
          }
        ],
        transformer: {__MODULE__, :transform, []},
        concurrency: 1
      ],
      processors: [
        default: [min_demand: 0, max_demand: 50, concurrency: 4]
      ],
      batchers: [
        default: [batch_size: 100, batch_timeout: 500, concurrency: 2]
      ]
    )
  end

  @impl true
  def handle_message(_processor, %Message{data: %{channel: channel, data: data}} = message, _context) do
    case channel do
      "trades" -> MyApp.TradeProcessor.process(data)
      "orderbook" -> MyApp.OrderbookProcessor.process(data)
      _ -> :ok
    end
    message
  end

  @impl true
  def handle_batch(_batcher, messages, _batch_info, _context) do
    # Batch persist to database
    payloads = Enum.map(messages, & &1.data)
    MyApp.Storage.bulk_insert(payloads)
    messages
  end

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

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

### Start OffBroadwayWebSocket Producer

Source: https://context7.com/mpol1t/off_broadway_websocket/llms.txt

Configure and start the OffBroadwayWebSocket producer within a Broadway pipeline. This example demonstrates setting up a connection to a WebSocket stream with custom Gun options and message transformation.

```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: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}
          }
        },
        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
    # Process incoming websocket payload
    IO.inspect(payload, label: "Received")
    message
  end

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

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

### Attach Telemetry Handler Example

Source: https://github.com/mpol1t/off_broadway_websocket/blob/main/docs/guides/telemetry.md

An example demonstrating how to attach a telemetry handler to listen for `:success` events on the websocket connection.

```APIDOC
## Example

```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
)
```
```

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

### Minimal Broadway Setup with off_broadway_websocket

Source: https://github.com/mpol1t/off_broadway_websocket/blob/main/docs/guides/getting_started.md

This Elixir code demonstrates a minimal Broadway setup using OffBroadwayWebSocket.Producer. Configure the producer with your websocket URL, path, and telemetry ID.

```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
```

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

### Configure gun_opts for websocket connections

Source: https://github.com/mpol1t/off_broadway_websocket/blob/main/docs/guides/configuration.md

Example configuration for the underlying gun library, including TLS settings, keepalive intervals, and protocol versions.

```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"
  }
}
```

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

### Custom Retry Strategy for WebSocket Connections

Source: https://context7.com/mpol1t/off_broadway_websocket/llms.txt

Configure custom retry behavior for WebSocket reconnections using :ws_retry_opts and :ws_retry_fun. This example implements exponential backoff with jitter, capped at 60 seconds, to manage reconnection attempts after failures.

```elixir
defmodule MyApp.RetryStrategy do
  @doc """
  Exponential backoff with jitter, capped at 60 seconds.
  """
  def exponential_backoff(%{retries_left: 0} = opts), do: opts

  def exponential_backoff(%{retries_left: n, delay: delay, max_retries: max} = opts) do
    attempt = max - n + 1
    base_delay = min(delay * :math.pow(2, attempt - 1), 60_000)
    jitter = :rand.uniform(round(base_delay * 0.1))

    %{opts | retries_left: n - 1, delay: round(base_delay + jitter)}
  end
end

defmodule MyApp.Broadway do
  use Broadway

  def start_link(_opts) do
    Broadway.start_link(__MODULE__,
      name: __MODULE__,
      producer: [
        module: {
          OffBroadwayWebSocket.Producer,
          url: "wss://api.example.com",
          path: "/feed",
          ws_retry_opts: %{
            max_retries: 10,
            retries_left: 10,
            delay: 1_000  # Initial delay: 1 second
          },
          ws_retry_fun: &MyApp.RetryStrategy.exponential_backoff/1
        },
        transformer: {__MODULE__, :transform, []},
        concurrency: 1
      ],
      processors: [default: []]
    )
  end

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

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

### Fetch Dependencies

Source: https://github.com/mpol1t/off_broadway_websocket/blob/main/docs/guides/getting_started.md

Run this command in your terminal to fetch all the dependencies listed in your mix.exs file, including off_broadway_websocket.

```bash
mix deps.get
```

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

### Run project tests

Source: https://github.com/mpol1t/off_broadway_websocket/blob/main/README.md

Execute the test suite using the Mix build tool.

```bash
mix test
```

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

### Configure on_upgrade callback

Source: https://github.com/mpol1t/off_broadway_websocket/blob/main/docs/guides/on_upgrade_bootstrap.md

Pass an MFA tuple to the :on_upgrade option to define the bootstrap logic.

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

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

### Producer Configuration Options

Source: https://context7.com/mpol1t/off_broadway_websocket/llms.txt

Overview of required and optional configuration parameters for the OffBroadwayWebSocket.Producer, including URL, path, timeouts, retry options, and Gun client settings.

```elixir
# Required options
url: "wss://api.exchange.example"   # WebSocket base URL
path: "/v1/stream"                   # Request path and query string

# Common optional settings
ws_timeout: 30_000                   # Idle timeout in milliseconds
await_timeout: 10_000                # Timeout for :gun.await_up/2
telemetry_id: :my_producer           # Prefix for telemetry events
headers: [{"x-api-key", "secret"}]   # Static websocket upgrade headers

# Retry behavior
ws_retry_opts: %{
  max_retries: 5,
  retries_left: 5,
  delay: 10_000
}

# Gun HTTP client options
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"
  }
}
```

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

### Attach Telemetry Event Handlers

Source: https://context7.com/mpol1t/off_broadway_websocket/llms.txt

Attach handlers to track connection lifecycle events like success, failure, disconnection, timeouts, and status changes. Call `setup/0` during application startup.

```elixir
defmodule MyApp.TelemetryHandler do
  require Logger

  def setup do
    events = [
      [:my_producer, :connection, :success],
      [:my_producer, :connection, :failure],
      [:my_producer, :connection, :disconnected],
      [:my_producer, :connection, :timeout],
      [:my_producer, :connection, :status]
    ]

    :telemetry.attach_many(
      "my-app-websocket-telemetry",
      events,
      &__MODULE__.handle_event/4,
      nil
    )
  end

  def handle_event([:my_producer, :connection, :success], %{count: 1}, %{url: url}, _config) do
    Logger.info("WebSocket connected: #{url}")
  end

  def handle_event([:my_producer, :connection, :failure], %{count: 1}, %{reason: reason}, _config) do
    Logger.error("WebSocket connection failed: #{inspect(reason)}")
  end

  def handle_event([:my_producer, :connection, :disconnected], %{count: 1}, %{reason: reason}, _config) do
    Logger.warning("WebSocket disconnected: #{inspect(reason)}")
  end

  def handle_event([:my_producer, :connection, :timeout], %{count: 1}, _metadata, _config) do
    Logger.warning("WebSocket idle timeout triggered")
  end

  def handle_event([:my_producer, :connection, :status], %{value: value}, _metadata, _config) do
    Logger.debug("WebSocket status: #{if value == 1, do: "connected", else: "disconnected"}")
  end
end

# Call during application startup
MyApp.TelemetryHandler.setup()
```

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

### Websocket Upgrade Bootstrap Configuration

Source: https://github.com/mpol1t/off_broadway_websocket/blob/main/docs/guides/on_upgrade_bootstrap.md

Configure the :on_upgrade option with a callback module and function to send bootstrap frames after a websocket upgrade.

```APIDOC
## POST /websocket

### Description
This endpoint is used to establish a websocket connection and can be configured to send bootstrap frames after the upgrade is complete.

### Method
POST

### Endpoint
/websocket

### Parameters
#### Query Parameters
- **on_upgrade** (tuple) - Required - A tuple specifying the callback module, function, and arguments for sending bootstrap frames. Example: `{MyApp.Subscriptions, :bootstrap_frames, [args]}`

### Request Body
This endpoint does not typically have a request body for the initial upgrade.

### Response
#### Success Response (101 Switching Protocols)
- **Connection** (string) - Indicates the connection is being upgraded to a websocket.
- **Upgrade** (string) - Indicates the protocol is websocket.

#### Response Example
(No specific JSON response body for the upgrade itself, but the websocket connection is established.)

## Callback Contract

The callback function specified in `:on_upgrade` must adhere to the following contract:

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

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

### Failure Behavior
Bootstrap failures include:
- The callback returning `{:error, reason}`.
- The callback returning an invalid result.
- Immediate failure of `:gun.ws_send/3` for any returned frame.
- The callback raising an exception.

Bootstrap failures trigger the same reconnect and backoff mechanisms as connection failures.
```

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

### OffBroadwayWebSocket.Producer Configuration Options

Source: https://github.com/mpol1t/off_broadway_websocket/blob/main/docs/guides/configuration.md

This section outlines the primary configuration options for the OffBroadwayWebSocket.Producer.

```APIDOC
## Configuration Options for OffBroadwayWebSocket.Producer

### Required Options

- **:url** (string) - The base URL for the WebSocket connection.
- **:path** (string) - The request path and query string for the WebSocket connection.

### Common Optional Settings

- **:ws_timeout** (integer) - Idle timeout for the WebSocket connection in milliseconds.
- **:await_timeout** (integer) - Timeout for the `:gun.await_up/2` function.
- **:headers** (list of tuples) - WebSocket upgrade headers (e.g., `[{"Authorization", "Bearer token"}]`).
- **:headers_fn** (callback) - An optional zero-arity callback function that returns headers, used for per-connect header refresh.
- **:telemetry_id** (string) - A prefix for telemetry events.
- **:gun_opts** (map) - Options to be forwarded to the underlying `:gun` library for connection configuration.
- **:min_demand** (integer) - Broadway producer setting for minimum demand.
- **:max_demand** (integer) - Broadway producer setting for maximum demand.

### Retry Behavior

- **:ws_retry_opts** (map) - Configuration for retry state.
- **:ws_retry_fun** (function) - A function used to compute the next retry state.

**Default Retry State:**
```elixir
%{max_retries: 5, retries_left: 5, delay: 10_000}
```

### Bootstrap and Session Hooks

- **:on_upgrade** (MFA tuple) - An optional MFA (Module.Function.Arity) to be run after the WebSocket upgrade and before the producer is ready.
- **:frame_handler** (MFA tuple) - An optional MFA to process normalized inbound data frames.
- **:frame_handler_state** (any) - The initial connection-local state for the `:frame_handler`.

### Header Refresh Strategy

- Use static `:headers` for long-lived authentication tokens.
- Use `:headers_fn` when authentication headers need to be refreshed on each reconnect.

**Return values for `:headers_fn`:**
- `[{"header", "value"}]` - A list of headers.
- `{:ok, [{"header", "value"}]}` - Success with a list of headers.
- `{:error, reason}` - Indicates a connection failure, triggering retry/backoff.

### `gun_opts` Example

Use `:gun_opts` to configure transport, TLS verification, HTTP version, and WebSocket keepalive behavior.

```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"
  }
}
```
```

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

### On-Upgrade Bootstrap Callback

Source: https://github.com/mpol1t/off_broadway_websocket/blob/main/README.md

Defines a callback function for sending initial frames (e.g., subscribe or auth) immediately after a WebSocket upgrade. It must return {:ok, []}, {:ok, [{:text | :binary, iodata()}, ...]}, or {:error, reason}.

```elixir
defmodule MyApp.WebSocket do
  # ...

  def subscription_frames(_args) do
    {:ok, ["subscribe", :"my_channel"]}
  end

  def handle_frame(payload, state) do
    # Process inbound frame based on state
    {:emit, [payload], state}
  end
end
```

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

### Configure Frame Handler and State

Source: https://github.com/mpol1t/off_broadway_websocket/blob/main/docs/guides/frame_handler.md

Configure the `:frame_handler` with a module, function, and arguments, and optionally provide an initial `:frame_handler_state`. This is used for session-aware WebSocket protocols.

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

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

### Refreshing Headers on Each Connect

Source: https://github.com/mpol1t/off_broadway_websocket/blob/main/docs/guides/auth_refresh_and_handshake_failures.md

Use `headers_fn` to provide dynamic headers, such as authorization tokens, that are refreshed before each connection attempt. The function is invoked prior to every connection, including automatic reconnects.

```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
  }
]
```

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

### Add off_broadway_websocket Dependency

Source: https://github.com/mpol1t/off_broadway_websocket/blob/main/docs/guides/getting_started.md

Add the off_broadway_websocket dependency to your project's mix.exs file to include it in your dependencies.

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

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

### On-Upgrade Bootstrap Frames

Source: https://context7.com/mpol1t/off_broadway_websocket/llms.txt

Use `:on_upgrade` when a websocket API requires subscribe or authentication frames to be sent immediately after the upgrade completes and before the connection is considered ready. The callback must return {:ok, frames} or {:error, reason}.

```elixir
defmodule MyApp.Subscriptions do
  @doc """
  Called after websocket upgrade to send subscription frames.
  Must return {:ok, frames} or {:error, reason}.
  """
  def bootstrap_frames(channels) do
    frames = Enum.map(channels, fn channel ->
      payload = Jason.encode!(%{action: "subscribe", channel: channel})
      {:text, payload}
    end)
    {:ok, frames}
  end
end
```

```elixir
defmodule MyApp.Broadway do
  use Broadway

  def start_link(_opts) do
    Broadway.start_link(__MODULE__,
      name: __MODULE__,
      producer: [
        module: {
          OffBroadwayWebSocket.Producer,
          url: "wss://ws.exchange.example",
          path: "/realtime",
          on_upgrade: {MyApp.Subscriptions, :bootstrap_frames, [["trades", "orderbook"]]},
          telemetry_id: :exchange_ws
        },
        transformer: {__MODULE__, :transform, []},
        concurrency: 1
      ],
      processors: [default: []]
    )
  end

  # on_upgrade callback return values:
  # - {:ok, []}                              # No frames to send
  # - {:ok, [{:text, "payload"}, ...]}       # Send text frames
  # - {:ok, [{:binary, <<data>>}, ...]}      # Send binary frames
  # - {:error, reason}                       # Bootstrap failure, triggers reconnect

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

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

### Configure Broadway Producer for WebSocket

Source: https://github.com/mpol1t/off_broadway_websocket/blob/main/README.md

Configure a Broadway producer to use OffBroadwayWebSocket for ingesting data from a WebSocket stream. Includes options for URL, path, timeouts, TLS settings, and custom 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
```

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

### Stateful Inbound Frame Handling Callback

Source: https://github.com/mpol1t/off_broadway_websocket/blob/main/README.md

Defines a callback function for handling raw WebSocket frames that depend on connection-local state. It receives the normalized payload and current state, returning {:emit, [payload, ...], new_state}, {:skip, new_state}, or {:error, reason, new_state}.

```elixir
defmodule MyApp.WebSocket do
  # ...

  def handle_frame(payload, state) do
    # Example: Process message and update state
    new_state = Map.put(state, :last_message_id, payload.id)
    {:emit, [payload.data], new_state}
  end
end
```

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

### Dynamic Header Refresh with headers_fn

Source: https://context7.com/mpol1t/off_broadway_websocket/llms.txt

Use `:headers_fn` when authentication headers need to be refreshed on each connection attempt, such as short-lived tokens that expire between reconnects. The function is called before each connection attempt, including reconnects.

```elixir
defmodule MyApp.Broadway do
  use Broadway

  def start_link(_opts) do
    Broadway.start_link(__MODULE__,
      name: __MODULE__,
      producer: [
        module: {
          OffBroadwayWebSocket.Producer,
          url: "wss://api.exchange.example",
          path: "/v1/stream",
          headers_fn: fn ->
            # Called before each connection attempt, including reconnects
            token = MyApp.Auth.refresh_ws_token!()
            [{"authorization", "Bearer " <> token}]
          end
        },
        transformer: {__MODULE__, :transform, []},
        concurrency: 1
      ],
      processors: [default: []]
    )
  end

  # headers_fn can also return:
  # - {:ok, [{"header", "value"}]}
  # - {:error, reason}  # triggers retry/backoff

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

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

### Telemetry Notes

Source: https://github.com/mpol1t/off_broadway_websocket/blob/main/docs/guides/telemetry.md

Important notes regarding the emission of telemetry events. Connection success telemetry is emitted only after websocket upgrade and successful `:on_upgrade` bootstrap if configured. Frame handler failures are reported through the connection failure path.

```APIDOC
## Notes

- connection-success telemetry is emitted only after websocket upgrade and successful `:on_upgrade` bootstrap, if configured
- frame-handler failures are reported through the connection failure path
```

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

### Attach Telemetry Handler for Success Events

Source: https://github.com/mpol1t/off_broadway_websocket/blob/main/docs/guides/telemetry.md

Attaches a handler to capture and inspect telemetry events related to connection success. This is useful for monitoring and debugging.

```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 Frame Handler for WebSocket

Source: https://context7.com/mpol1t/off_broadway_websocket/llms.txt

Implement a stateful frame handler to process, filter, or transform incoming WebSocket frames based on connection-local state. Use this for session-aware protocols like channel multiplexing or subscription acknowledgements. The handler receives the payload and current state, returning {:skip, new_state}, {:emit, [payload, ...], new_state}, or {:error, reason, new_state}.

```elixir
defmodule MyApp.Transport do
  @doc """
  Handles normalized inbound data frames with connection-local state.
  Called for each incoming data frame after websocket upgrade.
  """
  def handle_frame(payload, state) do
    case Jason.decode(payload) do
      {:ok, %{"type" => "subscribed", "channel_id" => id, "channel" => name}} ->
        # Track subscription confirmations without emitting downstream
        new_state = Map.put(state.subscriptions, id, name)
        {:skip, %{state | subscriptions: new_state}}

      {:ok, %{"type" => "heartbeat"}} ->
        # Swallow heartbeats, but still refresh liveness
        {:skip, state}

      {:ok, %{"type" => "data", "channel_id" => id, "payload" => data}} ->
        # Emit data with resolved channel name
        channel = Map.get(state.subscriptions, id, "unknown")
        {:emit, [%{channel: channel, data: data}], state}

      {:ok, %{"type" => "error", "message" => reason}} ->
        # Signal error to trigger reconnect
        {:error, {:server_error, reason}, state}

      {:error, _decode_error} ->
        # Skip malformed frames
        {:skip, state}
    end
  end
end

defmodule MyApp.Broadway do
  use Broadway

  def start_link(_opts) do
    Broadway.start_link(__MODULE__,
      name: __MODULE__,
      producer: [
        module: {
          OffBroadwayWebSocket.Producer,
          url: "wss://ws.exchange.example",
          path: "/stream",
          frame_handler: {MyApp.Transport, :handle_frame, []},
          frame_handler_state: %{subscriptions: %{}},  # Resets on reconnect
          telemetry_id: :exchange_ws
        },
        transformer: {__MODULE__, :transform, []},
        concurrency: 1
      ],
      processors: [default: []]
    )
  end

  # frame_handler callback return values:
  # - {:emit, [payload, ...], new_state}   # Queue payloads downstream, update state
  # - {:skip, new_state}                   # Skip frame, update state, refresh liveness
  # - {:error, reason, new_state}          # Fail connection, trigger reconnect

  def handle_message(_stage, %Broadway.Message{data: data} = message, _context) do
    IO.inspect(data, label: "Processed")
    message
  end

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

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

### Telemetry Events

Source: https://github.com/mpol1t/off_broadway_websocket/blob/main/docs/guides/telemetry.md

Telemetry events are emitted under the topic `[:<telemetry_id>, :connection, <event>]`. The available events are `:success`, `:failure`, `:disconnected`, `:timeout`, and `:status`.

```APIDOC
## Telemetry Events

Telemetry events are emitted under:

```elixir
[:<telemetry_id>, :connection, <event>]
```

### Events

- `:success`
- `:failure`
- `:disconnected`
- `:timeout`
- `:status`
```

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

### Define default retry state

Source: https://github.com/mpol1t/off_broadway_websocket/blob/main/docs/guides/configuration.md

The default configuration map used for managing websocket connection retry attempts.

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

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

### Telemetry Metadata

Source: https://github.com/mpol1t/off_broadway_websocket/blob/main/docs/guides/telemetry.md

Additional metadata included with specific telemetry events. `:success` events include the `url`. `:failure` and `:disconnected` events include the `reason`.

```APIDOC
## Metadata

- `:success` includes `%{url: String.t()}`
- `:failure` includes `%{reason: term()}
- `:disconnected` includes `%{reason: term()}`
```

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

### Telemetry Measurements

Source: https://github.com/mpol1t/off_broadway_websocket/blob/main/docs/guides/telemetry.md

Measurements associated with each telemetry event. `:success`, `:failure`, `:disconnected`, and `:timeout` events include a `count` of 1. The `:status` event includes a `value` which can be 0 or 1.

```APIDOC
## Measurements

- `:success` -> `%{count: 1}`
- `:failure` -> `%{count: 1}`
- `:disconnected` -> `%{count: 1}`
- `:timeout` -> `%{count: 1}`
- `:status` -> `%{value: 0 | 1}`
```

=== COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.