ZenWebsocket

A robust WebSocket client library for Elixir, built on Gun transport for production-grade reliability. Designed for financial APIs with automatic reconnection, comprehensive error handling, and real-world testing.

Features

• Gun Transport - Battle-tested HTTP/2 and WebSocket client
• Automatic Reconnection - Exponential backoff with state preservation
• Financial-Grade Reliability - Built for high-frequency trading systems
• Platform Adapters - Ready-to-use Deribit integration, extensible for others
• Real WebSocket Testing - Local real-stack servers plus opt-in live endpoint checks
• Simple API - connect, send, subscribe, monitor, and reconnect
• Comprehensive Error Handling - Categorized errors with recovery strategies
• Rate Limiting - Standalone token bucket you drive from your own code
• JSON-RPC 2.0 - Full protocol support with correlation tracking
• Pool Load Balancing - Health-based routing with automatic failover
• Session Recording - JSONL message recording for debugging and replay
• Test Utilities - Consumer-facing test helpers with mock server
• Self-Describing API - Progressive discovery via describe/0-2 for MCP tools and agents

Installation

Add zen_websocket to your dependencies in mix.exs:

def deps do
  [
    {:zen_websocket, "~> 0.7"}
  ]
end

Quick Start

Basic Connection

# Connect to a WebSocket endpoint (use your actual endpoint).
# Heartbeats are off unless you pass `heartbeat_config`.
{:ok, client} = ZenWebsocket.Client.connect("wss://api.example.com/ws", [
  timeout: 5000,
  heartbeat_config: %{type: :ping_pong, interval: 30_000}
])

# Send a message (must be binary — use Jason.encode!/1 for maps)
:ok = ZenWebsocket.Client.send_message(client, Jason.encode!(%{action: "ping"}))

# Receive messages in your process
receive do
  {:websocket_message, message} ->
    # Process the incoming message
    handle_message(message)
after
  5_000 -> {:error, :timeout}
end

# Close the connection
:ok = ZenWebsocket.Client.close(client)

Subscription Management

ZenWebsocket tracks Deribit-dialect JSON-RPC subscriptions and restores them after reconnection:

# Subscribe via Deribit's public/subscribe (no request id — tracked at send time)
{:ok, client} = ZenWebsocket.Client.connect("wss://api.example.com/ws")
:ok = ZenWebsocket.Client.subscribe(client, ["ticker.BTC", "trades.BTC"])

# On reconnect, tracked Deribit-dialect subscriptions are restored automatically

What the library handles:

• Tracking public/subscribe / public/unsubscribe via SubscriptionManager
• Client.subscribe/2 tracking is optimistic (applied at send; a server rejection stays in the restore set)
• Automatic restoration after reconnection (when restore_subscriptions: true, the default)
• Building restore messages as Deribit public/subscribe

What your client needs to handle:

• Subscription state and reconnect replay for non-Deribit dialects
• Processing subscription data messages (sent to your handler callback)
• Deribit unsubscription requests (public/unsubscribe updates tracking)
• Authentication before subscribing to private channels

Configuration:

# Disable automatic subscription restoration
{:ok, client} = ZenWebsocket.Client.connect("wss://api.example.com/ws",
  restore_subscriptions: false
)

For more detailed examples, see our working examples with fully tested implementations:

• Basic Usage - Connection management and messaging
• Error Handling - Robust error recovery patterns
• JSON-RPC Client - JSON-RPC 2.0 protocol usage
• Subscription Management - Channel subscription patterns

See the Examples Guide for an index of every in-tree example module and its test suite.

Session Recording

Record WebSocket sessions for debugging and replay:

# Enable recording when connecting
{:ok, client} = ZenWebsocket.Client.connect("wss://api.example.com/ws",
  record_to: "/tmp/session.jsonl"
)

# Use the connection normally - all messages are recorded
ZenWebsocket.Client.send_message(client, Jason.encode!(%{action: "subscribe", channel: "trades"}))

# Close to flush remaining buffer
ZenWebsocket.Client.close(client)

