Current section

Files

Jump to
extreme_system lib system message_handler.ex
Raw

lib/system/message_handler.ex

defmodule Extreme.System.MessageHandler do
defmacro __using__(opts) do
quote do
require Logger
alias Extreme.System.EventStore
alias Extreme.System.AggregatePidFacade, as: PidFacade
@aggregate_mod Keyword.fetch!(unquote(opts), :aggregate_mod)
@prefix Keyword.fetch!(unquote(opts), :prefix)
@es Extreme.System.EventStore.name(@prefix)
@pid_facade Extreme.System.AggregatePidFacade.name(@aggregate_mod)
defp aggregate_mod, do: @aggregate_mod
defp prefix, do: @prefix
defp with_new_aggregate(log_msg, cmd, id \\ nil, fun) do
Logger.info fn -> log_msg end
key = id || UUID.uuid1
with {:ok, pid} <- PidFacade.spawn_new(@pid_facade, key),
{:ok, transaction, events, version} <- fun.({:ok, pid, key}),
{:ok, last_event} <- apply_changes(pid, key, transaction, events, version)
do
{:created, key, last_event}
else
any ->
Logger.warn fn -> "New aggregate creation failed: #{inspect any}" end
PidFacade.exit_process @pid_facade, key, {:creation_failed, any}
any
end
end
defp with_aggregate(log_msg, id, fun) do
Logger.info fn -> log_msg end
with {:ok, pid} <- get_pid(id),
{:ok, transaction, events, version} <- fun.({:ok, pid})
do
apply_changes(pid, id, transaction, events, version)
else
any ->
Logger.info fn -> "Nothing to commit: #{inspect any}" end
any
end
end
defp with_aggregate(:no_commit, log_msg, id, fun) do
Logger.info fn -> log_msg end
result = case get_pid(id) do
{:ok, pid} ->
case fun.({:ok, pid}) do
{:ok, transaction, events, version} ->
{:ok, fn -> apply_changes(pid, id, transaction, events, version) end}
any ->
Logger.info fn -> "Nothing to commit: #{inspect any}" end
any
end
other -> other
end
end
defp get_pid(id),
do: PidFacade.get_pid(@pid_facade, id)
defp apply_changes(aggregate, _, transaction, [], expected_version) do
:ok = aggregate_mod().commit aggregate, transaction, expected_version
Logger.info fn -> "No events to commit" end
{:ok, expected_version}
end
defp apply_changes(aggregate, key, transaction, events, expected_version) do
case EventStore.save_events(@es, {aggregate_mod(), key}, events, Logger.metadata, expected_version) do
{:ok, last_event_number} ->
:ok = aggregate_mod().commit aggregate, transaction, expected_version, last_event_number
Logger.info fn -> "Successfull commit of events. New aggregate version: #{last_event_number}" end
{:ok, last_event_number}
error ->
Logger.error fn -> "Error saving events #{inspect error}" end
Process.exit aggregate, :kill
error
end
end
end
end
end