Packages
eventstore
0.4.0
1.4.8
1.4.7
1.4.6
1.4.4
1.4.3
1.4.2
1.4.1
1.4.0
1.3.2
1.3.1
1.3.0
1.2.3
1.2.2
1.2.1
1.2.0
1.1.0
1.0.3
1.0.2
1.0.1
1.0.0
1.0.0-rc.0
0.17.0
0.16.2
0.16.1
0.16.0
0.15.1
0.15.0
0.14.0
0.14.0-rc.0
0.13.2
0.13.1
0.13.0
0.12.1
0.12.0
0.11.0
0.11.0-rc.0
0.10.1
0.10.0
0.9.0
0.8.1
0.8.0
0.7.4
0.7.3
0.7.2
0.7.1
0.7.0
0.6.2
0.6.1
0.6.0
0.5.2
0.5.1
0.5.0
0.4.3
0.4.2
0.4.1
0.4.0
0.3.0
0.2.1
0.2.0
0.1.0
0.0.5
0.0.4
0.0.3
0.0.2
0.0.1
Event store using PostgreSQL for persistence.
Current section
Files
Jump to
Current section
Files
lib/event_store/subscriptions/subscription.ex
defmodule EventStore.Subscriptions.Subscription do
@moduledoc """
Subscription to a single, or all, event streams.
A subscription is persistent so that resuming the subscription will continue from the last acknowledged event.
This guarantees at least once delivery of every event appended to storage.
"""
use GenServer
require Logger
alias EventStore.Subscriptions.{AllStreamsSubscription,SingleStreamSubscription,Subscription}
@all_stream "$all"
defstruct stream_uuid: nil, stream: nil, subscription_name: nil, subscriber: nil, subscription: nil
def start_link(stream_uuid, stream, subscription_name, subscriber) do
GenServer.start_link(__MODULE__, %Subscription{
stream_uuid: stream_uuid,
stream: stream,
subscription_name: subscription_name,
subscriber: subscriber,
subscription: subscription_provider(stream_uuid).new
})
end
def notify_events(subscription, events) do
GenServer.cast(subscription, {:notify_events, events})
end
def unsubscribe(subscription) do
GenServer.call(subscription, {:unsubscribe})
end
def init(%Subscription{subscriber: subscriber} = state) do
Process.link(subscriber)
GenServer.cast(self, {:subscribe_to_stream})
{:ok, state}
end
def handle_cast({:subscribe_to_stream}, %Subscription{stream_uuid: stream_uuid, stream: stream, subscription_name: subscription_name, subscriber: subscriber, subscription: subscription} = state) do
subscription =
subscription
|> subscription_provider(stream_uuid).subscribe(stream_uuid, stream, subscription_name, subscriber)
handle_subscription_state(subscription.state)
{:noreply, %Subscription{state | subscription: subscription}}
end
def handle_cast({:notify_events, events}, %Subscription{stream_uuid: stream_uuid, subscription: subscription} = state) do
subscription =
subscription
|> subscription_provider(stream_uuid).notify_events(events)
handle_subscription_state(subscription.state)
{:noreply, %Subscription{state | subscription: subscription}}
end
def handle_cast({:catch_up}, %Subscription{stream_uuid: stream_uuid, subscription: subscription} = state) do
subscription =
subscription
|> subscription_provider(stream_uuid).catch_up
handle_subscription_state(subscription.state)
{:noreply, %Subscription{state | subscription: subscription}}
end
def handle_call({:unsubscribe}, _from, %Subscription{stream_uuid: stream_uuid, subscriber: subscriber, subscription: subscription} = state) do
Process.unlink(subscriber)
subscription =
subscription
|> subscription_provider(stream_uuid).unsubscribe
handle_subscription_state(subscription.state)
{:reply, :ok, %Subscription{state | subscription: subscription}}
end
defp handle_subscription_state(:catching_up) do
GenServer.cast(self, {:catch_up})
end
defp handle_subscription_state(_) do
# no-op
end
defp subscription_provider(@all_stream) do
AllStreamsSubscription
end
defp subscription_provider(_stream_uuid) do
SingleStreamSubscription
end
end