# Replay the recorded session
ZenWebsocket.Recorder.replay("/tmp/session.jsonl", fn entry ->
  IO.inspect(entry, label: "#{entry.dir} at #{entry.ts}")
end)

# Get session metadata
{:ok, meta} = ZenWebsocket.Recorder.metadata("/tmp/session.jsonl")
# => %{count: 42, inbound: 30, outbound: 12, duration_ms: 5000, ...}

Connection Pool Load Balancing

Route messages to the healthiest connection in a pool:

# Start the supervisor in your application
{:ok, _} = ZenWebsocket.ClientSupervisor.start_link([])

# Create multiple supervised connections
# Supply `handler:` when user code needs unsolicited inbound frames
opts = [handler: fn msg -> send(MyApp.Consumer, {:ws, msg}) end]

{:ok, _client1} = ZenWebsocket.ClientSupervisor.start_client("wss://api.example.com/ws", opts)
{:ok, _client2} = ZenWebsocket.ClientSupervisor.start_client("wss://api.example.com/ws", opts)
{:ok, _client3} = ZenWebsocket.ClientSupervisor.start_client("wss://api.example.com/ws", opts)

message = Jason.encode!(%{method: "public/test"})

# Send messages with automatic load balancing
# Routes to healthiest connection based on: pending requests, latency, errors
:ok = ZenWebsocket.ClientSupervisor.send_balanced(message)

# Automatic failover on connection failure (max 3 attempts by default)
:ok = ZenWebsocket.ClientSupervisor.send_balanced(message, max_attempts: 5)

# Check pool health
health = ZenWebsocket.PoolRouter.pool_health(ZenWebsocket.ClientSupervisor.list_clients())
# => [%{pid: #PID<0.123.0>, health: 95}, %{pid: #PID<0.124.0>, health: 87}, ...]

Rate Limiting

ZenWebsocket.RateLimiter is a standalone token bucket backed by a named ETS table. Client never consults it — call it from your own code before sending.

alias ZenWebsocket.RateLimiter

# All four config keys are required; a missing `:tokens` raises KeyError.
{:ok, :deribit_limiter} = RateLimiter.init(:deribit_limiter, %{
  tokens: 100,
  refill_rate: 10,
  refill_interval: 1_000,
  request_cost: &RateLimiter.deribit_cost/1
})

request = %{"method" => "public/test"}

case RateLimiter.consume(:deribit_limiter, request) do
  :ok -> ZenWebsocket.Client.send_message(client, Jason.encode!(request))
  {:error, :rate_limited} -> {:error, :rate_limited}
end

{:ok, %{tokens: tokens}} = RateLimiter.status(:deribit_limiter)

# ETS tables are not reclaimed on process exit — free it explicitly
:ok = RateLimiter.shutdown(:deribit_limiter)

init/2 schedules refills with Process.send_after/3 against the calling process. That process must handle {:refill, name} or refills stop permanently:

def handle_info({:refill, name}, state) do
  ZenWebsocket.RateLimiter.refill(name)
  {:noreply, state}
end

Deribit Integration

DeribitGenServerAdapter is the recommended supervised entry point. ZenWebsocket.Examples.DeribitAdapter is the functional/struct-based variant: connect/1 returns a struct you thread through authenticate/1, subscribe/2, unsubscribe/2, and send_request/2,3.

# Configure Deribit credentials. Opts are a keyword list; the endpoint is
# selected with `:url` (it defaults to test.deribit.com when omitted).
config = [
  client_id: System.get_env("DERIBIT_CLIENT_ID"),
  client_secret: System.get_env("DERIBIT_CLIENT_SECRET"),
  url: "wss://test.deribit.com/ws/api/v2"
]

# Start the supervised adapter
{:ok, adapter} = ZenWebsocket.Examples.DeribitGenServerAdapter.start_link(config)

# Authenticate before any private/* request. The adapter does not authenticate
# on first connect - it only re-authenticates after a reconnect.
:ok = ZenWebsocket.Examples.DeribitGenServerAdapter.authenticate(adapter)

