View Source EchoPubSub.Producer.State (EchoPubSub v0.2.0)
The producer's state and the only place that changes it. Other modules read fields directly but update them through the named functions here.
Summary
Functions
Records that node acked every message below target.
Acks every write for this node: Phoenix delivered them to local subscribers already.
Whether node has not acked every write yet.
Whether node is further behind than the buffer can replay.
Clears the flush timer once the flush runs.
The registration or delivery a task ref belongs to. Every ref is in exactly
one map: registrations and deliveries never share a ref.
Drops every trace of nodes, so they register from scratch if they return.
A fresh state for group: empty buffer, no known nodes.
What registered node still needs from the buffer.
Marks node registered with pid, keeping its cursor if it had one.
Keeps a failed registration's start cursor for the retry; no task runs meanwhile.
Whether a registration task for node is running (not a failed one waiting to retry).
The start cursor a failed registration of node retries from, or nil.
Records the pending flush timer.
Records when the last capacity warning fired.
Records a batch in flight to node, acking up to target once it succeeds.
Records a running registration task for node.
Writes the slowest node has not acked yet.
Forgets the batch in flight to node.
Stores message at the write cursor and moves the cursor on.
Types
@type cursor() :: non_neg_integer()
@type t() :: %EchoPubSub.Producer.State{ batch_interval: timeout(), buffer: EchoPubSub.Producer.RingBuffer.t(), call_timeout: timeout(), capacity_warning_interval: non_neg_integer(), capacity_warning_threshold: number(), flush_timer: (reference() | nil) | nil, group: atom(), in_flight: %{required(node()) => {reference(), cursor()}}, last_capacity_warning_at: (integer() | nil) | nil, read_cursors: %{required(node()) => cursor()}, registering: %{required(node()) => {reference() | nil, cursor(), pid()}}, workers: %{required(node()) => pid()}, write_cursor: cursor() }
Fields:
group- the:pggroup this producer delivers to (its delivery scope)write_cursor- count of messages ever written; the next write lands hereread_cursors- map of node to its next-needed cursor; everything below it that node has acked. A node is absent until its registration is ackedregistering- map of node to the{task reference, start cursor, worker pid}of its pending registration; the reference isnilwhile a failed attempt waits for its retryworkers- map of node to the worker pid this producer last registered within_flight- map of node to the{task reference, target cursor}of the batch being delivered to itbuffer- ring buffer of the lastbuffer_sizemessages; oldest overwritten on wrapbatch_interval- milliseconds to batch writes before a flush;0flushes immediatelycall_timeout- timeout in milliseconds for synchronous calls to remote nodescapacity_warning_threshold- buffer fill ratio that triggers a capacity warningcapacity_warning_interval- minimum seconds between capacity warningsflush_timer- reference of the pending flush timer, ornilwhen none is scheduledlast_capacity_warning_at- monotonic seconds of the last capacity warning, ornilif never
Functions
Records that node acked every message below target.
Acks every write for this node: Phoenix delivered them to local subscribers already.
Whether node has not acked every write yet.
Whether node is further behind than the buffer can replay.
Clears the flush timer once the flush runs.
@spec find_task(t(), reference()) :: {:registration, node(), cursor(), pid()} | {:delivery, node(), cursor()} | :unknown
The registration or delivery a task ref belongs to. Every ref is in exactly
one map: registrations and deliveries never share a ref.
Drops every trace of nodes, so they register from scratch if they return.
@spec new(atom(), [EchoPubSub.Producer.option()]) :: t()
A fresh state for group: empty buffer, no known nodes.
@spec pending(t(), node()) :: EchoPubSub.Producer.RingBuffer.pending()
What registered node still needs from the buffer.
Marks node registered with pid, keeping its cursor if it had one.
Keeps a failed registration's start cursor for the retry; no task runs meanwhile.
Whether a registration task for node is running (not a failed one waiting to retry).
The start cursor a failed registration of node retries from, or nil.
Records the pending flush timer.
Records when the last capacity warning fired.
Records a batch in flight to node, acking up to target once it succeeds.
Records a running registration task for node.
@spec unacked(t()) :: non_neg_integer()
Writes the slowest node has not acked yet.
Forgets the batch in flight to node.
Stores message at the write cursor and moves the cursor on.