Repository: commanded/commanded
Pull request #210 (2018-10-11 04:04:54)

Expands AggregateLifespan to include after_command/1 and after_error/1 
Previously, aggregates would not shut down if one or more events were
not returned after command execution. This change defines callbacks to
handle cases where an error or a [] is returned.


Patch:
diff --git a/lib/commanded/aggregates/aggregate.ex b/lib/commanded/aggregates/aggregate.ex
index 8a2ba81..43e74bf 100644
--- a/lib/commanded/aggregates/aggregate.ex
+++ b/lib/commanded/aggregates/aggregate.ex
@@ -179,13 +179,23 @@ defmodule Commanded.Aggregates.Aggregate do
   end
 
   @doc false
-  def handle_call({:execute_command, %ExecutionContext{} = context}, _from, %Aggregate{} = state) do
+  def handle_call(
+        {:execute_command, %ExecutionContext{lifespan: lifespan, command: cmd} = context},
+        _from,
+        %Aggregate{} = state
+      ) do
     {reply, state} = execute_command(context, state)
 
     lifespan_timeout =
       case reply do
         {:ok, _stream_version, events} ->
-          aggregate_lifespan_timeout(context, events)
+          aggregate_lifespan_timeout(context, events, &lifespan.after_event/1)
+
+        {:ok, _, []} ->
+          aggregate_lifespan_timeout(context, cmd, &lifespan.after_command/1)
+
+        {:error, reason} ->
+          aggregate_lifespan_timeout(context, reason, &lifespan.after_error/1)
 
         _reply ->
           :infinity
@@ -357,12 +367,14 @@ defmodule Commanded.Aggregates.Aggregate do
     end
   end
 
-  defp aggregate_lifespan_timeout(_context, []), do: :infinity
+  defp aggregate_lifespan_timeout(%ExecutionContext{} = context, [event], timeout_fun) do
+    aggregate_lifespan_timeout(%ExecutionContext{} = context, event, timeout_fun)
+  end
 
-  defp aggregate_lifespan_timeout(%ExecutionContext{} = context, [event]) do
+  defp aggregate_lifespan_timeout(%ExecutionContext{} = context, event, timeout_fun) do
     %ExecutionContext{lifespan: lifespan} = context
 
-    case lifespan.after_event(event) do
+    case timeout_fun.(event) do
       timeout when timeout in [:infinity, :hibernate, :stop] ->
         timeout
 
@@ -380,8 +392,8 @@ defmodule Commanded.Aggregates.Aggregate do
     end
   end
 
-  defp aggregate_lifespan_timeout(context, [_event | events]),
-    do: aggregate_lifespan_timeout(context, events)
+  defp aggregate_lifespan_timeout(context, [_event | events], timeout_fun),
+    do: aggregate_lifespan_timeout(context, events, timeout_fun)
 
   defp execute_command(%ExecutionContext{retry_attempts: retry_attempts}, %Aggregate{} = state)
        when retry_attempts < 0,
diff --git a/lib/commanded/aggregates/aggregate_lifespan.ex b/lib/commanded/aggregates/aggregate_lifespan.ex
index 4e3453d..314f42b 100644
--- a/lib/commanded/aggregates/aggregate_lifespan.ex
+++ b/lib/commanded/aggregates/aggregate_lifespan.ex
@@ -9,9 +9,15 @@ defmodule Commanded.Aggregates.AggregateLifespan do
   it in your router.
 
   After a command successfully executes, and creates at least one domain event,
-  the `c:after_event/1` function is called passing the last created event. The
-  returned inactivity timeout value is used to shutdown the aggregate process if
-  no other messages are received.
+  the `c:after_event/1` function is called passing the last created event.
+
+  When a command returns `[]`, the command struct is passed to the `c:after_command/1` function.
+
+  Finally, if there is an error executing the command, the error reason is passed to the
+  `c:after_error/1` function.
+
+  For all the above, the returned inactivity timeout value is used to shutdown the aggregate process
+  if no other messages are received.
 
   ## Supported return values
 
@@ -35,6 +41,10 @@ defmodule Commanded.Aggregates.AggregateLifespan do
       defmodule BankAccountLifespan do
         @behaviour Commanded.Aggregates.AggregateLifespan
 
+        def after_command(%CloseAccount{}), do: :stop
+
+        def after_error(:invalid_initial_balance), do: 10_000
+
         def after_event(%BankAccountOpened{}), do: :infinity
         def after_event(%MoneyDeposited{}), do: 60_000
         def after_event(%BankAccountClosed{}), do: :stop
