Packages
ex_esdb
0.7.0
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_worker.ex
defmodule ExESDB.SnapshotsWriterWorker do
@moduledoc """
A worker process for writing snapshots to the event store.
"""
use GenServer
alias ExESDB.Snapshots, as: Snapshots
alias ExESDB.SnapshotsWriter, as: SnapshotsWriter
alias ExESDB.Themes, as: Themes
################ PLUMBING ################
@doc """
Starts a new `ExESDB.SnapshotsWriterWorker` process.
"""
def start_link({store, source_uuid, stream_uuid, partition}) do
GenServer.start_link(
__MODULE__,
{store, source_uuid, stream_uuid, partition},
name: SnapshotsWriter.hr_snapshots_writer_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 = SnapshotsWriter.cluster_id(store, source_uuid, stream_uuid)
Swarm.register_name(cluster_id, self())
msg = "[#{inspect(self())}] is UP on partition #{inspect(partition)}, joining the cluster."
IO.puts("#{Themes.snapshots_writer_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 = SnapshotsWriter.cluster_id(store, source_uuid, stream_uuid)
IO.puts(
"#{Themes.snapshots_writer_worker(self(), "⚠️ Shutting down gracefully. Reason: #{inspect(reason)}")}"
)
# Unregister from Swarm
Swarm.unregister_name(cluster_id)
:ok
end
################ IMPLEMENTATION ################
@impl true
def handle_cast({:delete_snapshot, store, source_uuid, stream_uuid, version}, state) do
path = Snapshots.path(source_uuid, stream_uuid, version)
store
|> :khepri.delete!(path)
# Request asynchronous persistence
spawn(fn -> ExESDB.PersistenceWorker.request_persistence(store) end)
{:noreply, state}
end
@impl true
def handle_cast(
{:record_snapshot, store, source_uuid, stream_uuid, version, snapshot_record},
state
) do
path = Snapshots.path(source_uuid, stream_uuid, version)
store
|> :khepri.put!(path, snapshot_record)
# Request asynchronous persistence
spawn(fn -> ExESDB.PersistenceWorker.request_persistence(store) end)
{:noreply, state}
end
end