Phoenix.SocketClient.MessageProcessor (phoenix_socket_client v0.8.1)

View Source

High-performance asynchronous message processor for Phoenix Socket Client.

This module provides efficient, non-blocking JSON encoding/decoding operations with message batching, backpressure, and memory management for high-throughput scenarios.

Features

  • Synchronous mode (default) for direct JSON encoding/decoding
  • Asynchronous mode with message batching for high-throughput scenarios
  • Bounded queues with backpressure (async mode)
  • Comprehensive monitoring and telemetry

Sync Mode (Default)

By default, :sync_mode is true. In this mode, encode/4 and decode/4 call JSON.encode!/1 and JSON.decode!/1 directly without going through the worker pool, batching, or binary pooling. This is the recommended mode for most use cases since JSON encoding/decoding takes microseconds.

Async Mode

Set :sync_mode to false to enable the async batch processing pipeline with worker pools, message batching, and binary pooling. This may be useful for extremely high-throughput scenarios.

Usage

The processor is started automatically by the Socket supervisor and handles all message serialization/deserialization operations transparently.

Summary

Functions

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

decode(processor, raw_message, caller \\ self(), ref \\ make_ref())

Decodes a message.

encode(processor, message, caller \\ self(), ref \\ make_ref())

Encodes a message.

start_link(opts)

Starts the message processor with the given options.

stats(processor)

Gets current processing statistics.

Functions

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

See Supervisor.

decode(processor, raw_message, caller \\ self(), ref \\ make_ref())

@spec decode(pid(), String.t(), pid(), reference()) :: :ok | {:error, term()}

Decodes a message.

In sync mode (default), decodes directly and sends the result to the caller. In async mode, enqueues for batch processing by the worker pool.

Returns :ok on success. The decoded result is sent as {:decode_result, ref, {:ok, decoded}} to the caller process.

encode(processor, message, caller \\ self(), ref \\ make_ref())

@spec encode(pid(), Phoenix.SocketClient.Message.t(), pid(), reference()) ::
  :ok | {:error, term()}

Encodes a message.

In sync mode (default), encodes directly and sends the result to the caller. In async mode, enqueues for batch processing by the worker pool.

Returns :ok on success. The encoded result is sent as {:encode_result, ref, {:ok, encoded}} to the caller process.

start_link(opts)

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

Starts the message processor with the given options.

Options

  • :serializer - Message serializer module (required)
  • :json_library - JSON library module (defaults to JSON)
  • :registry_name - Registry name for channel lookup
  • :sync_mode - When true (default), encode/decode synchronously without worker pool
  • :batch_size - Maximum messages per batch (default: 20, async mode only)
  • :batch_interval - Batch timeout in ms (default: 10, async mode only)
  • :max_queue_size - Maximum queue size before backpressure (default: 1000, async mode only)
  • :concurrency - Number of worker processes (default: 8, async mode only)

stats(processor)

@spec stats(pid()) :: map()

Gets current processing statistics.