Packages
ex_esdb
0.2.5
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_writer.ex
defmodule ExESDB.SnapshotsWriter do
@moduledoc """
The API for interacting with ExESDB Snapshots Writers.
It functions as an API for SnapshotsWriterWorkers,
by requesting a worker from the Cluster.
If no worker is available for the specific combination of store, source_uuid, and stream_uuid,
then a new worker is started.
"""
def hr_snapshots_writer_name(store_id, source_uuid, stream_uuid),
do: :"snapshots_writer_#{store_id}_#{source_uuid}_#{stream_uuid}"
def cluster_id(store, source_uuid, stream_uuid),
do: {:snapshots_writer, 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()
key = :erlang.phash2({store, source_uuid, stream_uuid}, partitions)
key
end
defp start_worker(store, source_uuid, stream_uuid) do
partition = partition_for(store, source_uuid, stream_uuid)
case DynamicSupervisor.start_child(
{:via, PartitionSupervisor, {ExESDB.SnapshotsWriters, partition}},
{ExESDB.SnapshotsWriterWorker, {store, source_uuid, stream_uuid, partition}}
) do
{:ok, pid} -> pid
{:error, {:already_started, pid}} -> pid
{:error, reason} -> raise "failed to start snapshots writer: #{inspect(reason)}"
end
end
@spec record_snapshot(
store :: atom(),
source_uuid :: binary(),
stream_uuid :: binary(),
version :: integer(),
snapshot_record :: map()
) :: :ok
def record_snapshot(store, source_uuid, stream_uuid, version, snapshot_record) do
GenServer.cast(
get_worker(store, source_uuid, stream_uuid),
{:record_snapshot, store, source_uuid, stream_uuid, version, snapshot_record}
)
:ok
end
@spec delete_snapshot(
store :: atom(),
source_uuid :: binary(),
stream_uuid :: binary(),
version :: integer()
) :: :ok
def delete_snapshot(store, source_uuid, stream_uuid, version) do
GenServer.cast(
get_worker(store, source_uuid, stream_uuid),
{:delete_snapshot, store, source_uuid, stream_uuid, version}
)
:ok
end
end