# Subscribe to market data
:ok = ZenWebsocket.Examples.DeribitGenServerAdapter.subscribe(
  adapter,
  ["book.BTC-PERPETUAL.raw", "trades.BTC-PERPETUAL.raw"]
)

# Send a custom JSON-RPC request (e.g., place an order)
{:ok, response} = ZenWebsocket.Examples.DeribitGenServerAdapter.send_request(
  adapter,
  "private/buy",
  %{
    instrument_name: "BTC-PERPETUAL",
    amount: 10,
    type: "limit",
    price: 50000
  }
)

Architecture

ZenWebsocket follows a modular architecture with clear separation of concerns:

ZenWebsocket.Client              # Main client interface
ZenWebsocket.Config              # Configuration management
ZenWebsocket.Frame               # WebSocket frame handling
ZenWebsocket.Reconnection        # Internal reconnection helper (not in describe/0)
ZenWebsocket.MessageHandler      # Message parsing and routing
ZenWebsocket.ErrorHandler        # Error categorization
ZenWebsocket.RateLimiter         # API rate limiting
ZenWebsocket.JsonRpc             # JSON-RPC 2.0 protocol
ZenWebsocket.HeartbeatManager    # Internal heartbeat manager (not in describe/0)
ZenWebsocket.SubscriptionManager # Internal subscription tracking (not in describe/0)
ZenWebsocket.RequestCorrelator   # Internal request correlation (not in describe/0)
ZenWebsocket.Recorder            # Session recording (pure functions)
ZenWebsocket.RecorderServer      # Async file I/O for recording
ZenWebsocket.PoolRouter          # Health-based connection pool routing
ZenWebsocket.Testing             # Consumer-facing test utilities

Platform Integration

The library includes a complete Deribit adapter as a reference implementation. To integrate with other platforms:

1. Create an adapter module following the Deribit pattern
2. Implement platform-specific authentication
3. Handle platform message formats
4. Add comprehensive tests against the real API

Two Deribit adapters ship as reference implementations:

• ZenWebsocket.Examples.DeribitGenServerAdapter (lib/zen_websocket/examples/deribit_genserver_adapter.ex) — the recommended supervised entry point: start_link/1 plus authenticate/1, subscribe/2, send_request/2,3, get_state/1, with reconnect handling in the GenServer.
• ZenWebsocket.Examples.DeribitAdapter (lib/zen_websocket/examples/deribit_adapter.ex) — the functional/struct-based variant: connect/0,1 returns a struct you thread through authenticate/1, subscribe/2, unsubscribe/2, send_request/2,3.

Documentation

Comprehensive guides are available in the docs/guides/ directory:

Guide	Description
Building Adapters	Create platform adapters with heartbeat, auth, and reconnection patterns
Performance Tuning	Configure timeouts, rate limiting, and memory for your use case
Troubleshooting Reconnection	Debug connection issues and reconnection logic
Deployment Considerations	Latency, geography, architecture, and monitoring questions for trading apps
AGENTS.md	Guide for AI coding agents contributing to this project

See the full HexDocs documentation for API reference and module documentation.

Shipped elixir / iex fences are checked against the public API. Mark a deliberately incomplete fragment elixir illustrative, or start it with # illustrative, so the checker leaves it alone.

Configuration Options

Option	Description	Default
url	WebSocket endpoint URL	required
headers	Custom headers for connection	[]
timeout	Connection timeout in milliseconds	5000
retry_count	Maximum reconnection attempts	3
retry_delay	Initial retry delay in milliseconds	1000
heartbeat_config	Heartbeat mode (see below); heartbeats are off until you set it	:disabled
heartbeat_interval	Fallback interval used when heartbeat_config omits :interval	30000
reconnect_on_error	Enable automatic reconnection	true
restore_subscriptions	Restore tracked Deribit subscriptions after reconnect	true
record_to	Path to JSONL file for session recording	nil
debug	Enable verbose debug logging	false

Heartbeat Configuration

heartbeat_config is either the atom :disabled (the default — no heartbeat is ever sent) or a map with a :type:

