Repository: commanded/commanded
Pull request #109 (2017-11-08 13:35:31)

Support distributed dispatch consistency 
Allow command dispatch with `consistency: :strong` to be used when Commanded is distributed amongst a cluster of nodes


Patch:
diff --git a/lib/commanded/process_managers/supervisor.ex b/lib/commanded/process_managers/supervisor.ex
index fba7ee6..02c78bb 100644
--- a/lib/commanded/process_managers/supervisor.ex
+++ b/lib/commanded/process_managers/supervisor.ex
@@ -8,7 +8,7 @@ defmodule Commanded.ProcessManagers.Supervisor do
   end
 
   def start_process_manager(supervisor, process_manager_name, process_manager_module, process_uuid) do
-    Logger.debug(fn -> "starting process manager process for `#{process_manager_module}` with uuid #{process_uuid}" end)
+    Logger.debug(fn -> "Starting process manager process for `#{process_manager_module}` with uuid #{process_uuid}" end)
 
     Supervisor.start_child(supervisor, [process_manager_name, process_manager_module, process_uuid])
   end
diff --git a/lib/commanded/registration/local_registry.ex b/lib/commanded/registration/local_registry.ex
index 3d15121..f14a2c1 100644
--- a/lib/commanded/registration/local_registry.ex
+++ b/lib/commanded/registration/local_registry.ex
@@ -13,10 +13,7 @@ defmodule Commanded.Registration.LocalRegistry do
   @impl Commanded.Registration
   def child_spec do
     [
-      Supervisor.child_spec({Registry, [
-        keys: :unique,
-        name: __MODULE__
-      ]}, id: :commanded_local_registry),
+      {Registry, keys: :unique, name: __MODULE__},
     ]
   end
 
@@ -59,6 +56,16 @@ defmodule Commanded.Registration.LocalRegistry do
     end
   end
 
+  @doc """
+  Sends a message to the given dest and returns `:ok`.
+  """
+  @callback multi_send(dest :: atom(), message :: any()) :: :ok
+  @impl Commanded.Registration
+  def multi_send(server, message) do
+    send(server, message)
+    :ok
+  end
+
   @doc """
   Get the pid of a registered name.
 
diff --git a/lib/commanded/registration/registration.ex b/lib/commanded/registration/registration.ex
index 9550ed5..f9c361e 100644
--- a/lib/commanded/registration/registration.ex
+++ b/lib/commanded/registration/registration.ex
@@ -30,6 +30,11 @@ defmodule Commanded.Registration do
   """
   @callback start_link(name :: term(), module :: module(), args :: any()) :: {:ok, pid()} | {:error, reason :: term()}
 
+  @doc """
+  Sends a message to the given dest and returns `:ok`.
+  """
+  @callback multi_send(dest :: atom(), message :: any()) :: :ok
+
   @doc """
   Get the pid of a registered name.
 
@@ -54,6 +59,10 @@ defmodule Commanded.Registration do
   @spec start_link(name :: term(), module :: module(), args :: any()) :: {:ok, pid()} | {:error, reason :: term()}
   def start_link(name, module, args), do: registry_provider().start_link(name, module, args)
 
+  @doc false
+  @spec multi_send(server :: atom(), message :: any()) :: :ok
+  def multi_send(server, message), do: registry_provider().multi_send(server, message)
+
   @doc false
   @spec whereis_name(term()) :: pid() | :undefined
   def whereis_name(name), do: registry_provider().whereis_name(name)
diff --git a/lib/commanded/subscriptions/subscriptions.ex b/lib/commanded/subscriptions/subscriptions.ex
index 6f9c83f..e6147e4 100644
--- a/lib/commanded/subscriptions/subscriptions.ex
+++ b/lib/commanded/subscriptions/subscriptions.ex
@@ -5,6 +5,7 @@ defmodule Commanded.Subscriptions do
 
   alias Commanded.EventStore.RecordedEvent
   alias Commanded.Subscriptions
+  alias Commanded.Registration
 
   defstruct [
     subscriptions_table: nil,
@@ -13,8 +14,8 @@ defmodule Commanded.Subscriptions do
     subscribers: [],
   ]
 
-  def start_link(_arg) do
-    GenServer.start_link(__MODULE__, %Subscriptions{}, name: __MODULE__)
+  def start_link(arg) do
+    GenServer.start_link(__MODULE__, arg, name: __MODULE__)
   end
 
   @doc """
@@ -23,16 +24,17 @@ defmodule Commanded.Subscriptions do
   def register(name, consistency)
   def register(_name, :eventual), do: :ok
   def register(name, :strong) do
