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.