Sourced.EventStore.Subscription (sourced v0.1.0)

Copy Markdown View Source

Module for storing subscription information for an Event Store adapter.

Provides a deliver/2 convenience function that can be used by the adapter to send filtered events to the subscriber.

Knowing when a subscription ends

Delivery is at-most-once and a subscription is not durable, so a subscriber that must not miss events has to notice when its subscription ends and re-subscribe from the sequence it got to. :owner is what makes that possible: the process delivering the events, which the subscriber can Process.monitor/1.

{:ok, subscription} = Sourced.EventStore.subscribe(store, from: cursor)
Process.monitor(subscription.owner)

If the owner goes down, the subscription is gone and nothing further will be delivered on it. The subscriber owns its own cursor — it is the only process that knows which events it actually handled — so it resubscribes from there:

def handle_info({:sourced_events, ref, events}, %{ref: ref} = state) do
  {:noreply, %{state | cursor: List.last(events).sequence}}
end

def handle_info({:DOWN, _monitor_ref, :process, _pid, _reason}, state) do
  {:ok, subscription} = Sourced.EventStore.subscribe(store, from: state.cursor + 1)
  Process.monitor(subscription.owner)

  {:noreply, %{state | ref: subscription.ref}}
end

The implication runs one way: a dead owner means a dead subscription, but a live one is not a promise that delivery is healthy. What the monitor rules out is the case that has no other signal at all — delivery stopping with nothing sent to say so.

A :DOWN also arrives for an ordinary Sourced.EventStore.unsubscribe/2 on adapters whose owner is per-subscription, so a subscriber that resubscribes on :DOWN should demonitor when it unsubscribes deliberately.

Summary

Types

t()

:subscriber is the process events are delivered to; :owner is the process delivering them, whose death means the subscription is over. :owner is filled in by the adapter, so it is nil until Sourced.EventStore.Behaviour.subscribe/2 has returned.

Functions

Sends stored_events to the subscriber.

Generates a helper struct for handling a subscription.

Types

opts()

@type opts() :: [
  subscriber: pid(),
  query: Sourced.EventStore.Query.t() | nil,
  from: non_neg_integer()
]

t()

@type t() :: %Sourced.EventStore.Subscription{
  from: non_neg_integer(),
  owner: pid() | nil,
  query: Sourced.EventStore.Query.t() | nil,
  ref: reference(),
  store: Sourced.EventStore.Behaviour.store(),
  subscriber: pid()
}

:subscriber is the process events are delivered to; :owner is the process delivering them, whose death means the subscription is over. :owner is filled in by the adapter, so it is nil until Sourced.EventStore.Behaviour.subscribe/2 has returned.

Functions

deliver(subscription, stored_events)

@spec deliver(subscription :: t(), stored_events :: [Sourced.StoredEvent.t()]) :: :ok

Sends stored_events to the subscriber.

Filtering must be handled by the adapter prior to calling this function. The function transforms the events using the same middleware pipeline that was configured for the event store.

new(store, opts \\ [])

@spec new(store :: Sourced.EventStore.Behaviour.store(), opts :: opts()) :: t()

Generates a helper struct for handling a subscription.

Options

  • :subscriberrequired the process to deliver to.
  • :query — the Query to match live events against.
  • :from — the sequence live delivery starts at, inclusive.