Packages
ex_esdb
0.7.7
0.11.0
0.10.0
0.9.0
0.8.0
0.7.8
0.7.7
0.7.6
0.7.5
0.7.4
0.7.3
0.7.2
0.7.1
0.7.0
0.6.1
0.6.0
0.5.1
0.5.0
0.4.8
0.4.7
0.4.6
0.4.5
0.4.4
0.4.3
0.4.2
0.4.1
0.4.0
0.3.3
0.3.2
0.3.1
0.3.0
0.2.5
0.2.4
0.2.3
0.2.2
0.2.1
0.2.0
0.1.7
0.1.6
0.1.5
0.1.4
0.1.3
0.1.2
0.1.1
0.1.0
0.0.20
0.0.19
0.0.18
0.0.17
0.0.16
0.0.15
0.0.14-alpha
0.0.13-alpha
0.0.12-alpha
0.0.11-alpha
0.0.10-alpha
0.0.9-alpha
0.0.8-alpha
0.0.6-alpha
0.0.5-alpha
0.0.4-alpha
0.0.3-alpha
0.0.2-alfa
0.0.1-alfa
ExESDB is a reincarnation of rabbitmq/khepri, specialized for use as a BEAM-native event store.
Current section
Files
Jump to
Current section
Files
lib/ex_esdb/snapshots_reader_worker.ex
defmodule ExESDB.SnapshotsReaderWorker do
@moduledoc """
A worker process for reading snapshots from the event store.
"""
use GenServer
alias ExESDB.Snapshots, as: Snapshots
alias ExESDB.SnapshotsReader, as: SnapshotsReader
import ExESDB.Khepri.Conditions
alias ExESDB.Themes, as: Themes
################ PLUMBING ################
def start_link({store, source_uuid, stream_uuid, partition}) do
GenServer.start_link(
__MODULE__,
{store, source_uuid, stream_uuid, partition},
name: SnapshotsReader.hr_snapshots_reader_name(store, source_uuid, stream_uuid)
)
end
@impl true
def init({store, source_uuid, stream_uuid, partition}) do
Process.flag(:trap_exit, true)
cluster_id = SnapshotsReader.cluster_id(store, source_uuid, stream_uuid)
msg = "[#{inspect(self())}] is UP on partition #{inspect(partition)}, joining the cluster."
Swarm.register_name(cluster_id, self())
IO.puts("#{Themes.snapshots_reader_worker(self(), msg)}")
{:ok, {store, source_uuid, stream_uuid, partition}}
end
@impl true
def terminate(reason, {store, source_uuid, stream_uuid, _partition}) do
cluster_id = SnapshotsReader.cluster_id(store, source_uuid, stream_uuid)
IO.puts(
"#{Themes.snapshots_reader_worker(self(), "⚠️ Shutting down gracefully. Reason: #{inspect(reason)}")}"
)
# Unregister from Swarm
Swarm.unregister_name(cluster_id)
:ok
end
################ IMPLEMENTATION ################
@impl true
def handle_call({:read_snapshot, store, source_uuid, stream_uuid, version}, _from, state) do
path = Snapshots.path(source_uuid, stream_uuid, version)
case store
|> :khepri.get(path) do
{:ok, :undefined} ->
{:reply, {:error, :not_found}, state}
{:ok, snapshot} ->
{:reply, {:ok, snapshot}, state}
{:error, reason} ->
{:reply, {:error, reason}, state}
end
end
@impl true
def handle_call({:list_snapshots, store, source_uuid, stream_uuid}, _from, state) do
query = build_query(source_uuid, stream_uuid)
case store
|> :khepri.get_many(query) do
{:ok, snapshots} ->
{:reply, {:ok, snapshots}, state}
{:error, reason} ->
{:reply, {:error, reason}, state}
end
end
defp build_query(:any, :any),
do: [
:snapshots,
if_path_matches(regex: :any),
if_path_matches(regex: :any),
if_path_matches(regex: :any)
]
defp build_query(source_uuid, :any),
do: [
:snapshots,
source_uuid,
if_path_matches(regex: :any),
if_path_matches(regex: :any)
]
defp build_query(source_uuid, stream_uuid),
do: [:snapshots, source_uuid, stream_uuid, if_path_matches(regex: :any)]
end