Repository: slashdotdash/commanded
Pull request #67 (2017-08-03 14:08:36)

Command dispatch optionally returns aggregate version
You can optionally choose to include the aggregate's updated version as part of a command dispatch result by setting `include_aggregate_version` true:

```elixir
{:ok, aggregate_version} = BankRouter.dispatch(command, include_aggregate_version: true)
```

This is useful when you need to wait for an event handler (e.g. a read model projection) to be up-to-date before continuing or querying its data.

The default is to return only `:ok`, which is the same as before this change:

```elixir
:ok = BankRouter.dispatch(command)
```


Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 4b015ab..82f16be 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,5 +1,11 @@
 # Changelog
 
+## v0.13.0
+
+### Enhancements
+
+- Command dispatch optionally returns aggregate version, using `include_aggregate_version: true` during dispatch.
+
 ## v0.12.0
 
 ### Enhancements
diff --git a/README.md b/README.md
index 9c51a8b..108901f 100644
--- a/README.md
+++ b/README.md
@@ -56,7 +56,7 @@ The package can be installed from hex as follows.
 
     ```elixir
     def deps do
-      [{:commanded, "~> 0.12"}]
+      [{:commanded, "~> 0.13"}]
     end
     ```
 
@@ -340,7 +340,7 @@ end
 
 ```elixir
 # dispatch the open account command with a timeout of 2 seconds
-:ok = BankRouter.dispatch(%OpenAccount{account_number: "ACC123", initial_balance: 1_000}, 2_000)
+:ok = BankRouter.dispatch(%OpenAccount{account_number: "ACC123", initial_balance: 1_000}, timeout: 2_000)
 ```
 
 #### Multi-command registration
