Xirsys.Sockets.Accumulator behaviour (xturn_sockets v2.2.1)

View Source

Per-tier packet framing state machine.

Built-in implementations: Xirsys.Sockets.Accumulator.Raw, Xirsys.Sockets.Accumulator.LengthPrefixed, and Xirsys.Sockets.Accumulator.Reorder.

What problem this solves

Stream and datagram transports deliver arbitrary byte chunks. Before a handler runs, something must decide when a logical packet is complete. Each pipeline tier owns an accumulator that buffers input via push/3 and yields whole packets from pop/1.

RFCs

Accumulators implement wire framing, not STUN/TURN semantics:

Choose LengthPrefixed or Raw to match the tier's on-the-wire layout.

Bounded buffers

Every implementation should accept a :max_size option that limits how much data or how many whole packets may be held. Because push/3 returns only acc(), overflow is typically recorded during push/3 (drop oldest, set a flag) and surfaced once from the next pop/1 as {:error, :buffer_overflow, acc}. The engine emits :frame_error telemetry and continues draining after overflow.

Summary

Types

acc()

Opaque accumulator state. Shape is private to the implementation.

meta()

Per-packet metadata merged across push/3 calls until a packet is popped.

Callbacks

init(keyword)

Builds initial accumulator state.

pop(acc)

Extracts one whole packet, or reports that more data is needed.

push(acc, binary, meta)

Appends chunk and meta to the accumulator. Does not extract packets.

Types

acc()

@type acc() :: term()

Opaque accumulator state. Shape is private to the implementation.

meta()

@type meta() :: map()

Per-packet metadata merged across push/3 calls until a packet is popped.

Callbacks

init(keyword)

@callback init(keyword()) :: acc()

Builds initial accumulator state.

Parameters

  • opts - implementation-specific keyword list (:max_size, :header_size, ...)

pop(acc)

@callback pop(acc()) ::
  {:ok, binary(), meta(), acc()} | {:more, acc()} | {:error, term(), acc()}

Extracts one whole packet, or reports that more data is needed.

Parameters

  • acc - current state

Returns

  • {:ok, packet, meta, acc} - one complete packet
  • {:more, acc} - incomplete; wait for another push/3
  • {:error, reason, acc} - framing error (:buffer_overflow, ...); drain continues

push(acc, binary, meta)

@callback push(acc(), binary(), meta()) :: acc()

Appends chunk and meta to the accumulator. Does not extract packets.

Parameters

  • acc - current state from init/1 or a previous push/3 / pop/1
  • chunk - inbound bytes (one datagram, or a stream read)
  • meta - metadata merged into the next popped packet