Packages
ex_esdb
0.1.1
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.ex
defmodule ExESDB.SnapshotsReader do
@moduledoc """
Provides functions for reading snapshots
"""
def hr_snapshots_reader_name(store_id, source_uuid, stream_uuid),
do: :"#{store_id}_#{source_uuid}_#{stream_uuid}_snaps_rdr"
def cluster_id(store, source_uuid, stream_uuid),
do: {:snaps_reader, store, source_uuid, stream_uuid}
@spec worker_pids(
store :: atom(),
source_uuid :: binary(),
stream_uuid :: binary()
) :: [pid()]
defp worker_pids(store, source_uuid, stream_uuid) do
cluster_id = cluster_id(store, source_uuid, stream_uuid)
Swarm.registered()
|> Enum.filter(fn {name, _} -> match?(^cluster_id, name) end)
|> Enum.map(fn {_, pid} -> pid end)
end
@spec get_worker(
store :: atom(),
source_uuid :: binary(),
stream_uuid :: binary()
) :: pid()
defp get_worker(store, source_uuid, stream_uuid) do
case worker_pids(store, source_uuid, stream_uuid) do
[] ->
start_worker(store, source_uuid, stream_uuid)
worker_pids ->
worker_pids
|> Enum.random()
end
end
defp partition_for(store, source_uuid, stream_uuid) do
partitions = System.schedulers_online()
partition = :erlang.phash2({store, source_uuid, stream_uuid}, partitions)
partition
end
def start_worker(store, source_uuid, stream_uuid) do
partition = partition_for(store, source_uuid, stream_uuid)
case DynamicSupervisor.start_child(
{:via, PartitionSupervisor, {ExESDB.SnapshotsReaders, partition}},
{ExESDB.SnapshotsReaderWorker, {store, source_uuid, stream_uuid, partition}}
) do
{:ok, pid} -> pid
{:error, {:already_started, pid}} -> pid
{:error, reason} -> raise "failed to start snapshots reader: #{inspect(reason)}"
end
end
@doc """
## Description
Reads a snapshot version from the store
for the given source and stream uuids
## Parameters
* `store` - the store to read from
* `source_uuid` - the source uuid
* `stream_uuid` - the stream uuid
* `version` - the version of the snapshot to read
## Returns
* `{:ok, map()}` - the snapshot
"""
@spec read_snapshot(
store :: atom(),
source_uuid :: binary(),
stream_uuid :: binary(),
version :: non_neg_integer()
) :: {:ok, map()} | {:error, term()}
def read_snapshot(store, source_uuid, stream_uuid, version) do
GenServer.call(
get_worker(store, source_uuid, stream_uuid),
{:read_snapshot, store, source_uuid, stream_uuid, version}
)
end
@spec list_snapshots(
store :: atom(),
source_uuid :: binary() | :any,
stream_uuid :: binary() | :any
) :: {:ok, [map()]} | {:error, term()}
def list_snapshots(store, source_uuid \\ :any, stream_uuid \\ :any) do
GenServer.call(
get_worker(store, source_uuid, stream_uuid),
{:list_snapshots, store, source_uuid, stream_uuid}
)
end
end