@@ -355,6 +355,16 @@ defmodule BankRouter do
 end
 ```
 
+#### Dispatch returning aggregate version
+
+You can optionally choose to include the aggregate's version as part of the dispatch result by setting the  `include_aggregate_version` option to true:
+
+```elixir
+{:ok, aggregate_version} = BankRouter.dispatch(command, include_aggregate_version: true)
+```
+
+This is useful when you need to wait for an event handler, such as a read model projection, to be up-to-date before continuing execution or querying its data.
+
 #### Aggregate lifespan
 
 By default an aggregate instance process will run indefinitely once started. You can control this by implementing the `Commanded.Aggregates.AggregateLifespan` behaviour in a module.
diff --git a/lib/commanded/aggregates/aggregate.ex b/lib/commanded/aggregates/aggregate.ex
index 1f66f05..335126a 100644
--- a/lib/commanded/aggregates/aggregate.ex
+++ b/lib/commanded/aggregates/aggregate.ex
@@ -55,7 +55,7 @@ defmodule Commanded.Aggregates.Aggregate do
   - `timeout` is an integer greater than zero which specifies how many milliseconds to wait for a reply, or the atom :infinity to wait indefinitely.
     The default value is 5000.
 
-  Returns `:ok` on success, or `{:error, reason}` on failure
+  Returns `{:ok, aggregate_version}` on success, or `{:error, reason}` on failure
   """
   def execute(aggregate_uuid, command, handler, function \\ :execute, timeout \\ 5_000, lifespan \\ DefaultLifespan) do
     GenServer.call(via_tuple(aggregate_uuid), {:execute_command, handler, function, command, lifespan}, timeout)
@@ -146,8 +146,6 @@ defmodule Commanded.Aggregates.Aggregate do
   defp execute_command(handler, function, command, %Aggregate{aggregate_uuid: aggregate_uuid, aggregate_version: expected_version, aggregate_state: aggregate_state, aggregate_module: aggregate_module} = state) do
     case Kernel.apply(handler, function, [aggregate_state, command]) do
       {:error, _reason} = reply -> {reply, state}
-      nil -> {:ok, state}
-      [] -> {:ok, state}
       events ->
         pending_events = List.wrap(events)
 
@@ -160,7 +158,7 @@ defmodule Commanded.Aggregates.Aggregate do
           aggregate_version: stream_version
         }
 
-        {:ok, state}
+        {{:ok, stream_version}, state}
     end
   end
 
diff --git a/lib/commanded/commands/dispatcher.ex b/lib/commanded/commands/dispatcher.ex
index e6ade54..bea03d3 100644
--- a/lib/commanded/commands/dispatcher.ex
+++ b/lib/commanded/commands/dispatcher.ex
@@ -11,6 +11,7 @@ defmodule Commanded.Commands.Dispatcher do
       handler_module: nil,
       handler_function: nil,
       aggregate_module: nil,
+      include_aggregate_version: nil,
       identity: nil,
       timeout: nil,
       lifespan: nil,
@@ -54,7 +55,7 @@ defmodule Commanded.Commands.Dispatcher do
   end
 
   defp execute(
-    %Payload{handler_module: handler_module, handler_function: handler_function, timeout: timeout, lifespan: lifespan} = payload,
+    %Payload{handler_module: handler_module, handler_function: handler_function, include_aggregate_version: include_aggregate_version, timeout: timeout, lifespan: lifespan} = payload,
     %Pipeline{command: command} = pipeline,
     aggregate_uuid)
   do
@@ -71,9 +72,13 @@ defmodule Commanded.Commands.Dispatcher do
     end
 
     case result do
-      :ok = reply ->
+      {:ok, aggregate_version} ->
         after_dispatch(pipeline, payload)
-        reply
+
+        case include_aggregate_version do
+          true -> {:ok, aggregate_version}
+          false -> :ok
+        end
 
       {:error, error} = reply ->
         after_failure(pipeline, error, payload)
diff --git a/lib/commanded/commands/router.ex b/lib/commanded/commands/router.ex
index fc4af24..93a90a7 100644
--- a/lib/commanded/commands/router.ex
+++ b/lib/commanded/commands/router.ex
@@ -27,6 +27,14 @@ defmodule Commanded.Commands.Router do
 
   The aggregate root must implement an `execute/2` function that receives the aggregate's state and the command to execute.
 
+  ### Aggregate version
+
+  You can optionally choose to include the aggregate's version as part of the dispatch result by setting `include_aggregate_version` true.
+
+      {:ok, aggregate_version} = BankRouter.dispatch(command, include_aggregate_version: true)
+
+  This is useful when you need to wait for an event handler (e.g. a read model projection) to be up-to-date before continuing or querying its data.
+
   """
   defmacro __using__(_) do
     quote do
@@ -37,6 +45,7 @@ defmodule Commanded.Commands.Router do
       @registered_middleware []
       @default_dispatch_timeout 5_000
       @default_lifespan Commanded.Aggregates.Aggregate.DefaultLifespan
+      @include_aggregate_version false
     end
   end
 
@@ -53,25 +62,20 @@ defmodule Commanded.Commands.Router do
   end
 
   @doc """
-  Dispatch the given command to the corresponding handler for a given aggregate root uniquely identified
+  Dispatch the given command, or list of commands, to the corresponding handler for a given aggregate root
   """
-  defmacro dispatch(command_module, opts)
+  defmacro dispatch(command_module_or_modules, opts) do
+    opts = parse_opts(opts, [])
 
-  defmacro dispatch(command_modules, opts) when is_list(command_modules) do
-    Enum.map(command_modules, fn command_module ->
+    command_module_or_modules
+    |> List.wrap()
+    |> Enum.map(fn command_module ->
       quote do
-        dispatch(unquote(command_module), unquote(opts))
+        register(unquote(command_module), unquote(opts))
       end
     end)
   end
 
-  defmacro dispatch(command_module, opts) do
-    opts = parse_opts(opts, [])
-    quote do
-      register(unquote(command_module), unquote(opts))
-    end
-  end
-
   @register_params [:to, :function, :aggregate, :identity, :timeout, :lifespan]
 
   defmacro register(command_module, to: handler, function: function, aggregate: aggregate, identity: identity, timeout: timeout, lifespan: lifespan) do
@@ -94,31 +98,41 @@ defmodule Commanded.Commands.Router do
       """
       @spec dispatch(command :: struct) :: :ok | {:error, reason :: term}
       def dispatch(command)
-      def dispatch(%unquote(command_module){} = command) do
-        do_dispatch(command, unquote(timeout) || @default_dispatch_timeout)
-      end
+      def dispatch(%unquote(command_module){} = command), do: do_dispatch(command, [])
 
       @doc """
       Dispatch the given command to the registered handler providing a timeout.
 
-      - `timeout` is an integer greater than zero which specifies how many milliseconds to allow the command to be handled, or the atom :infinity to wait indefinitely.
-        The default value is 5000.
+      - `timeout_or_opts` is either an integer timeout or a keyword list of options.
+        The timeout must be an integer greater than zero which specifies how many milliseconds to allow the command to be handled, or the atom `:infinity` to wait indefinitely.
+        The default timeout value is 5000.
+        Alternatively, an options keyword list can be provided, it supports:
 
-      Returns `:ok` on success.
+        Options:
+
+          - `:timeout` as described above
+          - `:include_aggregate_version` set to true to include the aggregate stream version in the success response: `{:ok, aggregate_version}`
+            The default is false, to return just `:ok`
+
+      Returns `:ok` on success, unless `:include_aggregate_version` is enabled where it returns `{:ok, aggregate_version}`.
       """
-      @spec dispatch(command :: struct, timeout :: integer | :infinity) :: :ok | {:error, reason :: term}
-      def dispatch(command, timeout)
-      def dispatch(%unquote(command_module){} = command, timeout) do
-        do_dispatch(command, timeout)
-      end
+      @spec dispatch(command :: struct, timeout :: integer | :infinity | keyword()) :: :ok | {:error, reason :: term}
+      def dispatch(command, timeout_or_opts)
+      def dispatch(%unquote(command_module){} = command, :infinity), do: do_dispatch(command, timeout: :infinity)
+      def dispatch(%unquote(command_module){} = command, timeout) when is_integer(timeout), do: do_dispatch(command, timeout: timeout)
+      def dispatch(%unquote(command_module){} = command, opts), do: do_dispatch(command, opts)
+
+      defp do_dispatch(%unquote(command_module){} = command, opts) do
+        timeout = Keyword.get(opts, :timeout, unquote(timeout) || @default_dispatch_timeout)
+        include_aggregate_version = Keyword.get(opts, :include_aggregate_version, @include_aggregate_version)
 
-      defp do_dispatch(%unquote(command_module){} = command, timeout) do
         Commanded.Commands.Dispatcher.dispatch(%Commanded.Commands.Dispatcher.Payload{
           command: command,
           handler_module: unquote(handler),
           handler_function: unquote(function),
           aggregate_module: unquote(aggregate),
           identity: unquote(identity),
+          include_aggregate_version: include_aggregate_version,
           timeout: timeout,
           lifespan: unquote(lifespan) || @default_lifespan,
           middleware: @registered_middleware,
@@ -129,8 +143,13 @@ defmodule Commanded.Commands.Router do
 
   defmacro __before_compile__(_env) do
     quote do
-      # return error if an unregistered command is dispatched
-      def dispatch(command) do
+      @doc """
+      Return an error if an unregistered command is dispatched
+      """
+      def dispatch(command), do: unregistered_command(command)
+      def dispatch(command, opts), do: unregistered_command(command)
+
+      defp unregistered_command(command) do
         Logger.error(fn -> "attempted to dispatch an unregistered command: #{inspect command}" end)
         {:error, :unregistered_command}
       end
diff --git a/mix.exs b/mix.exs
index 473ea4a..7cf26ef 100644
--- a/mix.exs
+++ b/mix.exs
@@ -4,7 +4,7 @@ defmodule Commanded.Mixfile do
   def project do
     [
       app: :commanded,
-      version: "0.12.0",
+      version: "0.13.0",
       elixir: "~> 1.4",
       elixirc_paths: elixirc_paths(Mix.env),
       description: description(),


Test patch:
diff --git a/test/aggregates/event_persistence_test.exs b/test/aggregates/event_persistence_test.exs
index a35b6d8..8f0eba2 100644
--- a/test/aggregates/event_persistence_test.exs
+++ b/test/aggregates/event_persistence_test.exs
@@ -52,7 +52,7 @@ defmodule Commanded.Entities.EventPersistenceTest do
 
     {:ok, ^aggregate_uuid} = Commanded.Aggregates.Supervisor.open_aggregate(ExampleAggregate, aggregate_uuid)
 
-    :ok = Aggregate.execute(aggregate_uuid, %AppendItems{count: 10}, AppendItemsHandler, :handle)
+    {:ok, 10} = Aggregate.execute(aggregate_uuid, %AppendItems{count: 10}, AppendItemsHandler, :handle)
 
     recorded_events = @event_store.stream_forward(aggregate_uuid, 0) |> Enum.to_list()
 
@@ -64,7 +64,7 @@ defmodule Commanded.Entities.EventPersistenceTest do
 
     {:ok, ^aggregate_uuid} = Commanded.Aggregates.Supervisor.open_aggregate(ExampleAggregate, aggregate_uuid)
 
-    :ok = Aggregate.execute(aggregate_uuid, %AppendItems{count: 10}, AppendItemsHandler, :handle)
+    {:ok, 10} = Aggregate.execute(aggregate_uuid, %AppendItems{count: 10}, AppendItemsHandler, :handle)
 
     Commanded.Helpers.Process.shutdown(aggregate_uuid)
 
@@ -82,9 +82,9 @@ defmodule Commanded.Entities.EventPersistenceTest do
 
     {:ok, ^aggregate_uuid} = Commanded.Aggregates.Supervisor.open_aggregate(ExampleAggregate, aggregate_uuid)
 
-    :ok = Aggregate.execute(aggregate_uuid, %AppendItems{count: 100}, AppendItemsHandler, :handle)
-    :ok = Aggregate.execute(aggregate_uuid, %AppendItems{count: 100}, AppendItemsHandler, :handle)
-    :ok = Aggregate.execute(aggregate_uuid, %AppendItems{count: 1}, AppendItemsHandler, :handle)
+    {:ok, 100} = Aggregate.execute(aggregate_uuid, %AppendItems{count: 100}, AppendItemsHandler, :handle)
+    {:ok, 200} = Aggregate.execute(aggregate_uuid, %AppendItems{count: 100}, AppendItemsHandler, :handle)
+    {:ok, 201} = Aggregate.execute(aggregate_uuid, %AppendItems{count: 1}, AppendItemsHandler, :handle)
 
     Commanded.Helpers.Process.shutdown(aggregate_uuid)
 
diff --git a/test/aggregates/execute_command_for_aggregate_test.exs b/test/aggregates/execute_command_for_aggregate_test.exs
index 9df89ff..12745cf 100644
--- a/test/aggregates/execute_command_for_aggregate_test.exs
+++ b/test/aggregates/execute_command_for_aggregate_test.exs
@@ -17,7 +17,7 @@ defmodule Commanded.Entities.ExecuteCommandForAggregateTest do
 
     {:ok, ^account_number} = Commanded.Aggregates.Supervisor.open_aggregate(BankAccount, account_number)
 
-    :ok = Aggregate.execute(account_number, %OpenAccount{account_number: account_number, initial_balance: 1_000}, BankAccount, :open_account)
+    {:ok, 1} = Aggregate.execute(account_number, %OpenAccount{account_number: account_number, initial_balance: 1_000}, BankAccount, :open_account)
 
     Helpers.Process.shutdown(account_number)
 
@@ -33,7 +33,7 @@ defmodule Commanded.Entities.ExecuteCommandForAggregateTest do
 
     {:ok, ^account_number} = Commanded.Aggregates.Supervisor.open_aggregate(BankAccount, account_number)
 
-    :ok = Aggregate.execute(account_number, %OpenAccount{account_number: account_number, initial_balance: 1_000}, OpenAccountHandler, :handle)
+    {:ok, 1} = Aggregate.execute(account_number, %OpenAccount{account_number: account_number, initial_balance: 1_000}, OpenAccountHandler, :handle)
 
     Helpers.Process.shutdown(account_number)
 
@@ -49,7 +49,7 @@ defmodule Commanded.Entities.ExecuteCommandForAggregateTest do
 
     {:ok, ^account_number} = Commanded.Aggregates.Supervisor.open_aggregate(BankAccount, account_number)
 
-    :ok = Aggregate.execute(account_number, %OpenAccount{account_number: account_number, initial_balance: 1_000}, OpenAccountHandler, :handle)
+    {:ok, 1} = Aggregate.execute(account_number, %OpenAccount{account_number: account_number, initial_balance: 1_000}, OpenAccountHandler, :handle)
 
     state_before = Aggregate.aggregate_state(account_number)
 
diff --git a/test/commands/routing_commands_test.exs b/test/commands/routing_commands_test.exs
index e461b99..29e9900 100644
--- a/test/commands/routing_commands_test.exs
+++ b/test/commands/routing_commands_test.exs
@@ -119,8 +119,8 @@ defmodule Commanded.Commands.RoutingCommandsTest do
   end
 
   test "should allow multiple module registrations for multiple commands in a single dispatch" do
-    :ok = MultiCommandRouter.dispatch(%OpenAccount{account_number: "ACC123", initial_balance: 1_000})
-    :ok = MultiCommandRouter.dispatch(%CloseAccount{account_number: "ACC123"})
+    assert :ok == MultiCommandRouter.dispatch(%OpenAccount{account_number: "ACC123", initial_balance: 1_000})
+    assert :ok == MultiCommandRouter.dispatch(%CloseAccount{account_number: "ACC123"})
   end
 
   defmodule MultiCommandHandlerRouter do
@@ -132,7 +132,14 @@ defmodule Commanded.Commands.RoutingCommandsTest do
   end
 
   test "should allow multiple module registrations for different command handlers" do
-    :ok = MultiCommandHandlerRouter.dispatch(%OpenAccount{account_number: "ACC123", initial_balance: 1_000})
-    :ok = MultiCommandHandlerRouter.dispatch(%DepositMoney{account_number: "ACC123", amount: 100})
+    assert :ok == MultiCommandHandlerRouter.dispatch(%OpenAccount{account_number: "ACC123", initial_balance: 1_000})
+    assert :ok == MultiCommandHandlerRouter.dispatch(%DepositMoney{account_number: "ACC123", amount: 100})
+  end
+
+  describe "include aggregate version" do
+    test "should return aggregate's updated stream version" do
+      assert {:ok, 1} == MultiCommandHandlerRouter.dispatch(%OpenAccount{account_number: "ACC123", initial_balance: 1_000}, include_aggregate_version: true)
+      assert {:ok, 2} == MultiCommandHandlerRouter.dispatch(%DepositMoney{account_number: "ACC123", amount: 100}, include_aggregate_version: true)
+    end
   end
 end