Repository: commanded/commanded
Pull request #161 (2018-03-31 18:46:03)

Command dispatch consistency using explicit handler names 
Provide an explicit list of event handler and process manager modules, or their configured names, containing only those handlers you'd like to wait for during command dispatch. No other handlers will be awaited on, regardless of their own configured consistency setting.

```elixir
:ok = BankRouter.dispatch(command, consistency: [ExampleHandler, AnotherHandler])
:ok = BankRouter.dispatch(command, consistency: ["ExampleHandler", "AnotherHandler"])
```

Note you cannot opt-in to strong consistency for a handler that has been configured as eventually consistent.



Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index b63fa6f..189b420 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,5 +1,9 @@
 # Changelog
 
+## Next release
+
+- Command dispatch consistency using explicit handler names ([#161](https://github.com/commanded/commanded/pull/161)).
+
 ## v0.16.0-rc.0
 
 - Support composite command routers ([#111](https://github.com/commanded/commanded/pull/111)).
diff --git a/guides/Commands.md b/guides/Commands.md
index 812efc5..501438b 100644
--- a/guides/Commands.md
+++ b/guides/Commands.md
@@ -154,22 +154,44 @@ end
 
 ### Command dispatch consistency guarantee
 
-You can choose the consistency guarantee provided by a command dispatch using the `consistency` option:
-
-```elixir
-:ok = BankRouter.dispatch(command, consistency: :strong)
-```
-
-The available options are `:eventual` (default) and `:strong`.
+You can choose the consistency guarantee when dispatching a command.
 
 - *Strong consistency* offers up-to-date data but at the cost of high latency.
 - *Eventual consistency* offers low latency but read model queries may reply with stale data since they may not have processed the persisted events.
 
-When dispatching a command using `consistency: :strong` the dispatch will block until all of the strongly consistent event handlers and process managers have handled all events created by the command. This guarantees that when you receive the `:ok` response from dispatch your strongly consistent read models will have been updated and can safely be queried. Strong consistency helps to alleviate problems and workarounds you would otherwise encounter when dealing with eventual consistency in your own application.
+In Commanded, the available options during command dispatch are:
+
+  - `:eventual` (default) - don't block command dispatch and don't wait for any event handlers, regardless of their own consistency configuration.
+
+    ```elixir
+    :ok = BankRouter.dispatch(command)
+    :ok = BankRouter.dispatch(command, consistency: :eventual)
+    ```
+
+  - `:strong` - block command dispatch until all strongly consistent event handlers and process managers have successfully processed all events created by the command.
+
+    ```elixir
+    :ok = BankRouter.dispatch(command, consistency: :strong)
+    ```
+
+    Dispatching a command using `:strong` consistency but without any strongly consistent event handlers configured will have no effect.
+
+  - Provide an explicit list of event handler and process manager modules (or their configured names), containing only those handlers you'd like to wait for. No other handlers will be awaited on, regardless of their own configured consistency setting.
+
+    ```elixir
+    :ok = BankRouter.dispatch(command, consistency: [ExampleHandler, AnotherHandler])
+    :ok = BankRouter.dispatch(command, consistency: ["ExampleHandler", "AnotherHandler"])
+    ```
+
+    Note you cannot opt-in to strong consistency for a handler that has been configured as eventually consistent.
+
+#### Which consistency guarantee should I use?
+
+When dispatching a command using `consistency: :strong` the dispatch will block until all of the strongly consistent event handlers and process managers have handled all events created by the command. This guarantees that when you receive the `:ok` response from dispatch, your strongly consistent read models will have been updated and can safely be queried.
 
-Use `:strong` consistency when you want to query a read model immediately after dispatching a command. You *must* also configure the event handler to use `:strong` consistency. Dispatching a command using `:strong` consistency but without any strongly consistent event handlers configured will have no effect.
+Strong consistency helps to alleviate problems and workarounds you would otherwise encounter when dealing with eventual consistency in your own application. Use `:strong` consistency when you want to query a read model immediately after dispatching a command. You *must* also configure the event handler to use `:strong` consistency.
 
-Using `:eventual` consistency, or omitting the `consistency` option, will cause the command dispatch to immediately return without waiting for any event handlers. The handlers run independently and asynchronously in the background, therefore you will need to deal with potentially stale read model data.
+Using `:eventual` consistency, or omitting the `consistency` option, will cause the command dispatch to immediately return without waiting for any event handlers or process managers. The handlers run independently, and asynchronously, in the background, therefore you will need to deal with potentially stale read model data.
 
 #### Configure default consistency
 
diff --git a/lib/commanded/commands/dispatcher.ex b/lib/commanded/commands/dispatcher.ex
index c91b0e5..cbd1fb7 100644
--- a/lib/commanded/commands/dispatcher.ex
+++ b/lib/commanded/commands/dispatcher.ex
@@ -138,8 +138,11 @@ defmodule Commanded.Commands.Dispatcher do
               metadata: pipeline.metadata,
             }
           }
-        %{include_aggregate_version: true} -> {:ok, pipeline.assigns.aggregate_version}
-        _ -> :ok
+        %{include_aggregate_version: true} ->
+          {:ok, pipeline.assigns.aggregate_version}
+          
+        _ ->
+          :ok
       end
 
     Pipeline.respond(pipeline, response)
diff --git a/lib/commanded/commands/router.ex b/lib/commanded/commands/router.ex
index 91f6908..d6d8485 100644
--- a/lib/commanded/commands/router.ex
+++ b/lib/commanded/commands/router.ex
@@ -68,17 +68,36 @@ defmodule Commanded.Commands.Router do
 
   ## Consistency
 
-  You can choose to dispatch commands using either `:eventual` or `:strong`
-  consistency:
+  You can choose the consistency guarantee when dispatching a command. The
+  available options are:
 
-      :ok = BankRouter.dispatch(command, consistency: :strong)
+    - `:eventual` (default) - don't block command dispatch while waiting for
+      event handlers
 
-  Using `:strong` consistency will block command dispatch until all strongly
-  consistent event handlers and process managers have successfully process all
-  events created by the command.
+        :ok = BankRouter.dispatch(command)
+        :ok = BankRouter.dispatch(command, consistency: :eventual)
 
-  Use this when you have event handlers that update read models you need to
-  query immediately after dispatching the command.
+    - `:strong` - block command dispatch until all strongly
+      consistent event handlers and process managers have successfully processed
+      all events created by the command.
+
+      Use this when you have event handlers that update read models you need to
+      query immediately after dispatching the command.
+
+        :ok = BankRouter.dispatch(command, consistency: :strong)
+
+    - Provide an explicit list of event handler and process manager modules (or
+      their configured names), containing only those handlers you'd like to wait
+      for. No other handlers will be awaited on, regardless of their own
+      configured consistency setting.
+
+      ```elixir
+      :ok = BankRouter.dispatch(command, consistency: [ExampleHandler, AnotherHandler])
+      :ok = BankRouter.dispatch(command, consistency: ["ExampleHandler", "AnotherHandler"])
+      ```
+
+      Note you cannot opt-in to strong consistency for a handler that has been
+      configured as eventually consistent.
 
   ## Aggregate version
 
diff --git a/lib/commanded/event/handler.ex b/lib/commanded/event/handler.ex
index ce8cd5a..53cc3cf 100644
--- a/lib/commanded/event/handler.ex
+++ b/lib/commanded/event/handler.ex
@@ -274,6 +274,9 @@ defmodule Commanded.Event.Handler do
         Supervisor.child_spec(default, [])
       end
 
+      @doc false
+      def __name__, do: @name
+
       @doc false
       def init, do: :ok
 
@@ -404,7 +407,7 @@ defmodule Commanded.Event.Handler do
         {:stop, reason, state}
     end
   end
-  
+
   defp subscribe_to_all_streams(%Handler{} = state) do
     %Handler{
       handler_name: handler_name,
diff --git a/lib/commanded/middleware/consistency_guarantee.ex b/lib/commanded/middleware/consistency_guarantee.ex
index cc9d432..91d685d 100644
--- a/lib/commanded/middleware/consistency_guarantee.ex
+++ b/lib/commanded/middleware/consistency_guarantee.ex
@@ -16,20 +16,34 @@ defmodule Commanded.Middleware.ConsistencyGuarantee do
 
   import Pipeline
 
-  def before_dispatch(%Pipeline{consistency: :eventual} = pipeline), do: pipeline
-  def before_dispatch(%Pipeline{consistency: :strong} = pipeline) do
+  def before_dispatch(%Pipeline{} = pipeline) do
     Pipeline.assign(pipeline, :dispatcher_pid, self())
   end
 
   def after_dispatch(%Pipeline{consistency: :eventual} = pipeline), do: pipeline
-  def after_dispatch(%Pipeline{consistency: :strong, assigns: %{events: []}} = pipeline), do: pipeline
-  def after_dispatch(%Pipeline{consistency: :strong, assigns: %{aggregate_uuid: aggregate_uuid, aggregate_version: aggregate_version, dispatcher_pid: dispatcher_pid}} = pipeline) do
-    case Subscriptions.wait_for(aggregate_uuid, aggregate_version, [dispatcher_pid]) do
+
+  def after_dispatch(%Pipeline{assigns: %{events: []}} = pipeline), do: pipeline
+
+  def after_dispatch(%Pipeline{consistency: consistency, assigns: assigns} = pipeline) do
+    %{
+      aggregate_uuid: aggregate_uuid,
+      aggregate_version: aggregate_version,
+      dispatcher_pid: dispatcher_pid
+    } = assigns
+
+    opts = [consistency: consistency, exclude: dispatcher_pid]
+    
+    case Subscriptions.wait_for(aggregate_uuid, aggregate_version, opts) do
       :ok ->
         pipeline
 
       {:error, :timeout} ->
-        Logger.warn(fn -> "Consistency timeout waiting for aggregate #{inspect aggregate_uuid} at version #{inspect aggregate_version}" end)
+        Logger.warn(fn ->
+          "Consistency timeout waiting for aggregate #{inspect(aggregate_uuid)} at version #{
+            inspect(aggregate_version)
+          }"
+        end)
+
         respond(pipeline, {:error, :consistency_timeout})
     end
   end
diff --git a/lib/commanded/process_managers/process_manager.ex b/lib/commanded/process_managers/process_manager.ex
index cda8d25..ec49e6b 100644
--- a/lib/commanded/process_managers/process_manager.ex
+++ b/lib/commanded/process_managers/process_manager.ex
@@ -246,6 +246,9 @@ defmodule Commanded.ProcessManagers.ProcessManager do
 
         Supervisor.child_spec(default, [])
       end
+
+      @doc false
+      def __name__, do: @name
     end
   end
 
diff --git a/lib/commanded/subscriptions.ex b/lib/commanded/subscriptions.ex
index ae8d39d..c14ee46 100644
--- a/lib/commanded/subscriptions.ex
+++ b/lib/commanded/subscriptions.ex
@@ -29,8 +29,12 @@ defmodule Commanded.Subscriptions do
   handler.
   """
   def ack_event(name, consistency, event)
+
   def ack_event(_name, :eventual, _event), do: :ok
-  def ack_event(name, :strong, %RecordedEvent{stream_id: stream_id, stream_version: stream_version}) do
+
+  def ack_event(name, :strong, %RecordedEvent{} = event) do
+    %RecordedEvent{stream_id: stream_id, stream_version: stream_version} = event
+
     PubSub.broadcast(@ack_topic, {:ack_event, name, stream_id, stream_version})
   end
 
@@ -45,20 +49,25 @@ defmodule Commanded.Subscriptions do
   @doc """
   Have all the registered handlers processed the given event?
   """
-  def handled?(stream_uuid, stream_version, exclude \\ [])
-  def handled?(stream_uuid, stream_version, exclude) do
-    GenServer.call(__MODULE__, {:handled?, stream_uuid, stream_version, exclude})
+  def handled?(stream_uuid, stream_version, opts \\ []) do
+    GenServer.call(__MODULE__, {:handled?, stream_uuid, stream_version, opts})
   end
 
   @doc """
   Wait until all strongly consistent event store subscriptions have received
   and successfully handled the given event, by stream identity and version.
 
+  Options:
+
+    - `:consistency` - override the default consistency (`:strong`), by
+      providing an explicit list of handler modules, or their configured names,
+      to wait for.
+    - `:exclude` - a PID, or list of PIDs, to exclude from waiting.
+
   Returns `:ok` on success, or `{:error, :timeout}` on failure due to timeout.
   """
-  def wait_for(stream_uuid, stream_version, exclude \\ [], timeout \\ default_consistency_timeout())
-  def wait_for(stream_uuid, stream_version, exclude, timeout) do
-    :ok = GenServer.call(__MODULE__, {:subscribe, stream_uuid, stream_version, exclude, self()})
+  def wait_for(stream_uuid, stream_version, opts \\ [], timeout \\ default_consistency_timeout()) do
+    :ok = GenServer.call(__MODULE__, {:subscribe, stream_uuid, stream_version, opts, self()})
 
     receive do
       {:ok, ^stream_uuid, ^stream_version} -> :ok
@@ -83,16 +92,21 @@ defmodule Commanded.Subscriptions do
     {:reply, :ok, initial_state()}
   end
 
-  def handle_call({:handled?, stream_uuid, stream_version, excluding}, _from, %Subscriptions{} = state) do
-    reply = handled_by_all?(stream_uuid, stream_version, MapSet.new(excluding), state)
+  def handle_call({:handled?, stream_uuid, stream_version, opts}, _from, %Subscriptions{} = state) do
+    {consistency, exclude} = parse_opts(opts)
+
+    reply = handled_by_all?(stream_uuid, stream_version, consistency, exclude, state)
 
     {:reply, reply, state}
   end
 
-  def handle_call({:subscribe, stream_uuid, stream_version, exclude, pid}, _from, %Subscriptions{subscribers: subscribers} = state) do
-    exclusions = MapSet.new(exclude)
+  def handle_call({:subscribe, stream_uuid, stream_version, opts, pid}, _from, %Subscriptions{} = state) do
+    {consistency, exclude} = parse_opts(opts)
+
+    %Subscriptions{subscribers: subscribers} = state
+
     state =
-      case handled_by_all?(stream_uuid, stream_version, exclusions, state) do
+      case handled_by_all?(stream_uuid, stream_version, consistency, exclude, state) do
         true ->
           # immediately notify subscriber since all handlers have already
           # processed the requested event
@@ -105,9 +119,9 @@ defmodule Commanded.Subscriptions do
           # processed by all handlers
           Process.monitor(pid)
 
-          %Subscriptions{state |
-            subscribers: [{pid, stream_uuid, stream_version, exclusions} | subscribers]
-          }
+          subscription = {pid, stream_uuid, stream_version, consistency, exclude}
+
+          %Subscriptions{state | subscribers: [subscription | subscribers]}
       end
 
     {:reply, :ok, state}
@@ -166,9 +180,16 @@ defmodule Commanded.Subscriptions do
   defp subscriptions, do: PubSub.list(@subscriptions_topic)
 
   # Have all subscriptions handled the event for the given stream and version
-  defp handled_by_all?(stream_uuid, stream_version, exclude, %Subscriptions{} = state) do
+  defp handled_by_all?(stream_uuid, stream_version, consistency, exclude, %Subscriptions{} = state) do
     subscriptions()
     |> Enum.reject(fn {_name, pid} -> MapSet.member?(exclude, pid) end)
+    |> Enum.filter(fn {name, _pid} ->
+      # Optionally filter subscriptions to those provided by the `consistency` option
+      case consistency do
+        :strong -> true
+        consistency when is_list(consistency) -> Enum.member?(consistency, name)
+      end
+    end)
     |> Enum.all?(fn {name, _pid} -> handled_by?(name, stream_uuid, stream_version, state) end)
   end
 
@@ -182,20 +203,20 @@ defmodule Commanded.Subscriptions do
 
   defp remove_by_pid(subscribers, pid) do
     Enum.reduce(subscribers, subscribers, fn
-     ({^pid, _, _, _} = subscriber, subscribers) -> subscribers -- [subscriber]
+     ({^pid, _, _, _, _} = subscriber, subscribers) -> subscribers -- [subscriber]
      (_subscriber, subscribers) -> subscribers
     end)
   end
 
-  # notify any subscribers waiting on a given stream if it is at the expected version
+  # Notify any subscribers waiting on a given stream if it is at the expected version
   defp notify_subscribers(stream_uuid, %Subscriptions{subscribers: subscribers} = state) do
     Enum.reduce(subscribers, subscribers, fn
-      ({pid, ^stream_uuid, expected_stream_version, exclude} = subscriber, subscribers) ->
-        case handled_by_all?(stream_uuid, expected_stream_version, exclude, state) do
+      ({pid, ^stream_uuid, expected_stream_version, consistency, exclude} = subscriber, subscribers) ->
+        case handled_by_all?(stream_uuid, expected_stream_version, consistency, exclude, state) do
           true ->
             notify_subscriber(pid, stream_uuid, expected_stream_version)
 
-            # remove subscriber
+            # Remove subscriber
             subscribers -- [subscriber]
 
           false ->
@@ -210,12 +231,12 @@ defmodule Commanded.Subscriptions do
     send(pid, {:ok, stream_uuid, stream_version})
   end
 
-  # send an info message to the process to purge expired stream ack's
+  # Send an info message to the process to purge expired stream ack's
   defp schedule_purge_streams(ttl \\ default_ttl()) do
     Process.send_after(self(), {:purge_expired_streams, ttl}, ttl)
   end
 
-  # delete subscription ack's that are older than the configured ttl
+  # Delete subscription ack's that are older than the configured ttl
   defp purge_expired_streams(ttl, %Subscriptions{streams_table: streams_table, started_at: started_at}) do
     stale_epoch = monotonic_time() - started_at - (ttl / 1_000)
 
@@ -224,6 +245,34 @@ defmodule Commanded.Subscriptions do
     |> Enum.each(&:ets.delete(streams_table, &1))
   end
 
+  defp parse_opts(opts) do
+    consistency = opts |> Keyword.get(:consistency, :strong) |> parse_consistency()
+    exclude = opts |> Keyword.get(:exclude, []) |> parse_exclude()
+
+    {consistency, exclude}
+  end
+
+  defp parse_consistency(consistency) when consistency in [:eventual, :strong], do: consistency
+  defp parse_consistency(consistency) when is_list(consistency) do
+    Enum.map(consistency, fn
+      name when is_bitstring(name) ->
+        name
+
+      module when is_atom(module) ->
+        if function_exported?(module, :__name__, 0) do
+          apply(module, :__name__, [])
+        else
+          raise "Expected module #{module} to contain a public `__name__/0` function"
+        end
+
+      invalid ->
+        raise "Invalid consistency: #{inspect invalid}"
+    end)
+  end
+  defp parse_consistency(invalid), do: raise "Invalid consistency: #{inspect invalid}"
+
+  defp parse_exclude(exclude), do: exclude |> List.wrap() |> MapSet.new()
+
   defp monotonic_time, do: System.monotonic_time(:seconds)
 
   @default_ttl :timer.hours(1)


Test patch:
diff --git a/test/commands/dispatch_consistency_test.exs b/test/commands/dispatch_consistency_test.exs
index 15eece4..77d9cef 100644
--- a/test/commands/dispatch_consistency_test.exs
+++ b/test/commands/dispatch_consistency_test.exs
@@ -7,6 +7,7 @@ defmodule Commanded.Commands.DispatchConsistencyTest do
     ConsistencyPrefixRouter,
     EventuallyConsistentEventHandler,
     ExecutionResult,
+    OptionalStronglyConsistentEventHandler,
     StronglyConsistentEventHandler,
   }
   alias Commanded.EventStore
@@ -55,6 +56,33 @@ defmodule Commanded.Commands.DispatchConsistencyTest do
     assert {:error, :consistency_timeout} = ConsistencyRouter.dispatch(command, consistency: :strong)
   end
 
+  describe "optional `:strong` consistency" do
+    setup do
+      {:ok, handler3} = OptionalStronglyConsistentEventHandler.start_link()
+
+      on_exit fn ->
+        ProcessHelper.shutdown(handler3)
+      end
+
+      :ok
+    end
+
+    test "should only wait for opt-in strongly consistent event handler to handle event" do
+      command = %ConsistencyCommand{uuid: UUID.uuid4(), delay: 100}
+      opts = [consistency: [OptionalStronglyConsistentEventHandler]]
+
+      assert :ok = ConsistencyRouter.dispatch(command, opts)
+    end
+
+    # default consistency timeout set to 100ms test config
+    test "should timeout waiting for strongly consistent event handler to handle event" do
+      command = %ConsistencyCommand{uuid: UUID.uuid4(), delay: 5_000}
+      opts = [consistency: ["OptionalStronglyConsistentEventHandler"]]
+
+      assert {:error, :consistency_timeout} = ConsistencyRouter.dispatch(command, opts)
+    end
+  end
+
   describe "aggregate identity prefix" do
     test "should wait for strongly consistent event handler to handle event" do
       uuid = UUID.uuid4()
diff --git a/test/commands/support/consistency/optional_strongly_consistent_handler.ex b/test/commands/support/consistency/optional_strongly_consistent_handler.ex
new file mode 100644
index 0000000..62afed7
--- /dev/null
+++ b/test/commands/support/consistency/optional_strongly_consistent_handler.ex
@@ -0,0 +1,31 @@
+defmodule Commanded.Commands.OptionalStronglyConsistentEventHandler do
+  use Commanded.Event.Handler,
+    name: "OptionalStronglyConsistentEventHandler",
+    consistency: :strong
+
+  alias Commanded.Commands.{
+    ConsistencyAggregateRoot,
+    ConsistencyRouter
+  }
+
+  alias ConsistencyAggregateRoot.{
+    ConsistencyCommand,
+    ConsistencyEvent,
+    DispatchRequestedEvent
+  }
+
+  def handle(%ConsistencyEvent{delay: delay}, _metadata) do
+    :timer.sleep(round(delay / 10))
+    :ok
+  end
+
+  # handle event by dispatching a command
+  def handle(%DispatchRequestedEvent{uuid: uuid, delay: delay}, _metadata) do
+    :timer.sleep(round(delay / 10))
+
+    ConsistencyRouter.dispatch(
+      %ConsistencyCommand{uuid: uuid, delay: delay},
+      consistency: :strong
+    )
+  end
+end
diff --git a/test/event/event_handler_macro_test.exs b/test/event/event_handler_macro_test.exs
index a6da598..c6fd418 100644
--- a/test/event/event_handler_macro_test.exs
+++ b/test/event/event_handler_macro_test.exs
@@ -2,14 +2,14 @@ defmodule Commanded.Event.EventHandlerMacroTest do
   use Commanded.StorageCase
 
   alias Commanded.Event.IgnoredEvent
-  alias Commanded.Helpers.{EventFactory,ProcessHelper,Wait}
-  alias Commanded.ExampleDomain.BankAccount.Events.{BankAccountOpened,MoneyDeposited}
+  alias Commanded.Helpers.{EventFactory, ProcessHelper, Wait}
+  alias Commanded.ExampleDomain.BankAccount.Events.{BankAccountOpened, MoneyDeposited}
   alias Commanded.ExampleDomain.BankAccount.AccountBalanceHandler
 
   setup do
-    on_exit fn ->
+    on_exit(fn ->
       ProcessHelper.shutdown(AccountBalanceHandler)
-    end
+    end)
   end
 
   test "should handle published events" do
@@ -23,14 +23,19 @@ defmodule Commanded.Event.EventHandlerMacroTest do
       [
         %BankAccountOpened{account_number: "ACC123", initial_balance: 1_000},
         %MoneyDeposited{amount: 50, balance: 1_050},
-        %IgnoredEvent{name: "ignored"},
+        %IgnoredEvent{name: "ignored"}
       ]
       |> EventFactory.map_to_recorded_events()
 
     send(handler, {:events, recorded_events})
 
     Wait.until(fn ->
-      assert AccountBalanceHandler.current_balance == 1_050
+      assert AccountBalanceHandler.current_balance() == 1_050
     end)
   end
+
+  test "should provide `__name__/0` function" do
+    assert AccountBalanceHandler.__name__() ==
+             "Commanded.ExampleDomain.BankAccount.AccountBalanceHandler"
+  end
 end
diff --git a/test/process_managers/process_manager_instance_test.exs b/test/process_managers/process_manager_instance_test.exs
index 6b52da9..444416b 100644
--- a/test/process_managers/process_manager_instance_test.exs
+++ b/test/process_managers/process_manager_instance_test.exs
@@ -4,15 +4,21 @@ defmodule Commanded.ProcessManagers.ProcessManagerInstanceTest do
   alias Commanded.ExampleDomain.MoneyTransfer.Events.MoneyTransferRequested
   alias Commanded.ExampleDomain.TransferMoneyProcessManager
   alias Commanded.EventStore.RecordedEvent
-  alias Commanded.ProcessManagers.{ProcessManagerInstance,NullRouter}
+  alias Commanded.ProcessManagers.{ProcessManagerInstance, NullRouter}
   alias Commanded.Helpers.ProcessHelper
 
   test "process manager handles an event" do
-    transfer_uuid = UUID.uuid4
-    account1_uuid = UUID.uuid4
-    account2_uuid = UUID.uuid4
+    transfer_uuid = UUID.uuid4()
+    account1_uuid = UUID.uuid4()
+    account2_uuid = UUID.uuid4()
 
-    {:ok, process_manager} = ProcessManagerInstance.start_link(NullRouter, "TransferMoneyProcessManager", TransferMoneyProcessManager, transfer_uuid)
+    {:ok, process_manager} =
+      ProcessManagerInstance.start_link(
+        NullRouter,
+        "TransferMoneyProcessManager",
+        TransferMoneyProcessManager,
+        transfer_uuid
+      )
 
     event = %RecordedEvent{
       event_number: 1,
@@ -22,8 +28,8 @@ defmodule Commanded.ProcessManagers.ProcessManagerInstanceTest do
         transfer_uuid: transfer_uuid,
         debit_account: account1_uuid,
         credit_account: account2_uuid,
-        amount: 100,
-      },
+        amount: 100
+      }
     }
 
     :ok = ProcessManagerInstance.process_event(process_manager, event, self())
@@ -34,35 +40,39 @@ defmodule Commanded.ProcessManagers.ProcessManagerInstanceTest do
     ProcessHelper.shutdown(process_manager)
   end
 
+  test "should provide `__name__/0` function" do
+    assert TransferMoneyProcessManager.__name__() == "transfer_money_process_manager"
+  end
+
   test "should ensure a process manager name is provided" do
     assert_raise RuntimeError, "UnnamedProcessManager expects `:name` to be given", fn ->
-      Code.eval_string """
+      Code.eval_string("""
         defmodule UnnamedProcessManager do
           use Commanded.ProcessManagers.ProcessManager,
             router: Commanded.ExampleDomain.BankRouter
         end
-      """
+      """)
     end
   end
 
   test "should ensure a process manager router is provided" do
     assert_raise RuntimeError, "NoRouterProcessManager expects `:router` to be given", fn ->
-      Code.eval_string """
+      Code.eval_string("""
         defmodule NoRouterProcessManager do
           use Commanded.ProcessManagers.ProcessManager,
             name: "MyProcessManager"
         end
-      """
+      """)
     end
   end
 
   test "should allow using process manager module as name" do
-    Code.eval_string """
+    Code.eval_string("""
       defmodule MyProcessManager do
         use Commanded.ProcessManagers.ProcessManager,
           name: __MODULE__,
           router: Commanded.ExampleDomain.BankRouter
       end
-    """
+    """)
   end
 end
diff --git a/test/subscriptions/subscriptions_test.exs b/test/subscriptions/subscriptions_test.exs
index 7e2288e..20cd780 100644
--- a/test/subscriptions/subscriptions_test.exs
+++ b/test/subscriptions/subscriptions_test.exs
@@ -45,7 +45,7 @@ defmodule Commanded.SubscriptionsTest do
       :ok = Subscriptions.register("handler1", :strong)
 
       # current process should not block handler
-      assert Subscriptions.handled?("stream1", 1, [self()])
+      assert Subscriptions.handled?("stream1", 1, exclude: [self()])
     end
   end
 
@@ -66,7 +66,7 @@ defmodule Commanded.SubscriptionsTest do
     test "should immediately succeed when excluding handler process" do
       :ok = Subscriptions.register("handler", :strong)
 
-      assert :ok == Subscriptions.wait_for("stream1", 2, [self()])
+      assert :ok == Subscriptions.wait_for("stream1", 2, exclude: [self()])
     end
 
     test "should succeed when waited event is ack'd" do
@@ -117,6 +117,34 @@ defmodule Commanded.SubscriptionsTest do
 
       assert Subscriptions.handled?("stream1", 2)
     end
+
+    test "should allow per-handler consistency" do
+      :ok = Subscriptions.register("handler1", :strong)
+      :ok = Subscriptions.register("handler2", :strong)
+
+      :ok = Subscriptions.ack_event("handler1", :strong, %RecordedEvent{stream_id: "stream1", stream_version: 2})
+
+      refute Subscriptions.handled?("stream1", 2)
+      assert :ok == Subscriptions.wait_for("stream1", 2, consistency: ["handler1"])
+    end
+
+    test "should wait for each configured handler consistency" do
+      :ok = Subscriptions.register("handler1", :strong)
+      :ok = Subscriptions.register("handler2", :strong)
+      :ok = Subscriptions.register("handler3", :strong)
+      :ok = Subscriptions.register("handler3", :eventual)
+
+      :ok = Subscriptions.ack_event("handler1", :strong, %RecordedEvent{stream_id: "stream1", stream_version: 2})
+      :ok = Subscriptions.ack_event("handler2", :strong, %RecordedEvent{stream_id: "stream1", stream_version: 2})
+
+      refute Subscriptions.handled?("stream1", 2)
+      assert Subscriptions.handled?("stream1", 2, consistency: ["handler1", "handler2"])
+      refute Subscriptions.handled?("stream1", 2, consistency: ["handler1", "handler2", "handler3"])
+      assert Subscriptions.handled?("stream1", 2, consistency: ["handler1", "handler2", "handler4"])
+
+      assert :ok == Subscriptions.wait_for("stream1", 2, consistency: ["handler1", "handler2"])
+      assert {:error, :timeout} == Subscriptions.wait_for("stream1", 2, [consistency: ["handler1", "handler2", "handler3"]], 100)
+    end
   end
 
   describe "expire stream acks" do