-    GenServer.call(__MODULE__, {:register_subscription, name, self()})
+    Registration.multi_send(__MODULE__, {:register_subscription, name, self()})
   end
 
   @doc """
-  Acknowledge receipt and sucessful processing of the given event by the named handler
+  Acknowledge receipt and sucessful processing of the given event by the named
+  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
-    GenServer.cast(__MODULE__, {:ack_event, name, stream_id, stream_version, self()})
+    Registration.multi_send(__MODULE__, {:ack_event, name, stream_id, stream_version, self()})
   end
 
   @doc false
@@ -40,6 +42,11 @@ defmodule Commanded.Subscriptions do
     GenServer.call(__MODULE__, :all_subscriptions)
   end
 
+  @doc false
+  def reset do
+    GenServer.call(__MODULE__, :reset)
+  end
+
   @doc """
   Have all the registered handlers processed the given event?
   """
@@ -67,16 +74,10 @@ defmodule Commanded.Subscriptions do
     end
   end
 
-  def init(%Subscriptions{} = state) do
+  def init(_arg) do
     schedule_purge_streams()
 
-    state = %Subscriptions{state |
-      subscriptions_table: :ets.new(:subscriptions, [:set, :private]),
-      streams_table: :ets.new(:streams, [:set, :private]),
-      started_at: now(),
-    }
-
-    {:ok, state}
+    {:ok, initial_state()}
   end
 
   def handle_call(:all_subscriptions, _from, %Subscriptions{} = state) do
@@ -84,27 +85,37 @@ defmodule Commanded.Subscriptions do
     {:reply, reply, state}
   end
 
+  def handle_call(:reset, _from, %Subscriptions{streams_table: streams_table, subscriptions_table: subscriptions_table}) do
+    :ets.delete(streams_table)
+    :ets.delete(subscriptions_table)
+
+    {: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, excluding, state)
+    reply = handled_by_all?(stream_uuid, stream_version, MapSet.new(excluding), 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)
     state =
-      case handled_by_all?(stream_uuid, stream_version, exclude, state) do
+      case handled_by_all?(stream_uuid, stream_version, exclusions, state) do
         true ->
-          # immediately notify subscriber since all handlers have already processed the requested event
+          # immediately notify subscriber since all handlers have already
+          # processed the requested event
           notify_subscriber(pid, stream_uuid, stream_version)
 
           state
 
         false ->
-          # subscribe process to be notified when the requested event has been processed by all handlers
+          # subscribe process to be notified when the requested event has been
+          # processed by all handlers
           Process.monitor(pid)
 
           %Subscriptions{state |
-            subscribers: [{pid, stream_uuid, stream_version} | subscribers]
+            subscribers: [{pid, stream_uuid, stream_version, exclusions} | subscribers]
           }
       end
 
@@ -119,30 +130,30 @@ defmodule Commanded.Subscriptions do
     {:reply, :ok, state}
   end
 
-  def handle_call({:register_subscription, name, pid}, _from, %Subscriptions{subscriptions_table: subscriptions_table} = state) do
+  def handle_info({:register_subscription, name, pid}, %Subscriptions{subscriptions_table: subscriptions_table} = state) do
     :ets.insert(subscriptions_table, {name, pid})
 
-    {:reply, :ok, state}
+    {:noreply, state}
   end
 
-  def handle_cast({:ack_event, name, stream_uuid, stream_version, pid}, %Subscriptions{streams_table: streams_table, subscriptions_table: subscriptions_table, started_at: started_at} = state) do
-    # track insert date as ms since the subscripions process was started to support expiry of stale acks
+  def handle_info({:ack_event, name, stream_uuid, stream_version, pid}, %Subscriptions{streams_table: streams_table, subscriptions_table: subscriptions_table, started_at: started_at} = state) do
+    # track insert date as milliseconds since the subscriptions process was
+    # started to support expiry of stale acks
     inserted_at_epoch = NaiveDateTime.diff(now(), started_at, :second)
 
     :ets.insert(subscriptions_table, {name, pid})
     :ets.insert(streams_table, {{name, stream_uuid}, stream_version, inserted_at_epoch})
 
-    state =
-      case handled_by_all?(stream_uuid, stream_version, [], state) do
-        true -> notify_subscribers(stream_uuid, stream_version, state)
-        false -> state
-      end
+    state = %Subscriptions{state |
+      subscribers: notify_subscribers(stream_uuid, state),
+    }
 
     {:noreply, state}
   end
 
   @doc """
