Packages
eventstore
0.2.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 from the store.
"""
use GenServer
require Logger
alias EventStore.Subscriptions.{PersistentSubscription,Subscription}
defstruct stream_uuid: nil, subscription_name: nil, subscriber: nil, subscription: nil
def start_link(stream_uuid, subscription_name, subscriber) do
GenServer.start_link(__MODULE__, %Subscription{
stream_uuid: stream_uuid,
subscription_name: subscription_name,
subscriber: subscriber,
subscription: PersistentSubscription.new
})
end
def notify_events(subscription, events) do
GenServer.cast(subscription, {:notify_events, events})
end
def init(%Subscription{subscriber: subscriber} = state) do
Process.link(subscriber)
GenServer.cast(self, {:subscribe})
{:ok, state}
end
def handle_cast({:subscribe}, %Subscription{stream_uuid: stream_uuid, subscription_name: subscription_name, subscriber: subscriber, subscription: subscription} = state) do
subscription =
subscription
|> PersistentSubscription.subscribe(stream_uuid, subscription_name, subscriber)
|> PersistentSubscription.catch_up
{:noreply, %Subscription{state | subscription: subscription}}
end
def handle_cast({:notify_events, events}, %Subscription{subscription: subscription} = state) do
subscription =
subscription
|> PersistentSubscription.notify_events(events)
{:noreply, %Subscription{state | subscription: subscription}}
end
end