@@ -54,9 +64,23 @@ defmodule Commanded.Aggregates.AggregateLifespan do
 
   """
 
+  @type error :: {:error, any}
+
   @doc """
   Aggregate process will be stopped after specified inactivity timeout unless
   `:infinity`, `:hibernate`, or `:stop` are returned.
   """
   @callback after_event(event :: struct()) :: timeout() | :hibernate | :stop
+
+  @doc """
+  Aggregate process will be stopped after specified inactivity timeout unless
+  `:infinity`, `:hibernate`, or `:stop` are returned.
+  """
+  @callback after_command(cmd :: struct()) :: timeout() | :hibernate | :stop
+
+  @doc """
+  Aggregate process will be stopped after specified inactivity timeout unless
+  `:infinity`, `:hibernate`, or `:stop` are returned.
+  """
+  @callback after_error(atom) :: timeout() | :hibernate | :stop
 end
diff --git a/lib/commanded/aggregates/default_lifespan.ex b/lib/commanded/aggregates/default_lifespan.ex
index 8d0bfde..9b2c0a7 100644
--- a/lib/commanded/aggregates/default_lifespan.ex
+++ b/lib/commanded/aggregates/default_lifespan.ex
@@ -13,4 +13,6 @@ defmodule Commanded.Aggregates.DefaultLifespan do
   Aggregate will run indefinitely once started.
   """
   def after_event(_event), do: :infinity
+  def after_command(_event), do: :infinity
+  def after_error(_event), do: :infinity
 end


Test patch:
diff --git a/test/aggregates/aggregate_lifespan_test.exs b/test/aggregates/aggregate_lifespan_test.exs
index 7d56821..78e05aa 100644
--- a/test/aggregates/aggregate_lifespan_test.exs
+++ b/test/aggregates/aggregate_lifespan_test.exs
@@ -126,6 +126,32 @@ defmodule Commanded.Aggregates.AggregateLifespanTest do
 
       refute_receive {:DOWN, ^ref, :process, ^pid, _}
     end
+
+    test "should adhere to aggregate lifespan after error", %{
+      aggregate_uuid: aggregate_uuid,
+      ref: ref
+    } do
+      ib = "clearly invalid"
+
+      {:error, :invalid_initial_balance} =
+        BankRouter.dispatch(%OpenAccount{account_number: aggregate_uuid, initial_balance: ib})
+
+      refute_receive {:DOWN, ^ref, :process, _, :normal}, 29
+      assert_receive {:DOWN, ^ref, :process, _, :normal}, 35
+    end
+
+    test "should adhere to timeout when no event returned", %{
+      aggregate_uuid: aggregate_uuid,
+      ref: ref
+    } do
+      :ok =
+        BankRouter.dispatch(%OpenAccount{account_number: aggregate_uuid, initial_balance: 100})
+
+      :ok = BankRouter.dispatch(%CloseAccount{account_number: aggregate_uuid})
+      :ok = BankRouter.dispatch(%CloseAccount{account_number: aggregate_uuid})
+
+      assert_receive {:DOWN, ^ref, :process, _, :normal}
+    end
   end
 
   describe "deprecated `after_command/1` callback" do
diff --git a/test/aggregates/support/bank_account_lifespan.ex b/test/aggregates/support/bank_account_lifespan.ex
index 527ef0c..75c3b02 100644
--- a/test/aggregates/support/bank_account_lifespan.ex
+++ b/test/aggregates/support/bank_account_lifespan.ex
@@ -9,6 +9,13 @@ defmodule Commanded.Aggregates.BankAccountLifespan do
     MoneyDeposited
   }
 
+  alias Commanded.ExampleDomain.BankAccount.Commands.CloseAccount
+
+  def after_command(%CloseAccount{}), do: :stop
+  def after_command(_), do: 10
+
+  def after_error(:invalid_initial_balance), do: 30
+
   def after_event(%BankAccountOpened{}), do: 25
   def after_event(%MoneyDeposited{}), do: 50
   def after_event(%BankAccountClosed{}), do: :stop
diff --git a/test/example_domain/bank_account/bank_account.ex b/test/example_domain/bank_account/bank_account.ex
index 739766c..7bc8fd9 100644
--- a/test/example_domain/bank_account/bank_account.ex
+++ b/test/example_domain/bank_account/bank_account.ex
@@ -31,6 +31,10 @@ defmodule Commanded.ExampleDomain.BankAccount do
     BankAccountClosed
   }
 
+  def open_account(%BankAccount{state: nil}, %OpenAccount{initial_balance: "clearly invalid"}) do
+    {:error, :invalid_initial_balance}
+  end
+
   def open_account(%BankAccount{state: nil}, %OpenAccount{
         account_number: account_number,
         initial_balance: initial_balance
@@ -83,6 +87,10 @@ defmodule Commanded.ExampleDomain.BankAccount do
     end
   end
 
+  def close_account(%BankAccount{state: :closed}, %CloseAccount{}) do
+    []
+  end
+
   def close_account(%BankAccount{state: :active}, %CloseAccount{account_number: account_number}) do
     %BankAccountClosed{account_number: account_number}
   end