View Source EchoPubSub.Producer (EchoPubSub v0.2.0)

Sends this node's broadcasts to the workers on other nodes, one producer per pool group.

Every broadcast is written to a ring buffer of the last buffer_size messages. For each remote node the producer keeps a read cursor: the position up to which that node has acknowledged. A flush sends each node the messages from its cursor to the write cursor, and the cursor moves only when the node's worker replies :ok, so a failed batch is sent again on the next flush.

Remote calls run in tasks, so a slow or hung node never blocks broadcasts. A node is registered before its first batch:

  • {:register, node} when the producer has no cursor for it (new node, forgotten node, or this producer restarted), so a worker that knew this producer reports {:cursor_expired, node}
  • {:resume, node} when the cursor survived and only the worker restarted

A node more than buffer_size messages behind gets {:expired, node} instead of messages it can no longer be sent. Local subscribers never go through the producer: Phoenix delivers local broadcasts directly.

Summary

Types

Producer options; all are required. See EchoPubSub for what each one does.

Functions

Writes message to the group's buffer; it is sent to remote nodes on the next flush. Waits only for the local producer, never for remote nodes.

Returns a specification to start this module under a supervisor.

The registered name of the group's producer.

Starts the producer for opts[:group], registered under name/1.

Types

@type group() :: atom()
@type option() ::
  {:group, group()}
  | {:buffer_size, pos_integer()}
  | {:batch_interval, non_neg_integer()}
  | {:call_timeout, timeout()}
  | {:capacity_warning_threshold, number()}
  | {:capacity_warning_interval, non_neg_integer()}

Producer options; all are required. See EchoPubSub for what each one does.

Functions

Link to this function

buffer_and_send(group, message)

View Source
@spec buffer_and_send(group(), term()) :: :ok

Writes message to the group's buffer; it is sent to remote nodes on the next flush. Waits only for the local producer, never for remote nodes.

Returns a specification to start this module under a supervisor.

See Supervisor.

@spec name(group()) :: atom()

The registered name of the group's producer.

@spec start_link([option()]) :: GenServer.on_start()

Starts the producer for opts[:group], registered under name/1.