XSockets

Ported from (and deprecates) xturn-sockets

An Elixir library for listening, dialing, and reading whole messages on UDP, TCP, TLS, DTLS, and (optionally) SCTP -- without baking in STUN, TURN, RTP, or SIP. Those protocols live in your app. XSockets just handles the pipes and the "when is a full packet ready?" part.

Home: https://github.com/Lazarus404/xsockets

What problem this solves

If you have built a real-time server, you have probably written the same GenServer loop more than once: bind a socket, read, figure out where one message ends and the next begins, call your logic, write a reply, repeat.

XSockets turns that into three plugs:

  1. Transport -- how bytes move (:udp, :tcp, :tls, :dtls)
  2. Framing -- how you spot a whole message (:raw, :length_prefixed)
  3. Handler -- what you do with each message

A shared Engine drains every complete packet from each read before it asks the socket for more. That keeps control traffic (handshakes, requests) tidy. High-rate media can skip the engine and use Transport.UDP.open_relay/2 instead.

Start here

1. Write a handler

defmodule MyApp.Handler do

  @behaviour XSockets.Handler



  @impl true

  def handle_connect(conn), do: {:ok, conn.assigns}



  @impl true

  def handle_packet(packet, _meta, conn, state) do

    IO.inspect({packet, conn.client_ip, conn.client_port})

    {:ok, state}

  end

end

2. Serve (or dial)

# UDP -- each datagram is already a whole message (:raw)

{:ok, pid} =

  XSockets.serve(MyApp.Handler,

    transport: :udp,

    ip: {0, 0, 0, 0},

    port: 3478

  )



# TCP -- messages are length-prefixed (2-byte size header by default)

{:ok, pid} =

  XSockets.serve(MyApp.Handler,

    transport: :tcp,

    ip: {0, 0, 0, 0},

    port: 3478

  )



# Call out as a client (needs :ip and :port)

{:ok, conn} =

  XSockets.dial(MyApp.Handler,

    transport: :tcp,

    ip: {127, 0, 0, 1},

    port: 3478

  )

serve/2 and dial/2 are the friendly front door. They pick sensible framing defaults and call listen/1 or Client.connect/1 for you. Pass :accumulator or :pipeline when you want to override framing. Reach for listen/1 when you need SCTP, explicit modules, or multi-step pipelines.

Try the examples

mix run examples/echo_udp.exs
  1. examples/echo_udp.exs -- UDP echo
  2. examples/echo_tcp.exs -- TCP with length prefixes
  3. examples/echo_tls.exs -- TLS with a self-signed cert
  4. examples/echo_client.exs -- serve + dial together
  5. examples/pipeline_inline.exs -- two-step pipeline (advanced)
  6. examples/udp_relay.exs -- Engine vs open_relay/2 (advanced)

Main modules

Everyday tools:

When you need more structure:

STUN, TURN, RTP, SIP framing stays in yourAccumulator / Handler modules. This library only ships the generic plumbing.

Features (in plain terms)

Control vs media: put request-shaped traffic through the Engine and your Handler. Put high-rate media on UDP.open_relay/2 (or a raw socket) so it does not share that loop.

Two kinds of SCTP (easy to mix up):

Useful config (on :xsockets or your host :config_app):

Production configuration

Defaults lean toward "fail closed under load" (pool overflow drops work; FixedWindow only watches datagrams). Opt in when you need stricter delivery:

# Do not drop work on a busy :pool tier -- run it inline instead

config :xsockets, pool_on_overflow: :inline_fallback

# or per tier: on_overflow: :inline_fallback



# Also rate-limit TCP/TLS Engine chunks

config :xsockets, rate_limiter: XSockets.RateLimit.PerConnection

Async :task / :pool work is best-effort: a crashed worker can lose what was in its mailbox, and :drop throws away work when the pool is full. Prefer :inline for small control graphs, or :inline_fallback when a pool tier must not lose messages.

Installation

def deps do

  [

    {:xsockets, "~> 1.0.0"},

    {:telemetry, "~> 1.0"}

    # optional: {:ex_sctp, "~> 0.1"}

  ]

