Packages
eventstore
0.5.2
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.{StreamSubscription,Subscription}
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: StreamSubscription.new
})
end
def notify_events(subscription, events) when is_list(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
|> StreamSubscription.subscribe(stream_uuid, stream, subscription_name, self, subscriber, [])
handle_subscription_state(subscription.state)
{:noreply, %Subscription{state | subscription: subscription}}
end
def handle_cast({:notify_events, events}, %Subscription{subscription: subscription} = state) do
subscription =
subscription
|> StreamSubscription.notify_events(events)
handle_subscription_state(subscription.state)
{:noreply, %Subscription{state | subscription: subscription}}
end
def handle_cast({:catch_up}, %Subscription{subscription: subscription} = state) do
subscription =
subscription
|> StreamSubscription.catch_up
handle_subscription_state(subscription.state)
{:noreply, %Subscription{state | subscription: subscription}}
end
@doc """
Confirm receipt of an event by id
"""
def handle_info({:ack, last_seen_event_id}, %Subscription{subscription: subscription} = state) do
subscription =
subscription
|> StreamSubscription.ack(last_seen_event_id)
handle_subscription_state(subscription.state)
{:noreply, %Subscription{state | subscription: subscription}}
end
def handle_call({:unsubscribe}, _from, %Subscription{subscriber: subscriber, subscription: subscription} = state) do
Process.unlink(subscriber)
subscription =
subscription
|> StreamSubscription.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
end