Specs

subscribe_to_all_streams(
  subscription_name :: String.t(),
  subscriber :: pid(),
  opts :: persistent_subscription_options()
) ::
  {:ok, subscription :: pid()}
  | {:error, :already_subscribed}
  | {:error, :subscription_already_exists}
  | {:error, :too_many_subscribers}
  | {:error, reason :: term()}

Create a subscription to all streams. By default the subscription is persistent.

The subscriber process will be notified of each batch of events appended to any stream.

• subscription_name is used to uniquely identify the subscription.

• subscriber is a process that will be sent {:events, events} notification messages.

• opts is an optional map providing additional subscription configuration:

  • name the name of the event store if provided to start_link/1.
  • start_from is a pointer to the first event to receive. It must be one of:
    • :origin for all events from the start of the stream (default).
    • :current for any new events appended to the stream after the subscription has been created.
    • any positive integer for an event id to receive events after that exact event.
  • selector to define a function to filter each event, i.e. returns only those elements for which fun returns a truthy value
  • mapper to define a function to map each recorded event before sending to the subscriber.
  • concurrency_limit defines the maximum number of concurrent subscribers allowed to connect to the subscription. By default only one subscriber may connect. If too many subscribers attempt to connect to the subscription an {:error, :too_many_subscribers} is returned.
  • transient is an optional boolean flag to create a transient subscription. See subscribe_to_stream for the full information.

The subscription will resume from the last acknowledged event if it already exists. It will ignore the start_from argument in this case.

Returns {:ok, subscription} when subscription succeeds.

Example

{:ok, subscription} = EventStore.subscribe_to_all_streams("all_subscription", self())

# wait for the subscription confirmation
receive do
  {:subscribed, ^subscription} ->
    IO.puts "Successfully subscribed to all streams"
end

receive do
  {:events, events} ->
    IO.puts "Received events: " <> inspect(events)

    # acknowledge receipt
    EventStore.ack(subscription, events)
end