end

Advanced

Skip this until serve / dial are not enough. It covers multi-step pipelines, reordering, async dispatch, and related knobs.

Reordering

Sometimes packets arrive out of order and you need them sorted before your handler sees them. Accumulator.Reorder wraps another accumulator, tags each whole packet with a key (for example an RTP sequence number), and holds stragglers until the window is contiguous -- or until a time/size limit says "give up and flush."

accumulator: {

  XSockets.Accumulator.Reorder,

  name: :rtp,

  inner: XSockets.Accumulator.LengthPrefixed,

  inner_opts: [header_size: 2],

  key_fun: &MyApp.RTP.sequence/2,

  window: 32,

  max_delay_ms: 100,

  on_overflow: :flush_oldest

}

Tunable keys (window, max_delay_ms, on_overflow, enabled) merge with library defaults and app config:

config :xsockets,

  config_app: :my_app,

  reorder: [

    rtp: [window: 32, max_delay_ms: 150]

  ]



config :my_app,

  reorder: [

    rtp: [window: 48]

  ]

Precedence: library defaults < config :xsockets, :reorder < config :config_app, :reorder < keys in the accumulator spec. key_fun, inner, and name always come from code -- never from config.

If an accumulator may hold packets across reads (waiting on a missing sequence number), set tick_interval_ms on Connection or DatagramServer so the Engine can flush on a timer even when no new data arrives.

Pipelines

A pipeline is a small graph of stages. The root stage sees the wire; a handler can {:descend, :other_tier, payload, state} to hand work to another stage:

defmodule MyApp.Pipeline do

  use XSockets.Pipeline



  tier :root,

    accumulator: {XSockets.Accumulator.LengthPrefixed, header_size: 2},

    handler: MyApp.Handlers.Stun



  tier :rtp,

    accumulator: XSockets.Accumulator.Raw,

    handler: MyApp.Handlers.Rtp

end



{:ok, pid} =

  XSockets.listen(

    transport: XSockets.Transport.TCP,

    ip: {0, 0, 0, 0},

    port: 3478,

    pipeline: MyApp.Pipeline,

    assigns: %{}

  )

See examples/pipeline_inline.exs.

Worth knowing:

Dispatch strategies

Each tier can say how its work runs:

Start the supervisors (or let XSockets.Application do it):

{:ok, _} = XSockets.SockSupervisor.start_link()

{:ok, _} = XSockets.TierSupervisor.Task.start_link()

{:ok, _} = XSockets.TierSupervisor.Pool.start_link()

Bounded buffers

Built-in accumulators accept :max_size. Overflow shows up once as {:error, :buffer_overflow, acc}; the Engine emits [:xsockets, :frame_error] and keeps draining.

Telemetry

XSockets.Telemetry.emit/3 plus your own :telemetry handlers. Telemetry.attach_handlers/0 installs log/counter handlers for events the library actually emits (listeners, packets, TLS handshake, rate limit, tiers, send errors).

Common events when :telemetry_enabled is true:

Configuration

Point :config_app at your host app to override defaults:

config :xsockets,

  config_app: :my_app,

  buffer_size: 262_144,

  listener_buffer_size: 4_194_304,

  ssl_handshake_timeout: 10_000,

  rate_limit_enabled: true,

  telemetry_enabled: true



config :my_app,

  buffer_size: 131_072

Lookup order: config :config_app -> config :xsockets -> built-in default.

For TLS/DTLS certs, pass certfile / keyfile in listen opts, or set config :xsockets, certs: [...]. Prefer Transport.TLS.security_opts(:server) or :mutual_tls when building option lists.

Testing

mix compile --warnings-as-errors

mix test

mix dialyzer

CI runs the same three checks.

Optional Linux SCTP proof (when your host has no SCTP):

docker compose -f docker-compose.sctp.yml run --rm xsockets-sctp

Changelog

See CHANGELOG.md.

License

Apache 2.0 -- see LICENSE.md.