Repository: commanded/commanded
Pull request #293 (2019-07-11 16:26:59)

Reset event handler
In progress

see #120 

Signed-off-by: Freyskeyd <simon.paitrault@gmail.com>


Patch:
diff --git a/guides/Events.md b/guides/Events.md
index 9a6bb49..76da376 100644
--- a/guides/Events.md
+++ b/guides/Events.md
@@ -238,3 +238,26 @@ defimpl Commanded.Event.Upcaster, for: HistoricalEvent do
   end
 end
 ```
+
+## Reset an EventHandler
+
+An event handler can be reset (using mix task), it will restart the event store subscription from the configured
+`start_from`. This allow an individual handler to be restart while the app is still running.
+
+You can define a `before_reset/1` method that will be called before resetting the event handler.
+
+```elixir
+defmodule ExampleHandler do
+  use Commanded.Event.Handler, name: __MODULE__
+
+  require Logger
+
+  alias Commanded.Event.FailureContext
+
+  def before_reset do
+    # Do something
+    :ok
+  end
+
+end
+```
diff --git a/lib/commanded/event/handler.ex b/lib/commanded/event/handler.ex
index 2d72e53..b3d327e 100644
--- a/lib/commanded/event/handler.ex
+++ b/lib/commanded/event/handler.ex
@@ -302,7 +302,10 @@ defmodule Commanded.Event.Handler do
       @doc false
       def init, do: :ok
 
-      defoverridable init: 0
+      @doc false
+      def before_reset, do: :ok
+
+      defoverridable init: 0, before_reset: 0
     end
   end
 
@@ -399,6 +402,31 @@ defmodule Commanded.Event.Handler do
     {:noreply, subscribe_to_events(state)}
   end
 
+  @doc false
+  def handle_info(:reset, %Handler{} = state) do
+    %Handler{
+      handler_module: handler_module
+    } = state
+
+    case handler_module.before_reset() do
+      :ok ->
+        try do
+          state = state |> reset_subscription() |> subscribe_to_events()
+
+          {:noreply, state}
+        catch
+          {:error, reason} ->
+            {:stop, reason, state}
+        end
+
+      {:stop, reason} ->
+        Logger.debug(fn ->
+          describe(state) <> " `before_reset/0` callback has requested to stop. (reason: #{inspect(reason)})"
+        end)
+        {:stop, reason, state}
+    end
+  end
+
   @doc false
   # Subscription to event store has successfully subscribed, init event handler
   def handle_info({:subscribed, subscription}, %Handler{subscription: subscription} = state) do
@@ -453,6 +481,22 @@ defmodule Commanded.Event.Handler do
     Subscriptions.register(name, consistency)
   end
 
+  defp reset_subscription(%Handler{} = state) do
+    %Handler{
+      handler_name: handler_name,
+      subscribe_to: subscribe_to,
+      subscription_ref: subscription_ref,
+      subscription: subscription,
+    } = state
+
+    Process.demonitor(subscription_ref)
+
+    :ok = EventStore.unsubscribe(subscription)
+    :ok = EventStore.delete_subscription(subscribe_to, handler_name)
+
+    %Handler{state | last_seen_event: 0, subscription: nil, subscription_ref: nil}
+  end
+
   defp subscribe_to_events(%Handler{} = state) do
     %Handler{
       handler_name: handler_name,
diff --git a/lib/mix/tasks/commanded.reset.ex b/lib/mix/tasks/commanded.reset.ex
new file mode 100644
index 0000000..caedba4
--- /dev/null
+++ b/lib/mix/tasks/commanded.reset.ex
@@ -0,0 +1,36 @@
+defmodule Mix.Tasks.Commanded.Reset do
+  @moduledoc """
+  Reset an event handler.
+
+  ## Examples
+      mix commanded.reset MyApp.MyEventHandler
+  """
+
+  use Mix.Task
+  # import Mix.EventStore
+
+  # alias EventStore.Tasks.Init
+
+  @shortdoc "Reset an event handler to its start_from"
+
+  @switches [
+  ]
+
+  @aliases [
+  ]
+
+  @doc false
+  def run(args) do
+    module = args |> List.first
+
+    {:ok, _} = Application.ensure_all_started(:commanded)
+
+    case Commanded.Registration.whereis_name({Commanded.Event.Handler, module}) do
+      :undefined ->
+        IO.puts("No process found for #{module}")
+      pid ->
+        IO.puts("Resetting #{module}")
+        send(pid, :reset)
+    end
+  end
+end
diff --git a/mix.exs b/mix.exs
index 09bbd7e..969acaa 100644
--- a/mix.exs
+++ b/mix.exs
@@ -13,6 +13,7 @@ defmodule Commanded.Mixfile do
       description: description(),
       docs: docs(),
       package: package(),
+      aliases: aliases(),
       build_embedded: Mix.env() == :prod,
       start_permanent: Mix.env() == :prod,
       consolidate_protocols: Mix.env() == :prod,
@@ -117,4 +118,9 @@ defmodule Commanded.Mixfile do
       }
     ]
   end
+
+  defp aliases do
+    [
+    ]
+  end
 end


Test patch:
diff --git a/test/event/handle_event_test.exs b/test/event/handle_event_test.exs
index e80c9e7..eb4b346 100644
--- a/test/event/handle_event_test.exs
+++ b/test/event/handle_event_test.exs
@@ -8,7 +8,7 @@ defmodule Commanded.Event.HandleEventTest do
   alias Commanded.Event.{AppendingEventHandler, UninterestingEvent}
   alias Commanded.Helpers.EventFactory
   alias Commanded.Helpers.{ProcessHelper, Wait}
-  alias Commanded.ExampleDomain.BankAccount.AccountBalanceHandler
+  alias Commanded.ExampleDomain.BankAccount.{BankAccountHandler, AccountBalanceHandler}
   alias Commanded.ExampleDomain.BankAccount.Events.{BankAccountOpened, MoneyDeposited}
 
   describe "balance handler" do
@@ -61,6 +61,56 @@ defmodule Commanded.Event.HandleEventTest do
         assert AccountBalanceHandler.current_balance() == 1_050
       end)
     end
+
+    test "An event handler can be reset (:origin mode)" do
+      stream_uuid = UUID.uuid4()
+      initial_events = [%BankAccountOpened{account_number: "ACC123", initial_balance: 1_000}]
+
+      :ok = EventStore.append_to_stream(stream_uuid, 0, to_event_data(initial_events))
+
+      {:ok, handler} = BankAccountHandler.start_link()
+
+      Wait.until(fn ->
+        assert BankAccountHandler.current_accounts() == ["ACC123"]
+      end)
+
+      :ok = BankAccountHandler.change_prefix("PREF_")
+
+      send(handler, :reset)
+
+      Wait.until(fn ->
+        assert BankAccountHandler.current_accounts() == ["PREF_ACC123"]
+      end)
+    end
+
+    test "An event handler can be reset (:current mode)" do
+      stream_uuid = UUID.uuid4()
+
+      # Ignored events
+      initial_events = [%BankAccountOpened{account_number: "ACC123", initial_balance: 1_000}]
+      :ok = EventStore.append_to_stream(stream_uuid, 0, to_event_data(initial_events))
+
+      {:ok, handler} = BankAccountHandler.start_link(start_from: :current)
+
+      Wait.until(fn ->
+        assert BankAccountHandler.current_accounts() == []
+      end)
+
+      :ok = BankAccountHandler.change_prefix("PREF_")
+
+      send(handler, :reset)
+
+      new_event = [%BankAccountOpened{account_number: "ACC1234", initial_balance: 1_000}]
+      :ok = EventStore.append_to_stream(stream_uuid, 1, to_event_data(new_event))
+
+      wait_for_event(BankAccountOpened, fn event, recorded_event ->
+        event.account_number == "ACC1234" and recorded_event.event_number == 2
+      end)
+
+      Wait.until(fn ->
+        assert BankAccountHandler.current_accounts() == ["PREF_ACC1234"]
+      end)
+    end
   end
 
   defp to_event_data(events) do
@@ -207,4 +257,30 @@ defmodule Commanded.Event.HandleEventTest do
       end
     """)
   end
+
+  describe "Mix task must be able to reset" do
+    test "Can reset an event handler" do
+      stream_uuid = UUID.uuid4()
+      initial_events = [%BankAccountOpened{account_number: "ACC123", initial_balance: 1_000}]
+
+      :ok = EventStore.append_to_stream(stream_uuid, 0, to_event_data(initial_events))
+
+      {:ok, _handler} = BankAccountHandler.start_link()
+
+      Wait.until(fn ->
+        assert BankAccountHandler.current_accounts() == ["ACC123"]
+      end)
+
+      :ok = BankAccountHandler.change_prefix("PREF_")
+      pid = Commanded.Registration.whereis_name({Commanded.Event.Handler, "Commanded.ExampleDomain.BankAccount.BankAccountHandler"})
+
+      assert :undefined != pid
+
+      Mix.Tasks.Commanded.Reset.run(["Commanded.ExampleDomain.BankAccount.BankAccountHandler"])
+
+      Wait.until(fn ->
+        assert BankAccountHandler.current_accounts() == ["PREF_ACC123"]
+      end)
+    end
+  end
 end
diff --git a/test/example_domain/bank_account/bank_account_handler.ex b/test/example_domain/bank_account/bank_account_handler.ex
new file mode 100644
index 0000000..7225fcc
--- /dev/null
+++ b/test/example_domain/bank_account/bank_account_handler.ex
@@ -0,0 +1,54 @@
+defmodule Commanded.ExampleDomain.BankAccount.BankAccountHandler do
+  @moduledoc false
+
+  use Commanded.Event.Handler, name: __MODULE__, start_from: :origin
+
+  alias Commanded.ExampleDomain.BankAccount.Events.{
+    BankAccountOpened,
+  }
+
+  @agent_name {:global, __MODULE__}
+
+  def init do
+    case Agent.start_link(fn -> %{prefix: "", accounts: []} end, name: @agent_name) do
+      {:ok, _} -> :ok
+      {:error, {:already_started, _}} -> :ok
+      _ -> {:error, :unable_to_start}
+    end
+  end
+
+  def before_reset do
+    Agent.update(@agent_name, fn state -> %{state | accounts: []} end)
+  end
+
+  def handle(%BankAccountOpened{account_number: account_number}, _metadata) do
+    Agent.update(@agent_name, fn %{prefix: prefix, accounts: accounts} = state -> %{state | accounts: accounts ++ [prefix <> account_number]} end)
+  end
+
+  def subscribed? do
+    try do
+      Agent.get(@agent_name, fn _ -> true end)
+    catch
+      :exit, _reason -> false
+    end
+  end
+
+  def change_prefix(prefix) do
+    try do
+      Agent.update(@agent_name, fn s -> %{s | prefix: prefix} end)
+    catch
+      :exit, _reason ->
+        nil
+    end
+  end
+
+  def current_accounts do
+    try do
+      Agent.get(@agent_name, fn %{accounts: accounts} -> accounts end)
+    catch
+      # catch agent not started exits, return `nil` balance
+      :exit, _reason ->
+        nil
+    end
+  end
+end