-  Purge stream acks that are older than the configured ttl (default is one hour).
+  Purge stream acks that are older than the configured time-to-live (default is
+  one hour).
   """
   def handle_info({:purge_expired_streams, ttl}, %Subscriptions{} = state) do
     purge_expired_streams(ttl, state)
@@ -159,12 +170,19 @@ defmodule Commanded.Subscriptions do
     {:noreply, state}
   end
 
+  defp initial_state do
+    %Subscriptions{
+      subscriptions_table: :ets.new(:subscriptions, [:set, :private]),
+      streams_table: :ets.new(:streams, [:set, :private]),
+      started_at: now(),
+    }
+  end
+
   # Have all subscriptions handled the event for the given stream and version
   defp handled_by_all?(stream_uuid, stream_version, exclude, %Subscriptions{} = state) do
-    exclusions = MapSet.new(exclude)
     state
     |> subscriptions()
-    |> Enum.reject(fn {_name, pid} -> MapSet.member?(exclusions, pid) end)
+    |> Enum.reject(fn {_name, pid} -> MapSet.member?(exclude, pid) end)
     |> Enum.all?(fn {name, _pid} -> handled_by?(name, stream_uuid, stream_version, state) end)
   end
 
@@ -182,28 +200,33 @@ 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 who are waiting for handlers to have processed a given event
-  defp notify_subscribers(_stream_uuid, _stream_version, %Subscriptions{subscribers: []} = state), do: state
-  defp notify_subscribers(stream_uuid, stream_version, %Subscriptions{subscribers: subscribers} = state) do
-    subscribers =
-      Enum.reduce(subscribers, subscribers, fn
-        ({pid, ^stream_uuid, expected_stream_uuid} = subscriber, subscribers) when expected_stream_uuid <= stream_version ->
-          # notify subscriber and remove from subscribers list
-          notify_subscriber(pid, stream_uuid, stream_version)
-          subscribers -- [subscriber]
+  # 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
+          true ->
+            notify_subscriber(pid, stream_uuid, expected_stream_version)
 
-        (_subscriber, subscribers) -> subscribers
-      end)
+            # remove subscriber
+            subscribers -- [subscriber]
 
-    %Subscriptions{state | subscribers: subscribers}
+          false ->
+            subscribers
+        end
+
+      (_subscriber, subscribers) -> subscribers
+    end)
   end
 
-  defp notify_subscriber(pid, stream_uuid, stream_version), do: send(pid, {:ok, stream_uuid, stream_version})
+  defp notify_subscriber(pid, stream_uuid, stream_version) do
+    send(pid, {:ok, stream_uuid, stream_version})
+  end
 
   # send an info message to the process to purge expired stream ack's
   defp schedule_purge_streams(ttl \\ default_ttl()) do
diff --git a/lib/commanded/supervisor.ex b/lib/commanded/supervisor.ex
index 0a9a858..d64956c 100644
--- a/lib/commanded/supervisor.ex
+++ b/lib/commanded/supervisor.ex
@@ -10,7 +10,7 @@ defmodule Commanded.Supervisor do
 
   def init(_) do
     children = [
-      Supervisor.child_spec({Task.Supervisor, [name: Commanded.Commands.TaskDispatcher]}, []),
+      {Task.Supervisor, name: Commanded.Commands.TaskDispatcher},
       {Commanded.Aggregates.Supervisor, []},
       {Commanded.Subscriptions, []},
     ] ++ Registration.child_spec()
diff --git a/mix.exs b/mix.exs
index a6d7c7c..5c6ceee 100644
--- a/mix.exs
+++ b/mix.exs
@@ -30,7 +30,16 @@ defmodule Commanded.Mixfile do
     ]
   end
 
-  defp elixirc_paths(:test), do: ["lib", "test/aggregates", "test/commands", "test/event", "test/example_domain", "test/helpers", "test/process_managers"]
+  defp elixirc_paths(:test), do: [
+    "lib",
+    "test/aggregates",
+    "test/commands",
+    "test/event",
+    "test/example_domain",
+    "test/helpers",
+    "test/process_managers",
+    "test/subscriptions",
+  ]
   defp elixirc_paths(_), do: ["lib", "test/helpers"]
 
   defp deps do
@@ -80,6 +89,7 @@ Use Commanded to build your own Elixir applications following the CQRS/ES patter
         "test/example_domain",
         "test/helpers",
         "test/process_managers",
+        "test/subscriptions",
       ],
       maintainers: ["Ben Smith"],
       licenses: ["MIT"],


Test patch:
diff --git a/test/commands/correlation_causation_test.exs b/test/commands/correlation_causation_test.exs
index 9cae5b4..043993a 100644
--- a/test/commands/correlation_causation_test.exs
+++ b/test/commands/correlation_causation_test.exs
@@ -3,25 +3,26 @@ defmodule Commanded.Commands.CorrelationCasuationTest do
 
   import Commanded.Assertions.EventAssertions
 
+  alias Commanded.Commands.OpenAccountBonusHandler
   alias Commanded.EventStore
   alias Commanded.ExampleDomain.{
     BankRouter,
     TransferMoneyProcessManager,
   }
   alias Commanded.ExampleDomain.BankAccount.Commands.{
-    DepositMoney,
     OpenAccount,
     WithdrawMoney,
   }
   alias Commanded.ExampleDomain.BankAccount.Events.{
-    BankAccountOpened,
     MoneyDeposited,
   }
   alias Commanded.ExampleDomain.MoneyTransfer.Commands.TransferMoney
   alias Commanded.Helpers.{CommandAuditMiddleware,ProcessHelper}
 
   setup do
+    CommandAuditMiddleware.start_link()
     CommandAuditMiddleware.reset()
+
     {:ok, process_manager} = TransferMoneyProcessManager.start_link()
 
     on_exit fn ->
@@ -135,26 +136,6 @@ defmodule Commanded.Commands.CorrelationCasuationTest do
   describe "event handler dispatch command" do
     setup [:start_account_bonus_handler]
 
-    defmodule OpenAccountBonusHandler do
-      use Commanded.Event.Handler, name: "OpenAccountBonus"
-
-      def handle(
-        %BankAccountOpened{account_number: account_number},
-        %{event_id: causation_id, correlation_id: correlation_id})
-      do
-        deposit_welcome_bonus = %DepositMoney{
-          account_number: account_number,
-          transfer_uuid: UUID.uuid4(),
-          amount: 100,
-        }
-
-        BankRouter.dispatch(deposit_welcome_bonus,
-          causation_id: causation_id,
-          correlation_id: correlation_id,
-        )
-      end
-    end
-
     test "should copy `correlation_id` from handled event" do
       correlation_id = UUID.uuid4()
 
diff --git a/test/commands/routing_commands_test.exs b/test/commands/routing_commands_test.exs
index 736d71d..ab001e0 100644
--- a/test/commands/routing_commands_test.exs
+++ b/test/commands/routing_commands_test.exs
@@ -43,25 +43,8 @@ defmodule Commanded.Commands.RoutingCommandsTest do
   end
 
   describe "identify aggregate" do
-    defmodule IdentityCommand, do: defstruct [:uuid]
-    defmodule IdentityEvent, do: defstruct [:uuid]
-
-    defmodule IdentityAggregateRoot do
-      defstruct [uuid: nil]
-
-      def execute(%__MODULE__{}, %IdentityCommand{uuid: uuid}), do: %IdentityEvent{uuid: uuid}
-      def apply(aggregate, _event), do: aggregate
-    end
-
-    defmodule IdentityAggregateRouter do
-      use Commanded.Commands.Router
-
-      identify IdentityAggregateRoot,
-        by: :uuid,
-        prefix: "prefix-"
-
-      dispatch IdentityCommand, to: IdentityAggregateRoot
-    end
+    alias Commanded.Commands.IdentityAggregateRouter
+    alias Commanded.Commands.IdentityAggregate.IdentityCommand
 
     test "should dispatch command to registered handler" do
       assert :ok = IdentityAggregateRouter.dispatch(%IdentityCommand{uuid: UUID.uuid4})
@@ -77,34 +60,16 @@ defmodule Commanded.Commands.RoutingCommandsTest do
   end
 
   describe "identify aggregate using function" do
-    defmodule IdentityFunctionCommand, do: defstruct [:uuid]
-    defmodule IdentityFunctionEvent, do: defstruct [:uuid]
-
-    defmodule IdentityFunctionAggregateRoot do
-      defstruct [uuid: nil]
-
-      def execute(%__MODULE__{}, %IdentityFunctionCommand{uuid: uuid}), do: %IdentityFunctionEvent{uuid: uuid}
-      def apply(aggregate, _event), do: aggregate
-    end
-
-    defmodule IdentityFunctionAggregateRouter do
-      use Commanded.Commands.Router
-
-      identify IdentityFunctionAggregateRoot,
-        by: &IdentityFunctionAggregateRouter.aggregate_identity/1
-
-      dispatch IdentityFunctionCommand, to: IdentityFunctionAggregateRoot
-
-      def aggregate_identity(%{uuid: uuid}), do: "fun-prefix-" <> uuid
-    end
+    alias Commanded.Commands.IdentityFunctionRouter
+    alias Commanded.Commands.IdentityFunctionAggregate.IdentityFunctionCommand
 
     test "should dispatch command to registered handler" do
-      assert :ok = IdentityFunctionAggregateRouter.dispatch(%IdentityFunctionCommand{uuid: UUID.uuid4})
+      assert :ok = IdentityFunctionRouter.dispatch(%IdentityFunctionCommand{uuid: UUID.uuid4})
     end
 
     test "should append events to stream" do
       uuid = UUID.uuid4()
-      assert :ok = IdentityFunctionAggregateRouter.dispatch(%IdentityFunctionCommand{uuid: uuid})
+      assert :ok = IdentityFunctionRouter.dispatch(%IdentityFunctionCommand{uuid: uuid})
 
       recorded_events = EventStore.stream_forward("fun-prefix-" <> uuid, 0) |> Enum.to_list()
       assert length(recorded_events) == 1
diff --git a/test/commands/support/identity/identity_aggregate.ex b/test/commands/support/identity/identity_aggregate.ex
new file mode 100644
index 0000000..1af47d5
--- /dev/null
+++ b/test/commands/support/identity/identity_aggregate.ex
@@ -0,0 +1,10 @@
+defmodule Commanded.Commands.IdentityAggregate do
+  @moduledoc false
+  defstruct [uuid: nil]
+
+  defmodule IdentityCommand, do: defstruct [:uuid]
+  defmodule IdentityEvent, do: defstruct [:uuid]
+
+  def execute(%__MODULE__{}, %IdentityCommand{uuid: uuid}), do: %IdentityEvent{uuid: uuid}
+  def apply(aggregate, _event), do: aggregate
+end
diff --git a/test/commands/support/identity/identity_aggregate_router.ex b/test/commands/support/identity/identity_aggregate_router.ex
new file mode 100644
index 0000000..9d8ff33
--- /dev/null
+++ b/test/commands/support/identity/identity_aggregate_router.ex
@@ -0,0 +1,13 @@
+defmodule Commanded.Commands.IdentityAggregateRouter do
+  @moduledoc false
+  use Commanded.Commands.Router
+
+  alias Commanded.Commands.IdentityAggregate
+  alias Commanded.Commands.IdentityAggregate.IdentityCommand
+
+  identify IdentityAggregate,
+    by: :uuid,
+    prefix: "prefix-"
+
+  dispatch IdentityCommand, to: IdentityAggregate
+end
diff --git a/test/commands/support/identity/identity_function_aggregate.ex b/test/commands/support/identity/identity_function_aggregate.ex
new file mode 100644
index 0000000..20fa8d2
--- /dev/null
+++ b/test/commands/support/identity/identity_function_aggregate.ex
@@ -0,0 +1,10 @@
+defmodule Commanded.Commands.IdentityFunctionAggregate do
+  @moduledoc false
+  defstruct [uuid: nil]
+
+  defmodule IdentityFunctionCommand, do: defstruct [:uuid]
+  defmodule IdentityFunctionEvent, do: defstruct [:uuid]
+
+  def execute(%__MODULE__{}, %IdentityFunctionCommand{uuid: uuid}), do: %IdentityFunctionEvent{uuid: uuid}
+  def apply(aggregate, _event), do: aggregate
+end
diff --git a/test/commands/support/identity/identity_function_router.ex b/test/commands/support/identity/identity_function_router.ex
new file mode 100644
index 0000000..8eb16de
--- /dev/null
+++ b/test/commands/support/identity/identity_function_router.ex
@@ -0,0 +1,14 @@
+defmodule Commanded.Commands.IdentityFunctionRouter do
+  @moduledoc false
+  use Commanded.Commands.Router
+
+  alias Commanded.Commands.{IdentityFunctionAggregate,IdentityFunctionRouter}
+  alias Commanded.Commands.IdentityFunctionAggregate.IdentityFunctionCommand
+
+  identify IdentityFunctionAggregate,
+    by: &IdentityFunctionRouter.aggregate_identity/1
+
+  dispatch IdentityFunctionCommand, to: IdentityFunctionAggregate
+
+  def aggregate_identity(%{uuid: uuid}), do: "fun-prefix-" <> uuid
+end
diff --git a/test/commands/support/open_account_bonus_handler.ex b/test/commands/support/open_account_bonus_handler.ex
new file mode 100644
index 0000000..fc1b8c7
--- /dev/null
+++ b/test/commands/support/open_account_bonus_handler.ex
@@ -0,0 +1,23 @@
+defmodule Commanded.Commands.OpenAccountBonusHandler do
+  use Commanded.Event.Handler, name: "OpenAccountBonus"
+
+  alias Commanded.ExampleDomain.BankRouter
+  alias Commanded.ExampleDomain.BankAccount.Commands.DepositMoney
+  alias Commanded.ExampleDomain.BankAccount.Events.BankAccountOpened
+
+  def handle(
+    %BankAccountOpened{account_number: account_number},
+    %{event_id: causation_id, correlation_id: correlation_id})
+  do
+    deposit_welcome_bonus = %DepositMoney{
+      account_number: account_number,
+      transfer_uuid: UUID.uuid4(),
+      amount: 100,
+    }
+
+    BankRouter.dispatch(deposit_welcome_bonus,
+      causation_id: causation_id,
+      correlation_id: correlation_id,
+    )
+  end
+end
diff --git a/test/event/event_handler_error_handling_test.exs b/test/event/event_handler_error_handling_test.exs
index 438dca3..e70716e 100644
--- a/test/event/event_handler_error_handling_test.exs
+++ b/test/event/event_handler_error_handling_test.exs
@@ -12,6 +12,12 @@ defmodule Commanded.Event.EventHandlerErrorHandlingTest do
   alias Commanded.ExampleDomain.BankAccount.Commands.OpenAccount
   alias Commanded.ExampleDomain.BankAccount.Events.BankAccountOpened
   alias Commanded.ExampleDomain.BankRouter
+  alias Commanded.Helpers.CommandAuditMiddleware
+
+  setup do
+    CommandAuditMiddleware.start_link()
+    :ok
+  end
 
   test "should stop event handler on error" do
     account_number = UUID.uuid4()
diff --git a/test/helpers/command_audit_middleware.ex b/test/helpers/command_audit_middleware.ex
index f890c47..112da73 100644
--- a/test/helpers/command_audit_middleware.ex
+++ b/test/helpers/command_audit_middleware.ex
@@ -2,6 +2,8 @@ defmodule Commanded.Helpers.CommandAuditMiddleware do
   @moduledoc false
   @behaviour Commanded.Middleware
 
+  @agent_name {:global, __MODULE__}
+
   defmodule AuditLog do
     @moduledoc false
     defstruct [
@@ -12,11 +14,11 @@ defmodule Commanded.Helpers.CommandAuditMiddleware do
   end
 
   def start_link do
-    Agent.start_link(fn -> %AuditLog{} end, name: __MODULE__)
+    Agent.start_link(fn -> %AuditLog{} end, name: @agent_name)
   end
 
   def before_dispatch(pipeline) do
-    Agent.update(__MODULE__, fn %AuditLog{dispatched: dispatched} = audit ->
+    Agent.update(@agent_name, fn %AuditLog{dispatched: dispatched} = audit ->
       %AuditLog{audit | dispatched: dispatched ++ [pipeline]}
     end)
 
@@ -24,7 +26,7 @@ defmodule Commanded.Helpers.CommandAuditMiddleware do
   end
 
   def after_dispatch(pipeline) do
-    Agent.update(__MODULE__, fn %AuditLog{succeeded: succeeded} = audit ->
+    Agent.update(@agent_name, fn %AuditLog{succeeded: succeeded} = audit ->
       %AuditLog{audit | succeeded: succeeded ++ [pipeline]}
     end)
 
@@ -32,7 +34,7 @@ defmodule Commanded.Helpers.CommandAuditMiddleware do
   end
 
   def after_failure(pipeline) do
-    Agent.update(__MODULE__, fn %AuditLog{failed: failed} = audit ->
+    Agent.update(@agent_name, fn %AuditLog{failed: failed} = audit ->
       %AuditLog{audit | failed: failed ++ [pipeline]}
     end)
 
@@ -40,14 +42,14 @@ defmodule Commanded.Helpers.CommandAuditMiddleware do
   end
 
   def reset do
-    Agent.update(__MODULE__, fn _ -> %AuditLog{} end)
+    Agent.update(@agent_name, fn _ -> %AuditLog{} end)
   end
 
   @doc """
   Get the counts of the dispatched, succeeded, and failed commands
   """
   def count_commands do
-    Agent.get(__MODULE__, fn %AuditLog{dispatched: dispatched, succeeded: succeeded, failed: failed} ->
+    Agent.get(@agent_name, fn %AuditLog{dispatched: dispatched, succeeded: succeeded, failed: failed} ->
       {length(dispatched), length(succeeded), length(failed)}
     end)
   end
@@ -56,7 +58,7 @@ defmodule Commanded.Helpers.CommandAuditMiddleware do
   Access the dispatched commands the middleware received
   """
   def dispatched_commands(pluck \\ &(&1.command)) do