type	Behavior
:ping_pong	Sends a WebSocket ping frame each interval and measures the pong round trip
:deribit	Sends Deribit's public/set_heartbeat dialect and answers test_request
:binance	Inbound-only: incoming heartbeat messages are consumed, nothing is sent outbound

{:ok, client} = ZenWebsocket.Client.connect("wss://api.example.com/ws",
  heartbeat_config: %{type: :ping_pong, interval: 30_000}
)

:interval falls back to the heartbeat_interval option when omitted.

Debug Logging

Debug logging is disabled by default to keep library output quiet. Enable it for troubleshooting connection issues:

# Enable debug logging for troubleshooting
{:ok, client} = ZenWebsocket.Client.connect("wss://example.com", debug: true)

When enabled, you'll see detailed logs for connection establishment, WebSocket upgrades, frame handling, heartbeats, and reconnection attempts.

Using Debug in Custom Adapters:

If you're building a custom adapter or extension, use ZenWebsocket.Debug.log/2 with the Config struct:

alias ZenWebsocket.Debug

# Always pass the Config struct (not the full state map)
Debug.log(config, "Custom adapter initialized")
Debug.log(state.config, "Processing message: #{inspect(msg)}")

The function is a no-op when debug: false (the default), so you can leave debug statements in production code without performance impact.

Testing Your Application

ZenWebsocket provides test utilities for testing your own WebSocket clients:

defmodule MyApp.WebSocketTest do
  use ExUnit.Case

  alias ZenWebsocket.Testing

  setup do
    {:ok, server} = Testing.start_mock_server()
    on_exit(fn -> Testing.stop_server(server) end)
    {:ok, server: server}
  end

  test "client sends expected message", %{server: server} do
    # Connect your client to the mock server
    {:ok, client} = ZenWebsocket.Client.connect(server.url)

    # Send a message
    ZenWebsocket.Client.send_message(client, ~s({"type": "ping"}))

    # Assert the server received it (supports string, regex, map, or function matchers)
    assert Testing.assert_message_sent(server, %{"type" => "ping"}, 1000)

    ZenWebsocket.Client.close(client)
  end

  test "client handles server messages", %{server: server} do
    {:ok, client} = ZenWebsocket.Client.connect(server.url)

    # Inject a message from the server
    Testing.inject_message(server, ~s({"type": "notification", "data": "hello"}))

    # Your client should receive it
    assert_receive {:websocket_message, msg}, 1000
    assert String.contains?(msg, "notification")

    ZenWebsocket.Client.close(client)
  end

  test "client handles disconnection", %{server: server} do
    {:ok, client} = ZenWebsocket.Client.connect(server.url, reconnect_on_error: false)

    ref = Process.monitor(client.server_pid)

    # Simulate server disconnect
    Testing.simulate_disconnect(server, :going_away)

    # Verify client detected disconnect
    assert_receive {:DOWN, ^ref, :process, _, _}, 1000
    refute Process.alive?(client.server_pid)
  end
end

Testing Philosophy

Pure logic is unit-tested without sockets. Integration tests exercise either the local real WebSocket stack or opt-in live provider endpoints; exchange responses, authentication, and provider behavior are never simulated.

# Quick test health check
mix test.json --quiet --summary-only

# Iterate on failures
mix test.json --quiet --failed --first-failure

# Include integration tests
mix test --include integration

Contributing

1. Fork the repository
2. Create your feature branch (git checkout -b feature/amazing-feature)
3. Write unit tests and real-stack integration tests; do not simulate exchange behavior
4. Ensure the verification workflow passes (see Development Commands below)
5. Commit your changes
6. Push to the branch
7. Open a Pull Request

Development Commands

mix compile           # Compile the project
mix test.json --quiet --summary-only   # Quick test health check
mix test.json --quiet --failed --first-failure # Iterate on failures
mix dialyzer.json --quiet              # Type checking
mix credo --strict --format json       # Static analysis
mix security                           # Sobelow security scan
mix docs             # Generate documentation
mix test --include integration         # Full test suite with integration

License

This project is licensed under the MIT License.

Links

• Documentation
• Hex Package
• GitHub Repository

Acknowledgments

Built for the Elixir community by ZenHive.