ExESDB.Commanded.AggregateListener (ex_esdb_commanded v0.9.0)

A process that subscribes to the EventStore's Phoenix PubSub <store>:$all topic and filters events by stream_id for aggregate transient subscriptions.

Simplified with Swarm for process distribution.

Each AggregateListener processes:

  1. Subscribes to the <store>:$all Phoenix PubSub topic
  2. Filters incoming events based on the target stream_id
  3. Transforms ExESDB.Schema.EventRecord to Commanded.EventStore.RecordedEvent
  4. Forwards matching events to the subscriber process

Summary

Types

listener_config()

Functions

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

get_pid(store_id, stream_id)

Gets the PID of the AggregateListener for a given store_id and stream_id.

start(config)

Starts an AggregateListener without linking to the calling process. Useful for transient subscriptions where we don't want to crash the caller.

start_link(config)

Starts an AggregateListener for the given stream using Swarm.

stop(pid)

Stops the AggregateListener process.

swarm_key(store_id, stream_id)

Types

listener_config()

@type listener_config() :: %{
  store_id: atom(),
  stream_id: String.t(),
  subscriber: pid(),
  replay_historical_events?: boolean()
}

Functions

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

See Supervisor.

get_pid(store_id, stream_id)

@spec get_pid(atom(), String.t()) :: pid() | nil

Gets the PID of the AggregateListener for a given store_id and stream_id.

start(config)

@spec start(listener_config()) :: {:ok, pid()} | {:error, term()}

Starts an AggregateListener without linking to the calling process. Useful for transient subscriptions where we don't want to crash the caller.

start_link(config)

@spec start_link(listener_config()) :: {:ok, pid()} | {:error, term()}

Starts an AggregateListener for the given stream using Swarm.

Parameters

  • config: Map containing:
    • store_id: The EventStore identifier
    • stream_id: The target stream to filter events for
    • subscriber: The process to send filtered events to
    • replay_historical_events?: Whether to replay historical events on startup (defaults to true)

stop(pid)

@spec stop(pid()) :: :ok

Stops the AggregateListener process.

swarm_key(store_id, stream_id)