-    Agent.get(__MODULE__, fn %AuditLog{dispatched: dispatched} ->
+    Agent.get(@agent_name, fn %AuditLog{dispatched: dispatched} ->
       dispatched |> Enum.map(pluck)
     end)
   end
@@ -65,7 +67,7 @@ defmodule Commanded.Helpers.CommandAuditMiddleware do
   Access the dispatched commands that successfully executed
   """
   def succeeded_commands do
-    Agent.get(__MODULE__, fn %AuditLog{succeeded: succeeded} ->
+    Agent.get(@agent_name, fn %AuditLog{succeeded: succeeded} ->
       succeeded |> Enum.map(&(&1.command))
      end)
   end
@@ -74,7 +76,7 @@ defmodule Commanded.Helpers.CommandAuditMiddleware do
   Access the dispatched commands that failed to execute
   """
   def failed_commands do
-    Agent.get(__MODULE__, fn %AuditLog{failed: failed} ->
+    Agent.get(@agent_name, fn %AuditLog{failed: failed} ->
       failed |> Enum.map(&(&1.command))
     end)
   end
diff --git a/test/middleware/halting_middleware_test.exs b/test/middleware/halting_middleware_test.exs
index 11a945c..19af9ae 100644
--- a/test/middleware/halting_middleware_test.exs
+++ b/test/middleware/halting_middleware_test.exs
@@ -53,6 +53,7 @@ defmodule Commanded.Commands.Middleware.HaltingMiddlewareTest do
   end
 
   setup do
+    CommandAuditMiddleware.start_link()
     CommandAuditMiddleware.reset()
   end
 
diff --git a/test/middleware/middleware_test.exs b/test/middleware/middleware_test.exs
index 20d501e..2b87a77 100644
--- a/test/middleware/middleware_test.exs
+++ b/test/middleware/middleware_test.exs
@@ -61,6 +61,7 @@ defmodule Commanded.Commands.Middleware.MiddlewareTest do
   end
 
   setup do
+    CommandAuditMiddleware.start_link()
     CommandAuditMiddleware.reset()
   end
 
diff --git a/test/process_managers/process_manager_error_handling_test.exs b/test/process_managers/process_manager_error_handling_test.exs
index 62b36f6..d5b3230 100644
--- a/test/process_managers/process_manager_error_handling_test.exs
+++ b/test/process_managers/process_manager_error_handling_test.exs
@@ -1,140 +1,36 @@
 defmodule Commanded.ProcessManager.ProcessManagerErrorHandlingTest do
   use Commanded.StorageCase
 
-  defmodule ExampleAggregate do
-    @moduledoc false
-    defstruct [:process_uuid]
-
-    defmodule Commands do
-      defmodule StartProcess, do: defstruct [:process_uuid, :strategy, :delay, :reply_to]
-      defmodule AttemptProcess, do: defstruct [:process_uuid, :strategy, :delay, :reply_to]
-      defmodule ContinueProcess, do: defstruct [:process_uuid, :reply_to]
-    end
-
-    defmodule Events do
-      defmodule ProcessStarted, do: defstruct [:process_uuid, :strategy, :delay, :reply_to]
-      defmodule ProcessContinued, do: defstruct [:process_uuid, :reply_to]
-    end
-
-    alias Commands.{AttemptProcess,ContinueProcess,StartProcess}
-    alias Events.{ProcessContinued,ProcessStarted}
-
-    def execute(
-      %ExampleAggregate{},
-      %StartProcess{process_uuid: process_uuid, strategy: strategy, delay: delay, reply_to: reply_to})
-    do
-      %ProcessStarted{process_uuid: process_uuid, strategy: strategy, delay: delay, reply_to: reply_to}
-    end
-
-    def execute(%ExampleAggregate{}, %AttemptProcess{}),
-      do: {:error, :failed}
-
-    def execute(%ExampleAggregate{}, %ContinueProcess{process_uuid: process_uuid, reply_to: reply_to}),
-      do: %ProcessContinued{process_uuid: process_uuid, reply_to: reply_to}
-
-    def apply(%ExampleAggregate{} = aggregate, %ProcessStarted{process_uuid: process_uuid}),
-      do: %ExampleAggregate{aggregate | process_uuid: process_uuid}
-
-    def apply(%ExampleAggregate{} = aggregate, %ProcessContinued{}), do: aggregate
-  end
-
-  alias ExampleAggregate.Commands.{AttemptProcess,ContinueProcess,StartProcess}
-  alias ExampleAggregate.Events.{ProcessContinued,ProcessStarted}
-
-  defmodule ExampleRouter do
-    @moduledoc false
-    use Commanded.Commands.Router
-
-    dispatch [StartProcess,AttemptProcess,ContinueProcess],
-      to: ExampleAggregate, identity: :process_uuid
-  end
-
-  defmodule ErrorHandlingProcessManager do
-    @moduledoc false
-    use Commanded.ProcessManagers.ProcessManager,
-      name: "ErrorHandlingProcessManager",
-      router: ExampleRouter
-
-    defstruct [:process_uuid]
-
-    def interested?(%ProcessStarted{process_uuid: process_uuid}), do: {:start, process_uuid}
-    def interested?(%ProcessContinued{process_uuid: process_uuid}), do: {:continue, process_uuid}
-
-    def handle(
-      %ErrorHandlingProcessManager{},
-      %ProcessStarted{process_uuid: process_uuid, strategy: strategy, delay: delay, reply_to: reply_to})
-    do
-      %AttemptProcess{process_uuid: process_uuid, strategy: strategy, delay: delay, reply_to: reply_to}
-    end
-
-    def handle(%ErrorHandlingProcessManager{}, %ProcessContinued{reply_to: reply_to}) do
-      send(reply_to, :process_continued)
-      []
-    end
-
-    # stop after three attempts
-    def error({:error, :failed}, %AttemptProcess{strategy: :retry, reply_to: reply_to}, _pending_commands, %{attempts: attempts} = context)
-      when attempts >= 2
-    do
-      send(reply_to, {:error, :too_many_attempts, record_attempt(context)})
-
-      {:stop, :too_many_attempts}
-    end
-
-    # retry command with delay
-    def error({:error, :failed}, %AttemptProcess{strategy: :retry, delay: delay} = failed_command, _pending_commands, context)
-      when is_integer(delay)
-    do
-      context = record_attempt(context)
-      send_failure(failed_command, Map.put(context, :delay, delay))
-
-      {:retry, delay, context}
-    end
-
-    # retry command
-    def error({:error, :failed}, %AttemptProcess{strategy: :retry} = failed_command, _pending_commands, context) do
-      context = record_attempt(context)
-      send_failure(failed_command, context)
-
-      {:retry, context}
-    end
-
-    # skip failed command, continue pending
-    def error({:error, :failed}, %AttemptProcess{strategy: :skip, reply_to: reply_to}, _pending_commands, context) do
-      send(reply_to, {:error, :failed, record_attempt(context)})
-
-      {:skip, :continue_pending}
-    end
-
-    # continue with modified command
-    def error({:error, :failed}, %AttemptProcess{strategy: :continue, process_uuid: process_uuid, reply_to: reply_to}, pending_commands, context) do
-      context = record_attempt(context)
-      send(reply_to, {:error, :failed, context})
-
-      continue = %ContinueProcess{process_uuid: process_uuid, reply_to: reply_to}
-
-      {:continue, [continue | pending_commands], context}
-    end
-
-    defp record_attempt(context) do
-      Map.update(context, :attempts, 1, fn attempts -> attempts + 1 end)
-    end
-
-    defp send_failure(%AttemptProcess{reply_to: reply_to}, context) do
-      send(reply_to, {:error, :failed, context})
+  alias Commanded.Helpers.ProcessHelper
+  alias Commanded.ProcessManagers.{
+    ErrorHandlingProcessManager,
+    ErrorRouter,
+  }
+  alias Commanded.ProcessManagers.ErrorAggregate.Commands.StartProcess
+
+  setup do
+    reply_to = self()
+    {:ok, agent} = Agent.start_link(fn -> reply_to end, name: {:global, ErrorHandlingProcessManager})
+
+    on_exit fn ->
+      ProcessHelper.shutdown(agent)
     end
   end
 
   test "should retry the event until process manager requests stop" do
     process_uuid = UUID.uuid4()
-    command = %StartProcess{process_uuid: process_uuid, strategy: :retry, reply_to: self()}
+    command = %StartProcess{
+      process_uuid: process_uuid,
+      strategy: "retry",
+      reply_to: reply_to(),
+    }
 
     {:ok, process_router} = ErrorHandlingProcessManager.start_link()
 
     Process.unlink(process_router)
     ref = Process.monitor(process_router)
 
-    assert :ok = ExampleRouter.dispatch(command)
+    assert :ok = ErrorRouter.dispatch(command)
 
     assert_receive {:error, :failed, %{attempts: 1}}
     assert_receive {:error, :failed, %{attempts: 2}}
@@ -146,14 +42,19 @@ defmodule Commanded.ProcessManager.ProcessManagerErrorHandlingTest do
 
   test "should retry event with specified delay between attempts" do
     process_uuid = UUID.uuid4()
-    command = %StartProcess{process_uuid: process_uuid, strategy: :retry, delay: 10, reply_to: self()}
+    command = %StartProcess{
+      process_uuid: process_uuid,
+      strategy: "retry",
+      delay: 10,
+      reply_to: reply_to(),
+    }
 
     {:ok, process_router} = ErrorHandlingProcessManager.start_link()
 
     Process.unlink(process_router)
     ref = Process.monitor(process_router)
 
-    assert :ok = ExampleRouter.dispatch(command)
+    assert :ok = ErrorRouter.dispatch(command)
 
     assert_receive {:error, :failed, %{attempts: 1, delay: 10}}
     assert_receive {:error, :failed, %{attempts: 2, delay: 10}}
@@ -165,11 +66,15 @@ defmodule Commanded.ProcessManager.ProcessManagerErrorHandlingTest do
 
   test "should skip the event when error reply is `{:skip, :continue_pending}`" do
     process_uuid = UUID.uuid4()
-    command = %StartProcess{process_uuid: process_uuid, strategy: :skip, reply_to: self()}
+    command = %StartProcess{
+      process_uuid: process_uuid,
+      strategy: "skip",
+      reply_to: reply_to(),
+    }
 
     {:ok, process_router} = ErrorHandlingProcessManager.start_link()
 
-    assert :ok = ExampleRouter.dispatch(command)
+    assert :ok = ErrorRouter.dispatch(command)
 
     assert_receive {:error, :failed, %{attempts: 1}}
     refute_receive {:error, :failed, %{attempts: 2}}
@@ -180,11 +85,15 @@ defmodule Commanded.ProcessManager.ProcessManagerErrorHandlingTest do
 
   test "should continue with modified command" do
     process_uuid = UUID.uuid4()
-    command = %StartProcess{process_uuid: process_uuid, strategy: :continue, reply_to: self()}
+    command = %StartProcess{
+      process_uuid: process_uuid,
+      strategy: "continue",
+      reply_to: reply_to(),
+    }
 
     {:ok, process_router} = ErrorHandlingProcessManager.start_link()
 
-    assert :ok = ExampleRouter.dispatch(command)
+    assert :ok = ErrorRouter.dispatch(command)
 
     assert_receive {:error, :failed, %{attempts: 1}}
     assert_receive :process_continued
@@ -193,21 +102,6 @@ defmodule Commanded.ProcessManager.ProcessManagerErrorHandlingTest do
     assert Process.alive?(process_router)
   end
 
-  defmodule DefaultErrorHandlingProcessManager do
-    @moduledoc false
-    use Commanded.ProcessManagers.ProcessManager,
-      name: "DefaultErrorHandlingProcessManager",
-      router: ExampleRouter
-
-    defstruct [:process_uuid]
-
-    def interested?(%ProcessStarted{process_uuid: process_uuid}), do: {:start, process_uuid}
-
-    def handle(%ErrorHandlingProcessManager{}, %ProcessStarted{process_uuid: process_uuid}) do
-      %AttemptProcess{process_uuid: process_uuid}
-    end
-  end
-
   test "should stop process manager on error by default" do
     process_uuid = UUID.uuid4()
     command = %StartProcess{process_uuid: process_uuid}
@@ -217,10 +111,12 @@ defmodule Commanded.ProcessManager.ProcessManagerErrorHandlingTest do
     Process.unlink(process_router)
     ref = Process.monitor(process_router)
 
-    assert :ok = ExampleRouter.dispatch(command)
+    assert :ok = ErrorRouter.dispatch(command)
 
     # should shutdown process router
     assert_receive {:DOWN, ^ref, _, _, _}
     refute Process.alive?(process_router)
   end
+
+  defp reply_to, do: self() |> :erlang.pid_to_list()
 end
diff --git a/test/process_managers/process_manager_routing_test.exs b/test/process_managers/process_manager_routing_test.exs
index 99b3b52..5036de0 100644
--- a/test/process_managers/process_manager_routing_test.exs
+++ b/test/process_managers/process_manager_routing_test.exs
@@ -9,8 +9,14 @@ defmodule Commanded.ProcessManagers.ProcessManagerRoutingTest do
   alias Commanded.ExampleDomain.BankAccount.Events.{MoneyDeposited,MoneyWithdrawn}
   alias Commanded.ExampleDomain.MoneyTransfer.Commands.TransferMoney
   alias Commanded.ExampleDomain.MoneyTransfer.Events.MoneyTransferRequested
+  alias Commanded.Helpers.CommandAuditMiddleware
   alias Commanded.ProcessManagers.ProcessRouter
 
+  setup do
+    CommandAuditMiddleware.start_link()
+    :ok
+  end
+
   test "should start a process manager in response to an event" do
     account_number1 = UUID.uuid4
     account_number2 = UUID.uuid4
diff --git a/test/process_managers/support/error/default_error_handling_process_manager.ex b/test/process_managers/support/error/default_error_handling_process_manager.ex
new file mode 100644
index 0000000..ed49f59
--- /dev/null
+++ b/test/process_managers/support/error/default_error_handling_process_manager.ex
@@ -0,0 +1,26 @@
+defmodule Commanded.ProcessManagers.DefaultErrorHandlingProcessManager do
+  @moduledoc false
+
+  alias Commanded.ProcessManagers.{
+    DefaultErrorHandlingProcessManager,
+    ExampleRouter,
+  }
+  alias Commanded.ProcessManagers.ErrorAggregate.Commands.{
+    AttemptProcess,
+  }
+  alias Commanded.ProcessManagers.ErrorAggregate.Events.{
+    ProcessStarted,
+  }
+
+  use Commanded.ProcessManagers.ProcessManager,
+    name: "DefaultErrorHandlingProcessManager",
+    router: ExampleRouter
+
+  defstruct [:process_uuid]
+
+  def interested?(%ProcessStarted{process_uuid: process_uuid}), do: {:start, process_uuid}
+
+  def handle(%DefaultErrorHandlingProcessManager{}, %ProcessStarted{process_uuid: process_uuid}) do
+    %AttemptProcess{process_uuid: process_uuid}
+  end
+end
diff --git a/test/process_managers/support/error/error_aggregate.ex b/test/process_managers/support/error/error_aggregate.ex
new file mode 100644
index 0000000..231a34f
--- /dev/null
+++ b/test/process_managers/support/error/error_aggregate.ex
@@ -0,0 +1,37 @@
+defmodule Commanded.ProcessManagers.ErrorAggregate do
+  @moduledoc false
+  defstruct [:process_uuid]
+
+  defmodule Commands do
+    defmodule StartProcess, do: defstruct [:process_uuid, :strategy, :delay, :reply_to]
+    defmodule AttemptProcess, do: defstruct [:process_uuid, :strategy, :delay, :reply_to]
+    defmodule ContinueProcess, do: defstruct [:process_uuid, :reply_to]
+  end
+
+  defmodule Events do
+    defmodule ProcessStarted, do: defstruct [:process_uuid, :strategy, :delay, :reply_to]
+    defmodule ProcessContinued, do: defstruct [:process_uuid, :reply_to]
+  end
+
+  alias Commanded.ProcessManagers.ErrorAggregate
+  alias Commands.{AttemptProcess,ContinueProcess,StartProcess}
+  alias Events.{ProcessContinued,ProcessStarted}
+
+  def execute(
+    %ErrorAggregate{},
+    %StartProcess{process_uuid: process_uuid, strategy: strategy, delay: delay, reply_to: reply_to})
+  do
+    %ProcessStarted{process_uuid: process_uuid, strategy: strategy, delay: delay, reply_to: reply_to}
+  end
+
+  def execute(%ErrorAggregate{}, %AttemptProcess{}),
+    do: {:error, :failed}
+
+  def execute(%ErrorAggregate{}, %ContinueProcess{process_uuid: process_uuid, reply_to: reply_to}),
+    do: %ProcessContinued{process_uuid: process_uuid, reply_to: reply_to}
+
+  def apply(%ErrorAggregate{} = aggregate, %ProcessStarted{process_uuid: process_uuid}),
+    do: %ErrorAggregate{aggregate | process_uuid: process_uuid}
+
+  def apply(%ErrorAggregate{} = aggregate, %ProcessContinued{}), do: aggregate
+end
diff --git a/test/process_managers/support/error/error_handling_process_manager.ex b/test/process_managers/support/error/error_handling_process_manager.ex
new file mode 100644
index 0000000..b802cf4
--- /dev/null
+++ b/test/process_managers/support/error/error_handling_process_manager.ex
@@ -0,0 +1,95 @@
+defmodule Commanded.ProcessManagers.ErrorHandlingProcessManager do
+  @moduledoc false
+
+  alias Commanded.ProcessManagers.{
+    ErrorHandlingProcessManager,
+    ErrorRouter,
+  }
+  alias Commanded.ProcessManagers.ErrorAggregate.Commands.{
+    AttemptProcess,
+    ContinueProcess,
+  }
+  alias Commanded.ProcessManagers.ErrorAggregate.Events.{
+    ProcessContinued,
+    ProcessStarted,
+  }
+
+  use Commanded.ProcessManagers.ProcessManager,
+    name: "ErrorHandlingProcessManager",
+    router: ErrorRouter
+
+  defstruct [:process_uuid]
+
+  def interested?(%ProcessStarted{process_uuid: process_uuid}), do: {:start, process_uuid}
+  def interested?(%ProcessContinued{process_uuid: process_uuid}), do: {:continue, process_uuid}
+
+  def handle(
+    %ErrorHandlingProcessManager{},
+    %ProcessStarted{process_uuid: process_uuid, strategy: strategy, delay: delay})
+  do
+    %AttemptProcess{process_uuid: process_uuid, strategy: strategy, delay: delay}
+  end
+
+  def handle(%ErrorHandlingProcessManager{}, %ProcessContinued{}) do
+    reply(:process_continued)
+    []
+  end
+
+  # stop after three attempts
+  def error({:error, :failed}, %AttemptProcess{strategy: "retry"}, _pending_commands, %{attempts: attempts} = context)
+    when attempts >= 2
+  do
+    reply({:error, :too_many_attempts, record_attempt(context)})
+
+    {:stop, :too_many_attempts}
+  end
+
+  # retry command with delay
+  def error({:error, :failed}, %AttemptProcess{strategy: "retry", delay: delay}, _pending_commands, context)
+    when is_integer(delay)
+  do
+    context = record_attempt(context)
+    reply_failure(Map.put(context, :delay, delay))
+
+    {:retry, delay, context}
+  end
+
+  # retry command
+  def error({:error, :failed}, %AttemptProcess{strategy: "retry"}, _pending_commands, context) do
+    context = record_attempt(context)
+    reply_failure(context)
+
+    {:retry, context}
+  end
+
+  # skip failed command, continue pending
+  def error({:error, :failed}, %AttemptProcess{strategy: "skip"}, _pending_commands, context) do
+    reply({:error, :failed, record_attempt(context)})
+
+    {:skip, :continue_pending}
+  end
+
+  # continue with modified command
+  def error({:error, :failed}, %AttemptProcess{strategy: "continue", process_uuid: process_uuid}, pending_commands, context) do
+    context = record_attempt(context)
+    reply({:error, :failed, context})
+
+    continue = %ContinueProcess{process_uuid: process_uuid}
+
+    {:continue, [continue | pending_commands], context}
+  end
+
+  defp record_attempt(context) do
+    Map.update(context, :attempts, 1, fn attempts -> attempts + 1 end)
+  end
+
+  defp reply_failure(context) do
+    reply({:error, :failed, context})
+  end
+
+  defp reply(message) do
+    reply_to = Agent.get({:global, ErrorHandlingProcessManager}, fn reply_to -> reply_to end)
+
+    send(reply_to, message)
+  end
+end
diff --git a/test/process_managers/support/error/error_router.ex b/test/process_managers/support/error/error_router.ex
new file mode 100644
index 0000000..6e10090
--- /dev/null
+++ b/test/process_managers/support/error/error_router.ex
@@ -0,0 +1,14 @@
+defmodule Commanded.ProcessManagers.ErrorRouter do
+  @moduledoc false
+  use Commanded.Commands.Router
+
+  alias Commanded.ProcessManagers.ErrorAggregate
+  alias Commanded.ProcessManagers.ErrorAggregate.Commands.{
+    StartProcess,
+    AttemptProcess,
+    ContinueProcess,
+  }
+
+  dispatch [StartProcess,AttemptProcess,ContinueProcess],
+    to: ErrorAggregate, identity: :process_uuid
+end
diff --git a/test/subscriptions/subscriptions_test.exs b/test/subscriptions/subscriptions_test.exs
index 64a5f06..d3c3884 100644
--- a/test/subscriptions/subscriptions_test.exs
+++ b/test/subscriptions/subscriptions_test.exs
@@ -3,14 +3,9 @@ defmodule Commanded.SubscriptionsTest do
 
   alias Commanded.EventStore.RecordedEvent
   alias Commanded.Subscriptions
-  alias Commanded.Helpers.ProcessHelper
 
   setup do
-    Subscriptions.start_link([])
-
-    on_exit fn ->
-      ProcessHelper.shutdown(Subscriptions)
-    end
+    Subscriptions.reset()
   end
 
   describe "register event handler" do
diff --git a/test/test_helper.exs b/test/test_helper.exs
index cd8d559..869559e 100644
--- a/test/test_helper.exs
+++ b/test/test_helper.exs
@@ -1,3 +1,1 @@
 ExUnit.start()
-
-{:ok, _} = Commanded.Helpers.CommandAuditMiddleware.start_link()