Repository: commanded/commanded (elixir)

Issue:
Allow event handlers to subscribe to individual streams
By default, event handlers subscribe to all events appended to any stream. This approach causes issues when you have many events, but your handler is only interested in a subset of all events related to a single stream. 

It should be possible to subcribe a handler to a single stream, by its identity, using a `subscribe_to` option:

```elixir
defmodule ExampleHandler do
  use Commanded.Event.Handler, 
    name: __MODULE__,
    subscribe_to: "stream1234"

  def handle(%AnEvent{..}, metadata) do
    :ok
  end
end
```

In the above example, the configuration would guarantee only events appended to stream `stream1234` would be received and handled by the event handler.


Pull request description:
Allow event handlers to subscribe to individual streams
Event handlers subscribe to all events appended to any stream by default. For event handlers that are only interested in events appended to one stream it is now possible to subcribe to a single stream using the `subscribe_to` option:

```elixir
defmodule ExampleHandler do
  use Commanded.Event.Handler, 
    name: __MODULE__,
    subscribe_to: "stream1234"

  def handle(%AnEvent{..}, metadata) do
    :ok
  end
end
```

In the above example, the configuration would guarantee only events appended to stream `stream1234` would be received and handled by the event handler.

Closes #174.

Patch:
diff --git a/.travis.yml b/.travis.yml
index c01b57b..723b7d5 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -1,7 +1,7 @@
 language: elixir
 
 elixir:
-  - 1.7.2
+  - 1.7.3
 
 otp_release:
   - 21.0
diff --git a/CHANGELOG.md b/CHANGELOG.md
index d63bd5d..26ce82d 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -10,6 +10,7 @@
 - Allow aggregate identity to be of any type that implements the `String.Chars` protocol ([#166](https://github.com/commanded/commanded/pull/166)).
 - Process manager and event handler error & exception handling ([#192](https://github.com/commanded/commanded/pull/192)).
 - Process manager event handling timeout ([#193](https://github.com/commanded/commanded/pull/193)).
+- Allow event handlers to subscribe to individual streams ([#203](https://github.com/commanded/commanded/pull/203)).
 
 ### Bug fixes
 
diff --git a/guides/Events.md b/guides/Events.md
index 7c4be64..ba1cab9 100644
--- a/guides/Events.md
+++ b/guides/Events.md
@@ -62,6 +62,20 @@ Use the `:current` position when you don't want newly created event handlers to
 
 You should start your event handlers using a [supervisor](#supervision) to ensure they are restarted on error.
 
+### Subscribing to an individual stream
+
+By default event handlers will subscribe to all events appended to any stream. Provide a `subscribe_to` option to subscribe to a single stream.
+
+```elixir
+defmodule ExampleHandler do
+  use Commanded.Event.Handler,
+    name: __MODULE__,
+    subscribe_to: "stream1234"
+end
+```
+
+This will ensure the handler only receives events appended to that stream.
+
 ### `init/0` callback
 
 You can define an `init/0` function in your handler to be called once it has started and successfully subscribed to the event store.
diff --git a/lib/commanded/aggregates/aggregate.ex b/lib/commanded/aggregates/aggregate.ex
index c72e01e..8a2ba81 100644
--- a/lib/commanded/aggregates/aggregate.ex
+++ b/lib/commanded/aggregates/aggregate.ex
@@ -449,22 +449,23 @@ defmodule Commanded.Aggregates.Aggregate do
   defp persist_events(pending_events, aggregate_state, context, %Aggregate{} = state) do
     %Aggregate{aggregate_uuid: aggregate_uuid, aggregate_version: expected_version} = state
 
-    with {:ok, stream_version} <-
-           append_to_stream(pending_events, aggregate_uuid, expected_version, context) do
+    with :ok <- append_to_stream(pending_events, aggregate_uuid, expected_version, context) do
+      aggregate_version = expected_version + length(pending_events)
+
       state = %Aggregate{
         state
         | aggregate_state: aggregate_state,
-          aggregate_version: stream_version
+          aggregate_version: aggregate_version
       }
 
-      {{:ok, stream_version, pending_events}, state}
+      {{:ok, aggregate_version, pending_events}, state}
     else
       {:error, _error} = reply ->
         {reply, state}
     end
   end
 
-  defp append_to_stream([], _stream_uuid, expected_version, _context), do: {:ok, expected_version}
+  defp append_to_stream([], _stream_uuid, _expected_version, _context), do: :ok
 
   defp append_to_stream(pending_events, stream_uuid, expected_version, context) do
     %ExecutionContext{
@@ -473,7 +474,12 @@ defmodule Commanded.Aggregates.Aggregate do
       metadata: metadata
     } = context
 
-    event_data = Mapper.map_to_event_data(pending_events, causation_id, correlation_id, metadata)
+    event_data =
+      Mapper.map_to_event_data(pending_events,
+        causation_id: causation_id,
+        correlation_id: correlation_id,
+        metadata: metadata
+      )
 
     EventStore.append_to_stream(stream_uuid, expected_version, event_data)
   end
diff --git a/lib/commanded/assertions/event_assertions.ex b/lib/commanded/assertions/event_assertions.ex
index 5e92f95..cd81d30 100644
--- a/lib/commanded/assertions/event_assertions.ex
+++ b/lib/commanded/assertions/event_assertions.ex
@@ -174,10 +174,11 @@ defmodule Commanded.Assertions.EventAssertions do
   end
 
   defp create_subscription(subscription_name),
-    do: EventStore.subscribe_to_all_streams(subscription_name, self(), :origin)
+    do: EventStore.subscribe_to(:all, subscription_name, self(), :origin)
 
   defp remove_subscription(subscription_name),
-    do: EventStore.unsubscribe_from_all_streams(subscription_name)
+    do: EventStore.unsubscribe(subscription_name)
 
-  defp ack_events(subscription, events), do: EventStore.ack_event(subscription, List.last(events))
+  defp ack_events(subscription, events),
+    do: EventStore.ack_event(subscription, List.last(events))
 end
diff --git a/lib/commanded/event/handler.ex b/lib/commanded/event/handler.ex
index da3439f..db10686 100644
--- a/lib/commanded/event/handler.ex
+++ b/lib/commanded/event/handler.ex
@@ -65,6 +65,20 @@ defmodule Commanded.Event.Handler do
 
       {:ok, _handler} = ExampleHandler.start_link(start_from: :current)
 
+
+  ### Subscribing to an individual stream
+
+  By default event handlers will subscribe to all events appended to any stream.
+  Provide a `subscribe_to` option to subscribe to a single stream.
+
+      defmodule ExampleHandler do
+        use Commanded.Event.Handler,
+          name: __MODULE__,
+          subscribe_to: "stream1234"
+      end
+
+  This will ensure the handler only receives events appended to that stream.
+
   ## `c:init/0` callback
 
   You can define an `c:init/0` function in your handler to be called once it has
@@ -302,7 +316,7 @@ defmodule Commanded.Event.Handler do
     {valid, invalid} =
       module_opts
       |> Keyword.merge(local_opts)
-      |> Keyword.split([:consistency, :start_from] ++ additional_allowed_opts)
+      |> Keyword.split([:consistency, :start_from, :subscribe_to] ++ additional_allowed_opts)
 
     if Enum.any?(invalid) do
       raise "#{inspect(module)} specifies invalid options: #{inspect(Keyword.keys(invalid))}"
@@ -331,6 +345,7 @@ defmodule Commanded.Event.Handler do
     :handler_module,
     :last_seen_event,
     :subscribe_from,
+    :subscribe_to,
     :subscription
   ]
 
@@ -342,7 +357,8 @@ defmodule Commanded.Event.Handler do
       handler_name: handler_name,
       handler_module: handler_module,
       consistency: consistency(opts),
-      subscribe_from: start_from(opts)
+      subscribe_from: start_from(opts),
+      subscribe_to: subscribe_to(opts)
     }
 
     Registration.start_link(name, __MODULE__, handler)
@@ -367,14 +383,17 @@ defmodule Commanded.Event.Handler do
 
   @doc false
   def handle_call(:config, _from, %Handler{} = state) do
-    %Handler{consistency: consistency, subscribe_from: subscribe_from} = state
+    %Handler{consistency: consistency, subscribe_from: subscribe_from, subscribe_to: subscribe_to} =
+      state
 
-    {:reply, [consistency: consistency, start_from: subscribe_from], state}
+    config = [consistency: consistency, start_from: subscribe_from, subscribe_to: subscribe_to]
+
+    {:reply, config, state}
   end
 
   @doc false
   def handle_cast(:subscribe_to_events, %Handler{} = state) do
-    {:noreply, subscribe_to_all_streams(state)}
+    {:noreply, subscribe_to_events(state)}
   end
 
   @doc false
@@ -417,11 +436,15 @@ defmodule Commanded.Event.Handler do
     end
   end
 
-  defp subscribe_to_all_streams(%Handler{} = state) do
-    %Handler{handler_name: handler_name, subscribe_from: subscribe_from} = state
+  defp subscribe_to_events(%Handler{} = state) do
+    %Handler{
+      handler_name: handler_name,
+      subscribe_from: subscribe_from,
+      subscribe_to: subscribe_to
+    } = state
 
     {:ok, subscription} =
-      EventStore.subscribe_to_all_streams(handler_name, self(), subscribe_from)
+      EventStore.subscribe_to(subscribe_to, handler_name, self(), subscribe_from)
 
     %Handler{state | subscription: subscription}
   end
@@ -579,6 +602,14 @@ defmodule Commanded.Event.Handler do
     end
   end
 
+  defp subscribe_to(opts) do
+    case opts[:subscribe_to] || :all do
+      :all -> :all
+      stream when is_binary(stream) -> stream
+      invalid -> "Invalid `subscribe_to` option: #{inspect(invalid)}"
+    end
+  end
+
   defp describe(%Handler{handler_module: handler_module}),
     do: inspect(handler_module)
 end
diff --git a/lib/commanded/event/mapper.ex b/lib/commanded/event/mapper.ex
index 8954f5c..777b025 100644
--- a/lib/commanded/event/mapper.ex
+++ b/lib/commanded/event/mapper.ex
@@ -2,30 +2,25 @@ defmodule Commanded.Event.Mapper do
   @moduledoc false
 
   alias Commanded.EventStore.TypeProvider
-  alias Commanded.EventStore.{EventData,RecordedEvent}
+  alias Commanded.EventStore.{EventData, RecordedEvent}
 
-  def map_to_event_data(events, causation_id, correlation_id, metadata)
-    when is_list(events)
-  do
-    Enum.map(events, &map_to_event_data(&1, causation_id, correlation_id, metadata))
+  def map_to_event_data(events, fields) when is_list(events) do
+    Enum.map(events, &map_to_event_data(&1, fields))
   end
 
-  def map_to_event_data(event, causation_id, correlation_id, metadata) do
+  def map_to_event_data(event, fields) do
     %EventData{
-      causation_id: causation_id,
-      correlation_id: correlation_id,
+      causation_id: Keyword.get(fields, :causation_id),
+      correlation_id: Keyword.get(fields, :correlation_id),
       event_type: TypeProvider.to_string(event),
       data: event,
-      metadata: metadata,
+      metadata: Keyword.get(fields, :metadata, %{})
     }
   end
 
-  def map_from_recorded_events(recorded_events)
-    when is_list(recorded_events)
-  do
+  def map_from_recorded_events(recorded_events) when is_list(recorded_events) do
     Enum.map(recorded_events, &map_from_recorded_event/1)
   end
 
-  def map_from_recorded_event(%RecordedEvent{data: data}),
-    do: data
+  def map_from_recorded_event(%RecordedEvent{data: data}), do: data
 end
diff --git a/lib/commanded/event_store/adapters/in_memory.ex b/lib/commanded/event_store/adapters/in_memory.ex
index 46c5093..1e2a2be 100644
--- a/lib/commanded/event_store/adapters/in_memory.ex
+++ b/lib/commanded/event_store/adapters/in_memory.ex
@@ -10,34 +10,25 @@ defmodule Commanded.EventStore.Adapters.InMemory do
   defmodule State do
     @moduledoc false
 
-    defstruct serializer: nil,
-              persisted_events: [],
-              streams: %{},
-              transient_subscribers: %{},
-              persistent_subscriptions: %{},
-              snapshots: %{},
-              next_event_number: 1
+    defstruct [
+      :serializer,
+      persisted_events: [],
+      streams: %{},
+      transient_subscribers: %{},
+      persistent_subscriptions: %{},
+      snapshots: %{},
+      next_event_number: 1
+    ]
   end
 
   defmodule Subscription do
     @moduledoc false
 
-    defstruct name: nil,
-              subscriber: nil,
-              start_from: nil,
-              last_seen_event_number: 0
+    defstruct [:stream_uuid, :name, :subscriber, :start_from, last_seen_event_number: 0]
   end
 
-  alias Commanded.EventStore.Adapters.InMemory.{
-    State,
-    Subscription
-  }
-
-  alias Commanded.EventStore.{
-    EventData,
-    RecordedEvent,
-    SnapshotData
-  }
+  alias Commanded.EventStore.Adapters.InMemory.{State, Subscription}
+  alias Commanded.EventStore.{EventData, RecordedEvent, SnapshotData}
 
   def start_link(opts \\ []) do
     state = %State{
@@ -53,33 +44,30 @@ defmodule Commanded.EventStore.Adapters.InMemory do
   end
 
   @impl Commanded.EventStore
-  def append_to_stream(stream_uuid, expected_version, events)
-      when is_binary(stream_uuid) and is_integer(expected_version) and is_list(events) do
+  def append_to_stream(stream_uuid, expected_version, events) do
     GenServer.call(__MODULE__, {:append_to_stream, stream_uuid, expected_version, events})
   end
 
   @impl Commanded.EventStore
-  def stream_forward(stream_uuid, start_version \\ 0, read_batch_size \\ 1_000)
-
-  def stream_forward(stream_uuid, start_version, read_batch_size)
-      when is_binary(stream_uuid) and is_integer(start_version) and is_integer(read_batch_size) do
+  def stream_forward(stream_uuid, start_version \\ 0, _read_batch_size \\ 1_000) do
     GenServer.call(__MODULE__, {:stream_forward, stream_uuid, start_version})
   end
 
   @impl Commanded.EventStore
-  def subscribe(stream_uuid) when is_binary(stream_uuid) do
+  def subscribe(stream_uuid) do
     GenServer.call(__MODULE__, {:subscribe, stream_uuid, self()})
   end
 
   @impl Commanded.EventStore
-  def subscribe_to_all_streams(subscription_name, subscriber, start_from) do
+  def subscribe_to(stream_uuid, subscription_name, subscriber, start_from) do
     subscription = %Subscription{
+      stream_uuid: stream_uuid,
       name: subscription_name,
       subscriber: subscriber,
       start_from: start_from
     }
 
-    GenServer.call(__MODULE__, {:subscribe_to_all_streams, subscription})
+    GenServer.call(__MODULE__, {:subscribe_to, subscription})
   end
 
   @impl Commanded.EventStore
@@ -88,8 +76,8 @@ defmodule Commanded.EventStore.Adapters.InMemory do
   end
 
   @impl Commanded.EventStore
-  def unsubscribe_from_all_streams(subscription_name) do
-    GenServer.call(__MODULE__, {:unsubscribe_from_all_streams, subscription_name})
+  def unsubscribe(subscription_name) do
+    GenServer.call(__MODULE__, {:unsubscribe, subscription_name})
   end
 
   @impl Commanded.EventStore
@@ -115,36 +103,42 @@ defmodule Commanded.EventStore.Adapters.InMemory do
   def handle_call(
         {:append_to_stream, stream_uuid, expected_version, events},
         _from,
-        %State{streams: streams} = state
+        %State{} = state
       ) do
+    %State{streams: streams} = state
+
     case Map.get(streams, stream_uuid) do
       nil ->
         case expected_version do
           0 ->
-            {reply, state} = persist_events(stream_uuid, [], events, state)
+            {:ok, state} = persist_events(stream_uuid, [], events, state)
+
+            {:reply, :ok, state}
 
-            {:reply, reply, state}
+          :any_version ->
+            {:ok, state} = persist_events(stream_uuid, [], events, state)
+
+            {:reply, :ok, state}
 
           _ ->
             {:reply, {:error, :wrong_expected_version}, state}
         end
 
-      existing_events when length(existing_events) != expected_version ->
+      existing_events
+      when length(existing_events) != expected_version and expected_version != :any_version ->
         {:reply, {:error, :wrong_expected_version}, state}
 
       existing_events ->
-        {reply, state} = persist_events(stream_uuid, existing_events, events, state)
+        {:ok, state} = persist_events(stream_uuid, existing_events, events, state)
 
-        {:reply, reply, state}
+        {:reply, :ok, state}
     end
   end
 
   @impl GenServer
-  def handle_call(
-        {:stream_forward, stream_uuid, start_version},
-        _from,
-        %State{streams: streams} = state
-      ) do
+  def handle_call({:stream_forward, stream_uuid, start_version}, _from, %State{} = state) do
+    %State{streams: streams} = state
+
     reply =
       case Map.get(streams, stream_uuid) do
         nil ->
@@ -174,12 +168,10 @@ defmodule Commanded.EventStore.Adapters.InMemory do
   end
 
   @impl GenServer
-  def handle_call(
-        {:subscribe_to_all_streams,
-         %Subscription{name: subscription_name, subscriber: subscriber} = subscription},
-        _from,
-        %State{persistent_subscriptions: subscriptions} = state
-      ) do
+  def handle_call({:subscribe_to, %Subscription{} = subscription}, _from, %State{} = state) do
+    %Subscription{name: subscription_name, subscriber: subscriber} = subscription
+    %State{persistent_subscriptions: subscriptions} = state
+
     {reply, state} =
       case Map.get(subscriptions, subscription_name) do
         nil ->
@@ -201,11 +193,9 @@ defmodule Commanded.EventStore.Adapters.InMemory do
   end
 
   @impl GenServer
-  def handle_call(
-        {:unsubscribe_from_all_streams, subscription_name},
-        _from,
-        %State{persistent_subscriptions: subscriptions} = state
-      ) do
+  def handle_call({:unsubscribe, subscription_name}, _from, %State{} = state) do
+    %State{persistent_subscriptions: subscriptions} = state
+
     state = %State{state | persistent_subscriptions: Map.delete(subscriptions, subscription_name)}
 
     {:reply, :ok, state}
@@ -245,10 +235,9 @@ defmodule Commanded.EventStore.Adapters.InMemory do
   end
 
   @impl GenServer
-  def handle_cast(
-        {:ack_event, event, subscriber},
-        %State{persistent_subscriptions: subscriptions} = state
-      ) do
+  def handle_cast({:ack_event, event, subscriber}, %State{} = state) do
+    %State{persistent_subscriptions: subscriptions} = state
+
     state = %State{
       state
       | persistent_subscriptions: ack_subscription_by_pid(subscriptions, event, subscriber)
@@ -270,16 +259,13 @@ defmodule Commanded.EventStore.Adapters.InMemory do
     {:noreply, state}
   end
 
-  defp persist_events(
-         stream_uuid,
-         existing_events,
-         new_events,
-         %State{
-           persisted_events: persisted_events,
-           streams: streams,
-           next_event_number: next_event_number
-         } = state
-       ) do
+  defp persist_events(stream_uuid, existing_events, new_events, %State{} = state) do
+    %State{
+      persisted_events: persisted_events,
+      streams: streams,
+      next_event_number: next_event_number
+    } = state
+
     initial_stream_version = length(existing_events) + 1
     now = NaiveDateTime.utc_now()
 
@@ -307,9 +293,9 @@ defmodule Commanded.EventStore.Adapters.InMemory do
     publish_events = Enum.map(new_events, &deserialize(&1, state))
 
     publish_to_transient_subscribers(stream_uuid, publish_events, state)
-    publish_to_persistent_subscriptions(publish_events, state)
+    publish_to_persistent_subscriptions(stream_uuid, publish_events, state)
 
-    {{:ok, length(stream_events)}, state}
+    {:ok, state}
   end
 
   defp map_to_recorded_event(event_number, stream_uuid, stream_version, now, %EventData{} = event) do
@@ -410,15 +396,23 @@ defmodule Commanded.EventStore.Adapters.InMemory do
   end
 
   # publish events to subscribers
-  defp publish_to_persistent_subscriptions(events, %State{persistent_subscriptions: subscriptions}) do
+  defp publish_to_persistent_subscriptions(stream_uuid, events, %State{} = state) do
+    %State{persistent_subscriptions: subscriptions} = state
+
     subscribers =
       subscriptions
       |> Map.values()
+      |> Enum.filter(fn
+        %Subscription{stream_uuid: ^stream_uuid} -> true
+        %Subscription{stream_uuid: :all} -> true
+        %Subscription{} -> false
+      end)
       |> Enum.map(& &1.subscriber)
       |> Enum.filter(&is_pid/1)
 
-    for subscriber <- subscribers,
-        do: send(subscriber, {:events, events})
+    for subscriber <- subscribers do
+      send(subscriber, {:events, events})
+    end
   end
 
   defp serialize(data, %State{serializer: nil}), do: data
diff --git a/lib/commanded/event_store/event_store.ex b/lib/commanded/event_store/event_store.ex
index f94eb4b..459e05e 100644
--- a/lib/commanded/event_store/event_store.ex
+++ b/lib/commanded/event_store/event_store.ex
@@ -7,7 +7,7 @@ defmodule Commanded.EventStore do
 
   @type stream_uuid :: String.t()
   @type start_from :: :origin | :current | integer
-  @type stream_version :: integer
+  @type expected_version :: :any_version | :no_stream | :stream_exists | non_neg_integer()
   @type subscription_name :: String.t()
   @type source_uuid :: String.t()
   @type snapshot :: SnapshotData.t()
@@ -18,10 +18,10 @@ defmodule Commanded.EventStore do
   """
   @callback append_to_stream(
               stream_uuid,
-              expected_version :: non_neg_integer,
+              expected_version,
               events :: list(EventData.t())
             ) ::
-              {:ok, stream_version}
+              :ok
               | {:error, :wrong_expected_version}
               | {:error, error}
 
@@ -46,10 +46,13 @@ defmodule Commanded.EventStore do
 
   The subscriber does not need to acknowledge receipt of the events.
   """
-  @callback subscribe(stream_uuid) :: :ok | {:error, error}
+  @callback subscribe(stream_uuid | :all) :: :ok | {:error, error}
 
   @doc """
-  Create a persistent subscription to all event streams.
+  Create a persistent subscription to an event stream.
+
+  To subscribe to all events appended to any stream use `:all` as the stream
+  when subscribing.
 
   The event store will remember the subscribers last acknowledged event.
   Restarting the named subscription will resume from the next event following
@@ -65,8 +68,19 @@ defmodule Commanded.EventStore do
 
   The subscriber must ack each received, and successfully processed event, using
   `Commanded.EventStore.ack_event/2`.
+
+  ## Examples
+
+  Subscribe to all streams:
+
+      {:ok, subscription} = Commanded.EventStore.subscribe_to(:all, "Example", self())
+
+  Subscribe to a single stream:
+
+      {:ok, subscription} = Commanded.EventStore.subscribe_to("stream1234", "Example", self())
+
   """
-  @callback subscribe_to_all_streams(subscription_name, subscriber :: pid, start_from) ::
+  @callback subscribe_to(stream_uuid | :all, subscription_name, subscriber :: pid, start_from) ::
               {:ok, subscription :: pid}
               | {:error, :subscription_already_exists}
               | {:error, error}
@@ -78,9 +92,9 @@ defmodule Commanded.EventStore do
   @callback ack_event(pid, RecordedEvent.t()) :: :ok
 
   @doc """
-  Unsubscribe an existing subscriber from all event notifications.
+  Unsubscribe an existing subscriber from event notifications.
   """
-  @callback unsubscribe_from_all_streams(subscription_name) :: :ok
+  @callback unsubscribe(subscription_name) :: :ok
 
   @doc """
   Read a snapshot, if available, for a given source.
@@ -102,11 +116,14 @@ defmodule Commanded.EventStore do
   """
   @spec append_to_stream(
           stream_uuid,
-          expected_version :: non_neg_integer,
+          expected_version,
           events :: list(EventData.t())
-        ) :: {:ok, stream_version} | {:error, :wrong_expected_version} | {:error, error}
+        ) :: :ok | {:error, :wrong_expected_version} | {:error, error}
   def append_to_stream(stream_uuid, expected_version, events)
-      when is_binary(stream_uuid) and is_integer(expected_version) and is_list(events) do
+      when is_binary(stream_uuid) and
+             (is_integer(expected_version) or
+                expected_version in [:any_version, :no_stream, :stream_exists]) and
+             is_list(events) do
     event_store_adapter().append_to_stream(stream_uuid, expected_version, events)
   end
 
@@ -129,21 +146,21 @@ defmodule Commanded.EventStore do
   @doc """
   Create a transient subscription to a single event stream.
   """
-  @spec subscribe(stream_uuid) :: :ok | {:error, error}
+  @spec subscribe(stream_uuid | :all) :: :ok | {:error, error}
   def subscribe(stream_uuid) when is_binary(stream_uuid) do
     event_store_adapter().subscribe(stream_uuid)
   end
 
   @doc """
-  Create a persistent subscription to all event streams.
+  Create a persistent subscription to an event stream.
   """
-  @spec subscribe_to_all_streams(subscription_name, subscriber :: pid, start_from) ::
+  @spec subscribe_to(stream_uuid | :all, subscription_name, subscriber :: pid, start_from) ::
           {:ok, subscription :: pid}
           | {:error, :subscription_already_exists}
           | {:error, error}
-  def subscribe_to_all_streams(subscription_name, subscriber, start_from)
+  def subscribe_to(stream_uuid, subscription_name, subscriber, start_from)
       when is_binary(subscription_name) and is_pid(subscriber) do
-    event_store_adapter().subscribe_to_all_streams(subscription_name, subscriber, start_from)
+    event_store_adapter().subscribe_to(stream_uuid, subscription_name, subscriber, start_from)
   end
 
   @doc """
@@ -158,9 +175,9 @@ defmodule Commanded.EventStore do
   @doc """
   Unsubscribe an existing subscriber from all event notifications.
   """
-  @spec unsubscribe_from_all_streams(subscription_name) :: :ok
-  def unsubscribe_from_all_streams(subscription_name) when is_binary(subscription_name) do
-    event_store_adapter().unsubscribe_from_all_streams(subscription_name)
+  @spec unsubscribe(subscription_name) :: :ok
+  def unsubscribe(subscription_name) when is_binary(subscription_name) do
+    event_store_adapter().unsubscribe(subscription_name)
   end
 
   @doc """
diff --git a/lib/commanded/process_managers/process_router.ex b/lib/commanded/process_managers/process_router.ex
index 9adbb22..71969ed 100644
--- a/lib/commanded/process_managers/process_router.ex
+++ b/lib/commanded/process_managers/process_router.ex
@@ -198,7 +198,7 @@ defmodule Commanded.ProcessManagers.ProcessRouter do
       [] ->
         {:noreply, state}
 
-      pending ->
+      _pending ->
         Logger.error(fn ->
           describe(state) <>
             " has taken longer than " <>
@@ -228,7 +228,7 @@ defmodule Commanded.ProcessManagers.ProcessRouter do
     %State{process_manager_name: process_manager_name, subscribe_from: subscribe_from} = state
 
     {:ok, subscription} =
-      EventStore.subscribe_to_all_streams(process_manager_name, self(), subscribe_from)
+      EventStore.subscribe_to(:all, process_manager_name, self(), subscribe_from)
 
     %State{state | subscription: subscription}
   end


Test patch:
diff --git a/config/test.exs b/config/test.exs
index 650660b..dfcc96c 100644
--- a/config/test.exs
+++ b/config/test.exs
@@ -1,14 +1,11 @@
 use Mix.Config
 
-alias Commanded.EventStore.Adapters.InMemory
-alias Commanded.Serialization.JsonSerializer
-
 config :logger, :console, level: :warn, format: "[$level] $message\n"
 
 config :ex_unit,
   capture_log: true,
-  assert_receive_timeout: 200
+  assert_receive_timeout: 1_000
 
 config :commanded,
   dispatch_consistency_timeout: 100,
-  event_store_adapter: InMemory
+  event_store_adapter: Commanded.EventStore.Adapters.InMemory
diff --git a/test/aggregates/aggregate_concurrency_test.exs b/test/aggregates/aggregate_concurrency_test.exs
index fc3568a..d538362 100644
--- a/test/aggregates/aggregate_concurrency_test.exs
+++ b/test/aggregates/aggregate_concurrency_test.exs
@@ -8,8 +8,9 @@ defmodule Commanded.Aggregates.AggregateConcurrencyTest do
   alias Commanded.ExampleDomain.BankAccount.Events.MoneyDeposited
 
   setup do
-    expect(MockEventStore, :subscribe_to_all_streams, fn _handler_name, handler, _subscribe_from ->
-      {:ok, handler}
+    expect(MockEventStore, :subscribe_to, fn
+      _stream_uuid, _handler_name, handler, _subscribe_from ->
+        {:ok, handler}
     end)
 
     expect(MockEventStore, :subscribe, fn _aggregate_uuid -> :ok end)
@@ -64,9 +65,7 @@ defmodule Commanded.Aggregates.AggregateConcurrencyTest do
       end)
 
       # succeed on second attempt
-      expect(MockEventStore, :append_to_stream, fn ^account_number, 2, event_data ->
-        {:ok, 2 + length(event_data)}
-      end)
+      expect(MockEventStore, :append_to_stream, fn ^account_number, 2, _event_data -> :ok end)
 
       assert {:ok, 3, _events} = Aggregate.execute(BankAccount, account_number, context)
 
@@ -109,13 +108,8 @@ defmodule Commanded.Aggregates.AggregateConcurrencyTest do
     defp open_account(_context) do
       account_number = UUID.uuid4()
 
-      expect(MockEventStore, :stream_forward, fn ^account_number, 1, _batch_size ->
-        []
-      end)
-
-      expect(MockEventStore, :append_to_stream, fn ^account_number, 0, event_data ->
-        {:ok, length(event_data)}
-      end)
+      expect(MockEventStore, :stream_forward, fn ^account_number, 1, _batch_size -> [] end)
+      expect(MockEventStore, :append_to_stream, fn ^account_number, 0, _event_data -> :ok end)
 
       {:ok, ^account_number} =
         Commanded.Aggregates.Supervisor.open_aggregate(BankAccount, account_number)
diff --git a/test/aggregates/aggregate_lifespan_test.exs b/test/aggregates/aggregate_lifespan_test.exs
index 9879fcb..7d56821 100644
--- a/test/aggregates/aggregate_lifespan_test.exs
+++ b/test/aggregates/aggregate_lifespan_test.exs
@@ -42,7 +42,7 @@ defmodule Commanded.Aggregates.AggregateLifespanTest do
       :ok = BankRouter.dispatch(%OpenAccount{account_number: aggregate_uuid, initial_balance: 10})
       :ok = BankRouter.dispatch(%DepositMoney{account_number: aggregate_uuid, amount: 10})
 
-      refute_receive {:DOWN, ^ref, :process, _, :normal}, 10
+      refute_receive {:DOWN, ^ref, :process, _, :normal}, 25
       assert_receive {:DOWN, ^ref, :process, _, :normal}
     end
 
@@ -53,7 +53,7 @@ defmodule Commanded.Aggregates.AggregateLifespanTest do
       :ok = BankRouter.dispatch(%OpenAccount{account_number: aggregate_uuid, initial_balance: 10})
       :ok = BankRouter.dispatch(%WithdrawMoney{account_number: aggregate_uuid, amount: 10})
 
-      refute_receive {:DOWN, ^ref, :process, _, :normal}, 30
+      refute_receive {:DOWN, ^ref, :process, _, :normal}, 100
     end
 
     test "should stop process when requested", %{
diff --git a/test/aggregates/aggregate_subscription_test.exs b/test/aggregates/aggregate_subscription_test.exs
index e427f04..8c3d041 100644
--- a/test/aggregates/aggregate_subscription_test.exs
+++ b/test/aggregates/aggregate_subscription_test.exs
@@ -102,9 +102,7 @@ defmodule Commanded.Aggregates.AggregateSubscriptionTest do
         }
       }
 
-      {:ok, _} = EventStore.append_to_stream(account_number, 1, [event])
-
-      :ok
+      EventStore.append_to_stream(account_number, 1, [event])
     end
   end
 end
diff --git a/test/aggregates/event_persistence_test.exs b/test/aggregates/event_persistence_test.exs
index 8c1c6e5..d504d5c 100644
--- a/test/aggregates/event_persistence_test.exs
+++ b/test/aggregates/event_persistence_test.exs
@@ -3,18 +3,25 @@ defmodule Commanded.Aggregates.EventPersistenceTest do
 
   import Commanded.Enumerable, only: [pluck: 2]
 
-  alias Commanded.Aggregates.{Aggregate,AppendItemsHandler,ExampleAggregate}
-  alias Commanded.Aggregates.ExampleAggregate.Commands.{AppendItems,NoOp}
+  alias Commanded.Aggregates.{Aggregate, AppendItemsHandler, ExampleAggregate}
+  alias Commanded.Aggregates.ExampleAggregate.Commands.{AppendItems, NoOp}
   alias Commanded.EventStore
   alias Commanded.Helpers.ProcessHelper
   alias Commanded.Aggregates.ExecutionContext
 
   test "should persist pending events in order applied" do
-    aggregate_uuid = UUID.uuid4
+    aggregate_uuid = UUID.uuid4()
+
+    {:ok, ^aggregate_uuid} =
+      Commanded.Aggregates.Supervisor.open_aggregate(ExampleAggregate, aggregate_uuid)
 
-    {:ok, ^aggregate_uuid} = Commanded.Aggregates.Supervisor.open_aggregate(ExampleAggregate, aggregate_uuid)
+    {:ok, 10, events} =
+      Aggregate.execute(ExampleAggregate, aggregate_uuid, %ExecutionContext{
+        command: %AppendItems{count: 10},
+        handler: AppendItemsHandler,
+        function: :handle
+      })
 
-    {:ok, 10, events} = Aggregate.execute(ExampleAggregate, aggregate_uuid, %ExecutionContext{command: %AppendItems{count: 10}, handler: AppendItemsHandler, function: :handle})
     assert length(events) == 10
 
     recorded_events = EventStore.stream_forward(aggregate_uuid, 0) |> Enum.to_list()
@@ -28,14 +35,27 @@ defmodule Commanded.Aggregates.EventPersistenceTest do
   end
 
   test "should not persist events when command returns no events" do
-    aggregate_uuid = UUID.uuid4
+    aggregate_uuid = UUID.uuid4()
 
-    {:ok, ^aggregate_uuid} = Commanded.Aggregates.Supervisor.open_aggregate(ExampleAggregate, aggregate_uuid)
+    {:ok, ^aggregate_uuid} =
+      Commanded.Aggregates.Supervisor.open_aggregate(ExampleAggregate, aggregate_uuid)
+
+    {:ok, 1, events} =
+      Aggregate.execute(ExampleAggregate, aggregate_uuid, %ExecutionContext{
+        command: %AppendItems{count: 1},
+        handler: AppendItemsHandler,
+        function: :handle
+      })
 
-    {:ok, 1, events} = Aggregate.execute(ExampleAggregate, aggregate_uuid, %ExecutionContext{command: %AppendItems{count: 1}, handler: AppendItemsHandler, function: :handle})
     assert length(events) == 1
 
-    {:ok, 1, events} = Aggregate.execute(ExampleAggregate, aggregate_uuid, %ExecutionContext{command: %NoOp{}, handler: ExampleAggregate, function: :noop})
+    {:ok, 1, events} =
+      Aggregate.execute(ExampleAggregate, aggregate_uuid, %ExecutionContext{
+        command: %NoOp{},
+        handler: ExampleAggregate,
+        function: :noop
+      })
+
     assert length(events) == 0
 
     recorded_events = EventStore.stream_forward(aggregate_uuid, 0) |> Enum.to_list()
@@ -43,12 +63,19 @@ defmodule Commanded.Aggregates.EventPersistenceTest do
   end
 
   test "should persist event metadata" do
-    aggregate_uuid = UUID.uuid4
+    aggregate_uuid = UUID.uuid4()
 
-    {:ok, ^aggregate_uuid} = Commanded.Aggregates.Supervisor.open_aggregate(ExampleAggregate, aggregate_uuid)
+    {:ok, ^aggregate_uuid} =
+      Commanded.Aggregates.Supervisor.open_aggregate(ExampleAggregate, aggregate_uuid)
 
     metadata = %{"ip_address" => "127.0.0.1"}
-    context = %ExecutionContext{command: %AppendItems{count: 10}, metadata: metadata, handler: AppendItemsHandler, function: :handle}
+
+    context = %ExecutionContext{
+      command: %AppendItems{count: 10},
+      metadata: metadata,
+      handler: AppendItemsHandler,
+      function: :handle
+    }
 
     {:ok, 10, events} = Aggregate.execute(ExampleAggregate, aggregate_uuid, context)
     assert length(events) == 10
@@ -61,42 +88,71 @@ defmodule Commanded.Aggregates.EventPersistenceTest do
   end
 
   test "should reload persisted events when restarting aggregate process" do
-    aggregate_uuid = UUID.uuid4
+    aggregate_uuid = UUID.uuid4()
 
-    {:ok, ^aggregate_uuid} = Commanded.Aggregates.Supervisor.open_aggregate(ExampleAggregate, aggregate_uuid)
+    {:ok, ^aggregate_uuid} =
+      Commanded.Aggregates.Supervisor.open_aggregate(ExampleAggregate, aggregate_uuid)
+
+    {:ok, 10, events} =
+      Aggregate.execute(ExampleAggregate, aggregate_uuid, %ExecutionContext{
+        command: %AppendItems{count: 10},
+        handler: AppendItemsHandler,
+        function: :handle
+      })
 
-    {:ok, 10, events} = Aggregate.execute(ExampleAggregate, aggregate_uuid, %ExecutionContext{command: %AppendItems{count: 10}, handler: AppendItemsHandler, function: :handle})
     assert length(events) == 10
 
     ProcessHelper.shutdown_aggregate(ExampleAggregate, aggregate_uuid)
 
-    {:ok, ^aggregate_uuid} = Commanded.Aggregates.Supervisor.open_aggregate(ExampleAggregate, aggregate_uuid)
+    {:ok, ^aggregate_uuid} =
+      Commanded.Aggregates.Supervisor.open_aggregate(ExampleAggregate, aggregate_uuid)
 
     assert Aggregate.aggregate_version(ExampleAggregate, aggregate_uuid) == 10
+
     assert Aggregate.aggregate_state(ExampleAggregate, aggregate_uuid) == %ExampleAggregate{
-      items: 1..10 |> Enum.to_list(),
-      last_index: 10,
-    }
+             items: 1..10 |> Enum.to_list(),
+             last_index: 10
+           }
   end
 
   test "should reload persisted events in batches when restarting aggregate process" do
-    aggregate_uuid = UUID.uuid4
-
-    {:ok, ^aggregate_uuid} = Commanded.Aggregates.Supervisor.open_aggregate(ExampleAggregate, aggregate_uuid)
+    aggregate_uuid = UUID.uuid4()
 
-    {:ok, 100, _events} = Aggregate.execute(ExampleAggregate, aggregate_uuid, %ExecutionContext{command: %AppendItems{count: 100}, handler: AppendItemsHandler, function: :handle})
-    {:ok, 200, _events} = Aggregate.execute(ExampleAggregate, aggregate_uuid, %ExecutionContext{command: %AppendItems{count: 100}, handler: AppendItemsHandler, function: :handle})
-    {:ok, 201, _events} = Aggregate.execute(ExampleAggregate, aggregate_uuid, %ExecutionContext{command: %AppendItems{count: 1}, handler: AppendItemsHandler, function: :handle})
+    {:ok, ^aggregate_uuid} =
+      Commanded.Aggregates.Supervisor.open_aggregate(ExampleAggregate, aggregate_uuid)
+
+    {:ok, 100, _events} =
+      Aggregate.execute(ExampleAggregate, aggregate_uuid, %ExecutionContext{
+        command: %AppendItems{count: 100},
+        handler: AppendItemsHandler,
+        function: :handle
+      })
+
+    {:ok, 200, _events} =
+      Aggregate.execute(ExampleAggregate, aggregate_uuid, %ExecutionContext{
+        command: %AppendItems{count: 100},
+        handler: AppendItemsHandler,
+        function: :handle
+      })
+
+    {:ok, 201, _events} =
+      Aggregate.execute(ExampleAggregate, aggregate_uuid, %ExecutionContext{
+        command: %AppendItems{count: 1},
+        handler: AppendItemsHandler,
+        function: :handle
+      })
 
     ProcessHelper.shutdown_aggregate(ExampleAggregate, aggregate_uuid)
 
-    {:ok, ^aggregate_uuid} = Commanded.Aggregates.Supervisor.open_aggregate(ExampleAggregate, aggregate_uuid)
+    {:ok, ^aggregate_uuid} =
+      Commanded.Aggregates.Supervisor.open_aggregate(ExampleAggregate, aggregate_uuid)
 
     assert Aggregate.aggregate_version(ExampleAggregate, aggregate_uuid) == 201
+
     assert Aggregate.aggregate_state(ExampleAggregate, aggregate_uuid) == %ExampleAggregate{
-      items: 1..201 |> Enum.to_list,
-      last_index: 201,
-    }
+             items: 1..201 |> Enum.to_list(),
+             last_index: 201
+           }
   end
 
   test "should prefix stream UUID with aggregate indentity prefix" do
@@ -104,9 +160,14 @@ defmodule Commanded.Aggregates.EventPersistenceTest do
     prefix = "example-prefix-"
     prefixed_aggregate_uuid = prefix <> aggregate_uuid
 
-    {:ok, ^prefixed_aggregate_uuid} = Commanded.Aggregates.Supervisor.open_aggregate(ExampleAggregate, prefixed_aggregate_uuid)
+    {:ok, ^prefixed_aggregate_uuid} =
+      Commanded.Aggregates.Supervisor.open_aggregate(ExampleAggregate, prefixed_aggregate_uuid)
 
-    context = %ExecutionContext{command: %AppendItems{count: 1}, handler: AppendItemsHandler, function: :handle}
+    context = %ExecutionContext{
+      command: %AppendItems{count: 1},
+      handler: AppendItemsHandler,
+      function: :handle
+    }
 
     {:ok, 1, events} = Aggregate.execute(ExampleAggregate, prefixed_aggregate_uuid, context)
     assert length(events) == 1
diff --git a/test/aggregates/support/bank_account_lifespan.ex b/test/aggregates/support/bank_account_lifespan.ex
index 7b366b2..527ef0c 100644
--- a/test/aggregates/support/bank_account_lifespan.ex
+++ b/test/aggregates/support/bank_account_lifespan.ex
@@ -9,8 +9,8 @@ defmodule Commanded.Aggregates.BankAccountLifespan do
     MoneyDeposited
   }
 
-  def after_event(%BankAccountOpened{}), do: 5
-  def after_event(%MoneyDeposited{}), do: 20
+  def after_event(%BankAccountOpened{}), do: 25
+  def after_event(%MoneyDeposited{}), do: 50
   def after_event(%BankAccountClosed{}), do: :stop
   def after_event(_event), do: :infinity
 end
diff --git a/test/commands/command_timeout_test.exs b/test/commands/command_timeout_test.exs
index 175e9bc..e02c797 100644
--- a/test/commands/command_timeout_test.exs
+++ b/test/commands/command_timeout_test.exs
@@ -1,22 +1,26 @@
 defmodule Commanded.Commands.CommandTimeoutTest do
   use Commanded.StorageCase
 
-  alias Commanded.Commands.{TimeoutRouter,TimeoutCommand}
+  alias Commanded.Commands.{TimeoutRouter, TimeoutCommand}
 
   test "should allow timeout to be specified during command registration" do
     # handler is set to take longer than the configured timeout
-    case TimeoutRouter.dispatch(%TimeoutCommand{aggregate_uuid: UUID.uuid4, sleep_in_ms: 2_000}) do
-      {:error, :aggregate_execution_failed}  -> :ok
+    case TimeoutRouter.dispatch(%TimeoutCommand{aggregate_uuid: UUID.uuid4(), sleep_in_ms: 2_000}) do
+      {:error, :aggregate_execution_failed} -> :ok
       {:error, :aggregate_execution_timeout} -> :ok
-      reply -> flunk("received an unexpected response: #{inspect reply}")
+      reply -> flunk("received an unexpected response: #{inspect(reply)}")
     end
   end
 
   test "should succeed when handler completes within configured timeout" do
-    :ok = TimeoutRouter.dispatch(%TimeoutCommand{aggregate_uuid: UUID.uuid4, sleep_in_ms: 100})
+    :ok = TimeoutRouter.dispatch(%TimeoutCommand{aggregate_uuid: UUID.uuid4(), sleep_in_ms: 100})
   end
 
   test "should succeed when timeout is overridden during dispatch" do
-    :ok = TimeoutRouter.dispatch(%TimeoutCommand{aggregate_uuid: UUID.uuid4, sleep_in_ms: 100}, 2_000)
+    :ok =
+      TimeoutRouter.dispatch(
+        %TimeoutCommand{aggregate_uuid: UUID.uuid4(), sleep_in_ms: 100},
+        2_000
+      )
   end
 end
diff --git a/test/event/event_handler_subscribe_to_stream_test.exs b/test/event/event_handler_subscribe_to_stream_test.exs
new file mode 100644
index 0000000..94699ce
--- /dev/null
+++ b/test/event/event_handler_subscribe_to_stream_test.exs
@@ -0,0 +1,53 @@
+defmodule Commanded.Event.EventHandlerSubscribeToStreamTest do
+  use Commanded.StorageCase
+
+  alias Commanded.EventStore
+
+  defmodule AnEvent do
+    defstruct [:stream_uuid, :reply_to]
+  end
+
+  defmodule SingleStreamEventHandler do
+    use Commanded.Event.Handler, name: __MODULE__, subscribe_to: "stream2"
+
+    def handle(%AnEvent{} = event, _metadata) do
+      %AnEvent{stream_uuid: stream_uuid, reply_to: reply_to} = event
+
+      pid = :erlang.list_to_pid(reply_to)
+
+      Process.send(pid, {:event, stream_uuid}, [])
+    end
+  end
+
+  describe "single stream event handler" do
+    setup do
+      {:ok, handler} = SingleStreamEventHandler.start_link()
+
+      [handler: handler]
+    end
+
+    test "should be only be notified of events appended to subscribed stream" do
+      append_events_to_stream("stream1", 3)
+      append_events_to_stream("stream2", 3)
+      append_events_to_stream("stream3", 3)
+
+      assert_receive {:event, "stream2"}
+      assert_receive {:event, "stream2"}
+      assert_receive {:event, "stream2"}
+      refute_receive {:event, _stream_uuid}
+    end
+  end
+
+  defp append_events_to_stream(stream_uuid, count) do
+    reply_to = self() |> :erlang.pid_to_list()
+
+    events =
+      1..count
+      |> Enum.map(fn _i ->
+        %AnEvent{reply_to: reply_to, stream_uuid: stream_uuid}
+      end)
+      |> Commanded.Event.Mapper.map_to_event_data([])
+
+    EventStore.append_to_stream(stream_uuid, :any_version, events)
+  end
+end
diff --git a/test/event/handle_event_test.exs b/test/event/handle_event_test.exs
index 82892d9..dd325bd 100644
--- a/test/event/handle_event_test.exs
+++ b/test/event/handle_event_test.exs
@@ -5,11 +5,11 @@ defmodule Commanded.Event.HandleEventTest do
   import Commanded.Assertions.EventAssertions
 
   alias Commanded.EventStore
-  alias Commanded.Event.{AppendingEventHandler,UninterestingEvent}
+  alias Commanded.Event.{AppendingEventHandler, UninterestingEvent}
   alias Commanded.Helpers.EventFactory
-  alias Commanded.Helpers.{ProcessHelper,Wait}
+  alias Commanded.Helpers.{ProcessHelper, Wait}
   alias Commanded.ExampleDomain.BankAccount.AccountBalanceHandler
-  alias Commanded.ExampleDomain.BankAccount.Events.{BankAccountOpened,MoneyDeposited}
+  alias Commanded.ExampleDomain.BankAccount.Events.{BankAccountOpened, MoneyDeposited}
 
   describe "balance handler" do
     setup do
@@ -19,9 +19,9 @@ defmodule Commanded.Event.HandleEventTest do
         assert AccountBalanceHandler.subscribed?()
       end)
 
-      on_exit fn ->
+      on_exit(fn ->
         ProcessHelper.shutdown(handler)
-      end
+      end)
 
       [handler: handler]
     end
@@ -31,12 +31,13 @@ defmodule Commanded.Event.HandleEventTest do
         %BankAccountOpened{account_number: "ACC123", initial_balance: 1_000},
         %MoneyDeposited{amount: 50, balance: 1_050}
       ]
+
       recorded_events = EventFactory.map_to_recorded_events(events)
 
       send(handler, {:events, recorded_events})
 
       Wait.until(fn ->
-        assert AccountBalanceHandler.current_balance == 1_050
+        assert AccountBalanceHandler.current_balance() == 1_050
       end)
     end
 
@@ -51,44 +52,53 @@ defmodule Commanded.Event.HandleEventTest do
         %MoneyDeposited{amount: 50, balance: 1_050},
         %UninterestingEvent{}
       ]
+
       recorded_events = EventFactory.map_to_recorded_events(events)
 
       send(handler, {:events, recorded_events})
 
       Wait.until(fn ->
-        assert AccountBalanceHandler.current_balance == 1_050
+        assert AccountBalanceHandler.current_balance() == 1_050
       end)
     end
   end
 
+  defp to_event_data(events) do
+    Commanded.Event.Mapper.map_to_event_data(events,
+      causation_id: UUID.uuid4(),
+      correlation_id: UUID.uuid4(),
+      metadata: %{}
+    )
+  end
+
   describe "appending handler" do
     setup do
-      on_exit fn ->
+      on_exit(fn ->
         ProcessHelper.shutdown(AppendingEventHandler)
-      end
+      end)
     end
 
     test "should ignore events created before the event handler's subscription when starting from `:current`" do
-      stream_uuid = UUID.uuid4
+      stream_uuid = UUID.uuid4()
       initial_events = [%BankAccountOpened{account_number: "ACC123", initial_balance: 1_000}]
       new_events = [%MoneyDeposited{amount: 50, balance: 1_050}]
 
-      {:ok, 1} = EventStore.append_to_stream(stream_uuid, 0, Commanded.Event.Mapper.map_to_event_data(initial_events, UUID.uuid4(), UUID.uuid4(), %{}))
+      :ok = EventStore.append_to_stream(stream_uuid, 0, to_event_data(initial_events))
 
-      wait_for_event BankAccountOpened
+      wait_for_event(BankAccountOpened)
 
       {:ok, handler} = AppendingEventHandler.start_link(start_from: :current)
 
       assert GenServer.call(handler, :last_seen_event) == nil
 
-      {:ok, 2} = EventStore.append_to_stream(stream_uuid, 1, Commanded.Event.Mapper.map_to_event_data(new_events, UUID.uuid4(), UUID.uuid4(), %{}))
+      :ok = EventStore.append_to_stream(stream_uuid, 1, to_event_data(new_events))
 
-      wait_for_event MoneyDeposited
+      wait_for_event(MoneyDeposited)
 
       Wait.until(fn ->
         assert AppendingEventHandler.received_events() == new_events
 
-        [ metadata ] = AppendingEventHandler.received_metadata()
+        [metadata] = AppendingEventHandler.received_metadata()
 
         assert Map.get(metadata, :event_number) == 2
         assert Map.get(metadata, :stream_id) == stream_uuid
@@ -97,20 +107,20 @@ defmodule Commanded.Event.HandleEventTest do
 
         assert GenServer.call(handler, :last_seen_event) == 2
       end)
-  	end
+    end
 
     test "should receive events created before the event handler's subscription when starting from `:origin`" do
-      stream_uuid = UUID.uuid4
+      stream_uuid = UUID.uuid4()
       initial_events = [%BankAccountOpened{account_number: "ACC123", initial_balance: 1_000}]
       new_events = [%MoneyDeposited{amount: 50, balance: 1_050}]
 
-      {:ok, 1} = EventStore.append_to_stream(stream_uuid, 0, Commanded.Event.Mapper.map_to_event_data(initial_events, UUID.uuid4(), UUID.uuid4(), %{}))
+      :ok = EventStore.append_to_stream(stream_uuid, 0, to_event_data(initial_events))
 
       {:ok, _handler} = AppendingEventHandler.start_link(start_from: :origin)
 
-      {:ok, 2} = EventStore.append_to_stream(stream_uuid, 1, Commanded.Event.Mapper.map_to_event_data(new_events, UUID.uuid4(), UUID.uuid4(), %{}))
+      :ok = EventStore.append_to_stream(stream_uuid, 1, to_event_data(new_events))
 
-      wait_for_event MoneyDeposited
+      wait_for_event(MoneyDeposited)
 
       Wait.until(fn ->
         assert AppendingEventHandler.received_events() == initial_events ++ new_events
@@ -125,15 +135,16 @@ defmodule Commanded.Event.HandleEventTest do
           assert %NaiveDateTime{} = Map.get(metadata, :created_at)
         end)
       end)
-  	end
+    end
 
-  	test "should ignore already seen events" do
+    test "should ignore already seen events" do
       {:ok, handler} = AppendingEventHandler.start_link()
 
       events = [
         %BankAccountOpened{account_number: "ACC123", initial_balance: 1_000},
         %MoneyDeposited{amount: 50, balance: 1_050}
       ]
+
       recorded_events = EventFactory.map_to_recorded_events(events)
 
       Wait.until(fn ->
@@ -150,7 +161,7 @@ defmodule Commanded.Event.HandleEventTest do
         assert AppendingEventHandler.received_events() == events
         assert pluck(AppendingEventHandler.received_metadata(), :stream_version) == [1, 2]
       end)
-  	end
+    end
   end
 
   describe "event handler name" do
@@ -181,19 +192,19 @@ defmodule Commanded.Event.HandleEventTest do
 
   test "should ensure an event handler name is provided" do
     assert_raise RuntimeError, "UnnamedEventHandler expects `:name` to be given", fn ->
-      Code.eval_string """
+      Code.eval_string("""
         defmodule UnnamedEventHandler do
           use Commanded.Event.Handler
         end
-      """
+      """)
     end
   end
 
   test "should allow using event handler module as name" do
-    Code.eval_string """
+    Code.eval_string("""
       defmodule EventHandler do
         use Commanded.Event.Handler, name: __MODULE__
       end
-    """
+    """)
   end
 end
diff --git a/test/event/handler_config_test.exs b/test/event/handler_config_test.exs
index 1e58ca1..e1b379f 100644
--- a/test/event/handler_config_test.exs
+++ b/test/event/handler_config_test.exs
@@ -9,7 +9,7 @@ defmodule Commanded.Event.HandlerConfigTest do
   test "should default to `:eventual` consistency and start from `:origin`" do
     {:ok, handler} = DefaultConfigHandler.start_link()
 
-    assert_config handler, [consistency: :eventual, start_from: :origin]
+    assert_config(handler, consistency: :eventual, start_from: :origin, subscribe_to: :all)
   end
 
   describe "global config change to default consistency" do
@@ -17,7 +17,7 @@ defmodule Commanded.Event.HandlerConfigTest do
       Mix.Config.persist(commanded: [default_consistency: :strong])
       {:ok, handler} = DefaultConfigHandler.start_link()
 
-      assert_config handler, [consistency: :strong, start_from: :origin]
+      assert_config(handler, consistency: :strong, start_from: :origin, subscribe_to: :all)
 
       Mix.Config.persist(commanded: [default_consistency: :eventual])
     end
@@ -27,48 +27,62 @@ defmodule Commanded.Event.HandlerConfigTest do
     use Commanded.Event.Handler,
       name: __MODULE__,
       consistency: :strong,
-      start_from: :current
+      start_from: :current,
+      subscribe_to: "stream1"
   end
 
   describe "config provided to `start_link/1`" do
     test "should use config defined in handler" do
       {:ok, handler} = ExampleHandler.start_link()
 
-      assert_config handler, [consistency: :strong, start_from: :current]
+      assert_config(handler, consistency: :strong, start_from: :current, subscribe_to: "stream1")
     end
 
     test "should use overriden config when provided" do
-      {:ok, handler} = ExampleHandler.start_link(consistency: :eventual, start_from: :origin)
-
-      assert_config handler, [consistency: :eventual, start_from: :origin]
+      {:ok, handler} =
+        ExampleHandler.start_link(
+          consistency: :eventual,
+          start_from: :origin,
+          subscribe_to: "stream2"
+        )
+
+      assert_config(handler, consistency: :eventual, start_from: :origin, subscribe_to: "stream2")
     end
 
     test "should use default config when not provided" do
       {:ok, handler} = ExampleHandler.start_link(start_from: :origin)
 
-      assert_config handler, [consistency: :strong, start_from: :origin]
+      assert_config(handler, consistency: :strong, start_from: :origin, subscribe_to: "stream1")
     end
   end
 
   describe "config provided to `child_spec`" do
     test "should use config defined in handler" do
-      {:ok, sup} = Supervisor.start_link([
-        ExampleHandler,
-      ], strategy: :one_for_one)
+      {:ok, sup} =
+        Supervisor.start_link(
+          [
+            ExampleHandler
+          ],
+          strategy: :one_for_one
+        )
 
       [{_, handler, _, _}] = Supervisor.which_children(sup)
 
-      assert_config handler, [consistency: :strong, start_from: :current]
+      assert_config(handler, consistency: :strong, start_from: :current, subscribe_to: "stream1")
     end
 
     test "should use overriden config when provided" do
-      {:ok, sup} = Supervisor.start_link([
-        {ExampleHandler, [consistency: :eventual, start_from: :origin]},
-      ], strategy: :one_for_one)
+      {:ok, sup} =
+        Supervisor.start_link(
+          [
+            {ExampleHandler, [consistency: :eventual, start_from: :origin]}
+          ],
+          strategy: :one_for_one
+        )
 
       [{_, handler, _, _}] = Supervisor.which_children(sup)
 
-      assert_config handler, [consistency: :eventual, start_from: :origin]
+      assert_config(handler, consistency: :eventual, start_from: :origin, subscribe_to: "stream1")
     end
   end
 
@@ -79,13 +93,15 @@ defmodule Commanded.Event.HandlerConfigTest do
   end
 
   test "should validate provided config" do
-    assert_raise RuntimeError, "Commanded.Event.HandlerConfigTest.InvalidConfiguredHandler specifies invalid options: [:invalid_config_option]", fn ->
-      Code.eval_string """
-        alias Commanded.Event.HandlerConfigTest.InvalidConfiguredHandler
-
-        {:ok, _pid} = InvalidConfiguredHandler.start_link()
-      """
-    end
+    assert_raise RuntimeError,
+                 "Commanded.Event.HandlerConfigTest.InvalidConfiguredHandler specifies invalid options: [:invalid_config_option]",
+                 fn ->
+                   Code.eval_string("""
+                     alias Commanded.Event.HandlerConfigTest.InvalidConfiguredHandler
+
+                     {:ok, _pid} = InvalidConfiguredHandler.start_link()
+                   """)
+                 end
   end
 
   defp assert_config(handler, expected_config) do
diff --git a/test/event/handler_init_test.exs b/test/event/handler_init_test.exs
index 84b026f..cea7f6a 100644
--- a/test/event/handler_init_test.exs
+++ b/test/event/handler_init_test.exs
@@ -1,68 +1,42 @@
 defmodule Commanded.Event.HandlerInitTest do
-  use Commanded.StorageCase
+  use Commanded.MockEventStoreCase
 
   import Mox
 
   alias Commanded.Event.InitHandler
-  alias Commanded.Helpers.ProcessHelper
   alias Commanded.EventStore.Adapters.Mock, as: MockEventStore
 
   setup do
     reply_to = self()
-    {:ok, agent} = Agent.start_link(fn -> reply_to end, name: {:global, InitHandler})
 
-    on_exit(fn ->
-      ProcessHelper.shutdown(agent)
+    expect(MockEventStore, :subscribe_to, fn :all, _handler_name, handler, _subscribe_from ->
+      {:ok, handler}
     end)
-  end
-
-  describe "event handler `init/0` callback" do
-    setup do
-      {:ok, handler} = InitHandler.start_link()
 
-      on_exit(fn ->
-        ProcessHelper.shutdown(handler)
-      end)
+    {:ok, agent} = Agent.start_link(fn -> reply_to end, name: InitHandler)
+    {:ok, handler} = InitHandler.start_link()
 
-      [handler: handler]
-    end
+    [agent: agent, handler: handler]
+  end
 
+  describe "event handler `init/0` callback" do
     test "should be called", %{handler: handler} do
+      send_subscribed(handler)
+
       assert_receive {:init, ^handler}
     end
   end
 
   describe "event handler `init/0` callback after subscribed" do
-    setup do
-      set_mox_global()
-
-      # use mock event store adapter
-      default_event_store_adapter = Application.get_env(:commanded, :event_store_adapter)
-      :ok = Application.put_env(:commanded, :event_store_adapter, MockEventStore)
-
-      expect(MockEventStore, :subscribe_to_all_streams, fn _handler_name, handler, _subscribe_from ->
-        {:ok, handler}
-      end)
-
-      {:ok, handler} = InitHandler.start_link()
-
-      on_exit(fn ->
-        Application.put_env(:commanded, :event_store_adapter, default_event_store_adapter)
-        ProcessHelper.shutdown(handler)
-      end)
-
-      [handler: handler]
-    end
-
     test "should be called after subscription subscribed", %{handler: handler} do
       refute_receive {:init, ^handler}
 
-      simulate_subscribed(handler)
+      send_subscribed(handler)
 
       assert_receive {:init, ^handler}
     end
 
-    defp simulate_subscribed(handler) do
+    defp send_subscribed(handler) do
       send(handler, {:subscribed, handler})
     end
   end
diff --git a/test/event/support/init/init_handler.ex b/test/event/support/init/init_handler.ex
index e378689..0f66483 100644
--- a/test/event/support/init/init_handler.ex
+++ b/test/event/support/init/init_handler.ex
@@ -3,11 +3,11 @@ defmodule Commanded.Event.InitHandler do
 
   def init do
     send(reply_to(), {:init, self()})
-    
+
     :ok
   end
 
   defp reply_to do
-    Agent.get({:global, __MODULE__}, fn reply_to -> reply_to end)
+    Agent.get(__MODULE__, fn reply_to -> reply_to end)
   end
 end
diff --git a/test/event_store/adapters/in_memory_test.exs b/test/event_store/adapters/in_memory_test.exs
index 872d505..e416244 100644
--- a/test/event_store/adapters/in_memory_test.exs
+++ b/test/event_store/adapters/in_memory_test.exs
@@ -17,8 +17,9 @@ defmodule Commanded.EventStore.Adapters.InMemoryTest do
     test "wipes all data from memory" do
       pid = Process.whereis(InMemory)
       initial = :sys.get_state(pid)
+      events = [build_event(1)]
 
-      {:ok, 1} = InMemory.append_to_stream("stream", 0, [build_event(1)])
+      :ok = InMemory.append_to_stream("stream", 0, events)
       after_event = :sys.get_state(pid)
 
       InMemory.reset!()
diff --git a/test/event_store/support/append_events_test_case.ex b/test/event_store/support/append_events_test_case.ex
index 6edc313..37a6fc3 100644
--- a/test/event_store/support/append_events_test_case.ex
+++ b/test/event_store/support/append_events_test_case.ex
@@ -13,9 +13,9 @@ defmodule Commanded.EventStore.AppendEventsTestCase do
 
     describe "append events to a stream" do
       test "should append events" do
-        assert {:ok, 2} == EventStore.append_to_stream("stream", 0, build_events(2))
-        assert {:ok, 4} == EventStore.append_to_stream("stream", 2, build_events(2))
-        assert {:ok, 5} == EventStore.append_to_stream("stream", 4, build_events(1))
+        assert :ok == EventStore.append_to_stream("stream", 0, build_events(2))
+        assert :ok == EventStore.append_to_stream("stream", 2, build_events(2))
+        assert :ok == EventStore.append_to_stream("stream", 4, build_events(1))
       end
 
       test "should fail to append to a stream because of wrong expected version when no previous events" do
@@ -26,11 +26,17 @@ defmodule Commanded.EventStore.AppendEventsTestCase do
       end
 
       test "should fail to append to a stream because of wrong expected version" do
-        assert {:ok, 2} == EventStore.append_to_stream("stream", 0, build_events(2))
+        assert :ok == EventStore.append_to_stream("stream", 0, build_events(2))
 
         assert {:error, :wrong_expected_version} ==
                  EventStore.append_to_stream("stream", 1, build_events(1))
       end
+
+      test "should append events to any version" do
+        assert :ok == EventStore.append_to_stream("stream", :any_version, build_events(2))
+        assert :ok == EventStore.append_to_stream("stream", :any_version, build_events(2))
+        assert :ok == EventStore.append_to_stream("stream", :any_version, build_events(1))
+      end
     end
 
     describe "stream events from an unknown stream" do
@@ -45,7 +51,7 @@ defmodule Commanded.EventStore.AppendEventsTestCase do
         causation_id = UUID.uuid4()
         events = build_events(4, correlation_id, causation_id)
 
-        assert {:ok, 4} == EventStore.append_to_stream("stream", 0, events)
+        assert :ok == EventStore.append_to_stream("stream", 0, events)
 
         read_events = EventStore.stream_forward("stream") |> Enum.to_list()
         assert length(read_events) == 4
@@ -69,8 +75,8 @@ defmodule Commanded.EventStore.AppendEventsTestCase do
         events1 = build_events(2)
         events2 = build_events(4)
 
-        assert {:ok, 2} == EventStore.append_to_stream("stream", 0, events1)
-        assert {:ok, 4} == EventStore.append_to_stream("secondstream", 0, events2)
+        assert :ok == EventStore.append_to_stream("stream", 0, events1)
+        assert :ok == EventStore.append_to_stream("secondstream", 0, events2)
 
         read_events = EventStore.stream_forward("stream", 0) |> Enum.to_list()
         assert 2 == length(read_events)
@@ -84,7 +90,7 @@ defmodule Commanded.EventStore.AppendEventsTestCase do
       test "should read events in batches" do
         events = build_events(10)
 
-        assert {:ok, 10} == EventStore.append_to_stream("stream", 0, events)
+        assert :ok == EventStore.append_to_stream("stream", 0, events)
 
         read_events = EventStore.stream_forward("stream", 0, 2) |> Enum.to_list()
         assert length(read_events) == 10
diff --git a/test/event_store/support/subscriber.ex b/test/event_store/support/subscriber.ex
index d929b61..85e1ed7 100644
--- a/test/event_store/support/subscriber.ex
+++ b/test/event_store/support/subscriber.ex
@@ -3,7 +3,7 @@ defmodule Commanded.EventStore.Subscriber do
 
   alias Commanded.EventStore
   alias Commanded.EventStore.Subscriber
-  
+
   defmodule State do
     defstruct received_events: [],
               subscribed?: false,
@@ -15,7 +15,7 @@ defmodule Commanded.EventStore.Subscriber do
   def start_link, do: GenServer.start_link(__MODULE__, %State{})
 
   def init(%State{} = state) do
-    {:ok, subscription} = EventStore.subscribe_to_all_streams("subscriber", self(), :origin)
+    {:ok, subscription} = EventStore.subscribe_to(:all, "subscriber", self(), :origin)
 
     {:ok, %State{state | subscription: subscription}}
   end
diff --git a/test/event_store/support/subscription_test_case.ex b/test/event_store/support/subscription_test_case.ex
index 88315ad..2aecc94 100644
--- a/test/event_store/support/subscription_test_case.ex
+++ b/test/event_store/support/subscription_test_case.ex
@@ -16,19 +16,19 @@ defmodule Commanded.EventStore.SubscriptionTestCase do
 
         assert :ok = EventStore.subscribe(stream_uuid)
 
-        {:ok, 1} = EventStore.append_to_stream(stream_uuid, 0, build_events(1))
+        :ok = EventStore.append_to_stream(stream_uuid, 0, build_events(1))
 
         received_events = assert_receive_events(1, from: 1)
         assert Enum.map(received_events, & &1.stream_id) == [stream_uuid]
         assert Enum.map(received_events, & &1.stream_version) == [1]
 
-        {:ok, 3} = EventStore.append_to_stream(stream_uuid, 1, build_events(2))
+        :ok = EventStore.append_to_stream(stream_uuid, 1, build_events(2))
 
         received_events = assert_receive_events(2, from: 2)
         assert Enum.map(received_events, & &1.stream_id) == [stream_uuid, stream_uuid]
         assert Enum.map(received_events, & &1.stream_version) == [2, 3]
 
-        {:ok, 6} = EventStore.append_to_stream(stream_uuid, 3, build_events(3))
+        :ok = EventStore.append_to_stream(stream_uuid, 3, build_events(3))
 
         received_events = assert_receive_events(3, from: 4)
 
@@ -49,106 +49,167 @@ defmodule Commanded.EventStore.SubscriptionTestCase do
 
         assert :ok = EventStore.subscribe(stream_uuid)
 
-        {:ok, 1} = EventStore.append_to_stream(another_stream_uuid, 0, build_events(1))
-        {:ok, 3} = EventStore.append_to_stream(another_stream_uuid, 1, build_events(2))
+        :ok = EventStore.append_to_stream(another_stream_uuid, 0, build_events(1))
+        :ok = EventStore.append_to_stream(another_stream_uuid, 1, build_events(2))
 
         refute_receive {:events, _received_events}
       end
     end
 
+    describe "subscribe to single stream" do
+      test "should receive `:subscribed` message once subscribed" do
+        {:ok, subscription} = EventStore.subscribe_to("stream1", "subscriber", self(), :origin)
+
+        assert_receive {:subscribed, ^subscription}
+      end
+
+      test "should receive events appended to stream" do
+        {:ok, subscription} = EventStore.subscribe_to("stream1", "subscriber", self(), :origin)
+
+        assert_receive {:subscribed, ^subscription}
+
+        :ok = EventStore.append_to_stream("stream1", 0, build_events(1))
+        :ok = EventStore.append_to_stream("stream1", 1, build_events(2))
+        :ok = EventStore.append_to_stream("stream1", 3, build_events(3))
+
+        assert_receive_events(subscription, 1, from: 1)
+        assert_receive_events(subscription, 2, from: 2)
+        assert_receive_events(subscription, 3, from: 4)
+
+        refute_receive {:events, _received_events}
+      end
+
+      test "should not receive events appended to another stream" do
+        {:ok, subscription} = EventStore.subscribe_to("stream1", "subscriber", self(), :origin)
+
+        :ok = EventStore.append_to_stream("stream1", 0, build_events(1))
+        :ok = EventStore.append_to_stream("stream2", 0, build_events(2))
+        :ok = EventStore.append_to_stream("stream3", 0, build_events(3))
+
+        assert_receive_events(subscription, 1, from: 1)
+        refute_receive {:events, _received_events}
+      end
+
+      test "should skip existing events when subscribing from current position" do
+        :ok = EventStore.append_to_stream("stream1", 0, build_events(1))
+        :ok = EventStore.append_to_stream("stream1", 1, build_events(2))
+
+        wait_for_event_store()
+
+        {:ok, subscription} = EventStore.subscribe_to("stream1", "subscriber", self(), :current)
+
+        assert_receive {:subscribed, ^subscription}
+        refute_receive {:events, _events}
+
+        :ok = EventStore.append_to_stream("stream1", 3, build_events(3))
+        :ok = EventStore.append_to_stream("stream2", 0, build_events(3))
+        :ok = EventStore.append_to_stream("stream3", 0, build_events(3))
+
+        assert_receive_events(subscription, 3, from: 4)
+        refute_receive {:events, _events}
+      end
+
+      test "should prevent duplicate subscriptions" do
+        {:ok, _subscription} = EventStore.subscribe_to("stream1", "subscriber", self(), :origin)
+
+        assert {:error, :subscription_already_exists} ==
+                 EventStore.subscribe_to("stream1", "subscriber", self(), :origin)
+      end
+    end
+
     describe "subscribe to all streams" do
       test "should receive `:subscribed` message once subscribed" do
-        {:ok, subscription} = EventStore.subscribe_to_all_streams("subscriber", self(), :origin)
+        {:ok, subscription} = EventStore.subscribe_to(:all, "subscriber", self(), :origin)
 
         assert_receive {:subscribed, ^subscription}
       end
 
       test "should receive events appended to any stream" do
-        {:ok, subscription} = EventStore.subscribe_to_all_streams("subscriber", self(), :origin)
+        {:ok, subscription} = EventStore.subscribe_to(:all, "subscriber", self(), :origin)
 
         assert_receive {:subscribed, ^subscription}
 
-        {:ok, 1} = EventStore.append_to_stream("stream1", 0, build_events(1))
-        {:ok, 2} = EventStore.append_to_stream("stream2", 0, build_events(2))
-        {:ok, 3} = EventStore.append_to_stream("stream3", 0, build_events(3))
+        :ok = EventStore.append_to_stream("stream1", 0, build_events(1))
+        :ok = EventStore.append_to_stream("stream2", 0, build_events(2))
+        :ok = EventStore.append_to_stream("stream3", 0, build_events(3))
 
         assert_receive_events(subscription, 1, from: 1)
         assert_receive_events(subscription, 2, from: 2)
         assert_receive_events(subscription, 3, from: 4)
 
-        refute_receive({:events, _events})
+        refute_receive {:events, _received_events}
       end
 
       test "should skip existing events when subscribing from current position" do
-        {:ok, 1} = EventStore.append_to_stream("stream1", 0, build_events(1))
-        {:ok, 2} = EventStore.append_to_stream("stream2", 0, build_events(2))
+        :ok = EventStore.append_to_stream("stream1", 0, build_events(1))
+        :ok = EventStore.append_to_stream("stream2", 0, build_events(2))
 
         wait_for_event_store()
 
-        {:ok, subscription} = EventStore.subscribe_to_all_streams("subscriber", self(), :current)
+        {:ok, subscription} = EventStore.subscribe_to(:all, "subscriber", self(), :current)
 
         assert_receive {:subscribed, ^subscription}
-        refute_receive({:events, _events})
+        refute_receive {:events, _received_events}
 
-        {:ok, 3} = EventStore.append_to_stream("stream3", 0, build_events(3))
+        :ok = EventStore.append_to_stream("stream3", 0, build_events(3))
 
         assert_receive_events(subscription, 3, from: 4)
 
-        refute_receive({:events, _events})
+        refute_receive {:events, _received_events}
       end
 
       test "should prevent duplicate subscriptions" do
-        {:ok, _subscription} = EventStore.subscribe_to_all_streams("subscriber", self(), :origin)
+        {:ok, _subscription} = EventStore.subscribe_to(:all, "subscriber", self(), :origin)
 
         assert {:error, :subscription_already_exists} ==
-                 EventStore.subscribe_to_all_streams("subscriber", self(), :origin)
+                 EventStore.subscribe_to(:all, "subscriber", self(), :origin)
       end
     end
 
     describe "catch-up subscription" do
       test "should receive any existing events" do
-        {:ok, 1} = EventStore.append_to_stream("stream1", 0, build_events(1))
-        {:ok, 2} = EventStore.append_to_stream("stream2", 0, build_events(2))
+        :ok = EventStore.append_to_stream("stream1", 0, build_events(1))
+        :ok = EventStore.append_to_stream("stream2", 0, build_events(2))
 
         wait_for_event_store()
 
-        {:ok, subscription} = EventStore.subscribe_to_all_streams("subscriber", self(), :origin)
+        {:ok, subscription} = EventStore.subscribe_to(:all, "subscriber", self(), :origin)
 
         assert_receive {:subscribed, ^subscription}
 
         assert_receive_events(subscription, 1, from: 1)
         assert_receive_events(subscription, 2, from: 2)
 
-        {:ok, 3} = EventStore.append_to_stream("stream3", 0, build_events(3))
+        :ok = EventStore.append_to_stream("stream3", 0, build_events(3))
 
         assert_receive_events(subscription, 3, from: 4)
-        refute_receive({:events, _events})
+        refute_receive {:events, _received_events}
       end
     end
 
     describe "unsubscribe from all streams" do
       test "should not receive further events appended to any stream" do
-        {:ok, subscription} = EventStore.subscribe_to_all_streams("subscriber", self(), :origin)
+        {:ok, subscription} = EventStore.subscribe_to(:all, "subscriber", self(), :origin)
 
         assert_receive {:subscribed, ^subscription}
 
-        {:ok, 1} = EventStore.append_to_stream("stream1", 0, build_events(1))
+        :ok = EventStore.append_to_stream("stream1", 0, build_events(1))
 
         assert_receive_events(subscription, 1, from: 1)
 
-        :ok = EventStore.unsubscribe_from_all_streams("subscriber")
+        :ok = EventStore.unsubscribe("subscriber")
 
-        {:ok, 2} = EventStore.append_to_stream("stream2", 0, build_events(2))
-        {:ok, 3} = EventStore.append_to_stream("stream3", 0, build_events(3))
+        :ok = EventStore.append_to_stream("stream2", 0, build_events(2))
+        :ok = EventStore.append_to_stream("stream3", 0, build_events(3))
 
-        refute_receive({:events, _events})
+        refute_receive {:events, _received_events}
       end
     end
 
     describe "resume subscription" do
       test "should remember last seen event number when subscription resumes" do
-        {:ok, 1} = EventStore.append_to_stream("stream1", 0, build_events(1))
-        {:ok, 2} = EventStore.append_to_stream("stream2", 0, build_events(2))
+        :ok = EventStore.append_to_stream("stream1", 0, build_events(1))
+        :ok = EventStore.append_to_stream("stream2", 0, build_events(2))
 
         {:ok, subscriber} = Subscriber.start_link()
 
@@ -174,7 +235,7 @@ defmodule Commanded.EventStore.SubscriptionTestCase do
         received_events = Subscriber.received_events(subscriber)
         assert length(received_events) == 0
 
-        {:ok, 1} = EventStore.append_to_stream("stream3", 0, build_events(1))
+        :ok = EventStore.append_to_stream("stream3", 0, build_events(1))
 
         wait_until(fn ->
           received_events = Subscriber.received_events(subscriber)
diff --git a/test/helpers/event_factory.ex b/test/helpers/event_factory.ex
index 332632e..0cdf0c7 100644
--- a/test/helpers/event_factory.ex
+++ b/test/helpers/event_factory.ex
@@ -6,9 +6,10 @@ defmodule Commanded.Helpers.EventFactory do
     stream_id = UUID.uuid4()
     causation_id = UUID.uuid4()
     correlation_id = UUID.uuid4()
+    fields = [causation_id: causation_id, correlation_id: correlation_id, metadata: %{}]
 
     events
-    |> Commanded.Event.Mapper.map_to_event_data(causation_id, correlation_id, %{})
+    |> Commanded.Event.Mapper.map_to_event_data(fields)
     |> Enum.with_index(1)
     |> Enum.map(fn {event, index} ->
       %RecordedEvent{
@@ -21,7 +22,7 @@ defmodule Commanded.Helpers.EventFactory do
         event_type: event.event_type,
         data: event.data,
         metadata: event.metadata,
-        created_at: now(),
+        created_at: now()
       }
     end)
   end
diff --git a/test/helpers/storage_case.ex b/test/helpers/storage_case.ex
index ac60724..6c963da 100644
--- a/test/helpers/storage_case.ex
+++ b/test/helpers/storage_case.ex
@@ -6,7 +6,6 @@ defmodule Commanded.StorageCase do
   require Logger
 
   alias Commanded.EventStore.Adapters.InMemory
-  alias Commanded.Serialization.JsonSerializer
 
   setup do
     on_exit(fn ->
diff --git a/test/process_managers/process_manager_timeout_test.exs b/test/process_managers/process_manager_timeout_test.exs
index b06f717..b8dae6c 100644
--- a/test/process_managers/process_manager_timeout_test.exs
+++ b/test/process_managers/process_manager_timeout_test.exs
@@ -1,6 +1,7 @@
 defmodule Commanded.ProcessManagers.ProcessManagerTimeoutTest do
   use Commanded.StorageCase
 
+  alias Commanded.Helpers.Wait
   alias Commanded.ProcessManagers.ProcessRouter
   alias Commanded.ProcessManagers.{ExampleRouter, ExampleProcessManager}
   alias Commanded.ProcessManagers.ExampleAggregate.Commands.{Pause, Start}
@@ -13,7 +14,7 @@ defmodule Commanded.ProcessManagers.ProcessManagerTimeoutTest do
 
     :ok = ExampleRouter.dispatch(%Start{aggregate_uuid: aggregate_uuid})
 
-    process_instance = ProcessRouter.process_instance(process_router, aggregate_uuid)
+    process_instance = wait_for_process_instance(process_router, aggregate_uuid)
     instance_ref = Process.monitor(process_instance)
 
     :ok = ExampleRouter.dispatch(%Pause{aggregate_uuid: aggregate_uuid})
@@ -26,14 +27,14 @@ defmodule Commanded.ProcessManagers.ProcessManagerTimeoutTest do
   test "should timeout and shutdown process manager when `event_timeout` configured" do
     aggregate_uuid = UUID.uuid4()
 
-    {:ok, process_router} = ExampleProcessManager.start_link(event_timeout: 10)
-    router_ref = Process.monitor(process_router)
+    {:ok, process_router} = ExampleProcessManager.start_link(event_timeout: 100)
 
+    router_ref = Process.monitor(process_router)
     Process.unlink(process_router)
 
     :ok = ExampleRouter.dispatch(%Start{aggregate_uuid: aggregate_uuid})
 
-    process_instance = ProcessRouter.process_instance(process_router, aggregate_uuid)
+    process_instance = wait_for_process_instance(process_router, aggregate_uuid)
     instance_ref = Process.monitor(process_instance)
 
     :ok = ExampleRouter.dispatch(%Pause{aggregate_uuid: aggregate_uuid})
@@ -42,4 +43,14 @@ defmodule Commanded.ProcessManagers.ProcessManagerTimeoutTest do
     assert_receive {:DOWN, ^router_ref, _, _, :event_timeout}
     assert_receive {:DOWN, ^instance_ref, _, _, :shutdown}
   end
+
+  defp wait_for_process_instance(process_router, aggregate_uuid) do
+    Wait.until(fn ->
+      process_instance = ProcessRouter.process_instance(process_router, aggregate_uuid)
+
+      assert is_pid(process_instance)
+
+      process_instance
+    end)
+  end
 end
diff --git a/test/pubsub/local_pubsub_test.exs b/test/pubsub/local_pubsub_test.exs
index 9f3d88b..43d777f 100644
--- a/test/pubsub/local_pubsub_test.exs
+++ b/test/pubsub/local_pubsub_test.exs
@@ -6,12 +6,20 @@ defmodule Commanded.PubSub.LocalPubSubTest do
   setup do
     Application.put_env(:commanded, :pubsub, :local)
 
-    if Process.whereis(Commanded.PubSub.LocalPubSub) == nil do
-      {:ok, _pid} = Supervisor.start_link(LocalPubSub.child_spec(), strategy: :one_for_one)
-    end
+    pubsub_pid =
+      case Process.whereis(LocalPubSub) do
+        nil ->
+          {:ok, pid} = Supervisor.start_link(LocalPubSub.child_spec(), strategy: :one_for_one)
+          pid
+
+        pid ->
+          pid
+      end
 
     on_exit(fn ->
       Application.delete_env(:commanded, :pubsub)
     end)
+
+    [pubsub_pid: pubsub_pid]
   end
 end
diff --git a/test/pubsub/phoenix_pubsub_test.exs b/test/pubsub/phoenix_pubsub_test.exs
index 8ca92eb..b105370 100644
--- a/test/pubsub/phoenix_pubsub_test.exs
+++ b/test/pubsub/phoenix_pubsub_test.exs
@@ -13,10 +13,20 @@ defmodule Commanded.PubSub.PhoenixPubSubTest do
       ]
     )
 
-    {:ok, _pid} = Supervisor.start_link(PhoenixPubSub.child_spec(), strategy: :one_for_one)
+    pubsub_pid =
+      case Process.whereis(PhoenixPubSub) do
+        nil ->
+          {:ok, pid} = Supervisor.start_link(PhoenixPubSub.child_spec(), strategy: :one_for_one)
+          pid
+
+        pid ->
+          pid
+      end
 
     on_exit(fn ->
       Application.delete_env(:commanded, :pubsub)
     end)
+
+    [pubsub_pid: pubsub_pid]
   end
 end
diff --git a/test/registration/local_registry_test.exs b/test/registration/local_registry_test.exs
new file mode 100644
index 0000000..7de4593
--- /dev/null
+++ b/test/registration/local_registry_test.exs
@@ -0,0 +1,6 @@
+defmodule Commanded.LocalRegistryTest do
+  alias Commanded.RegistrationTestCase
+  alias Commanded.Registration.LocalRegistry
+
+  use RegistrationTestCase, registry: LocalRegistry
+end
diff --git a/test/registration/registration_test.exs b/test/registration/registration_test.exs
deleted file mode 100644
index 4742e80..0000000
--- a/test/registration/registration_test.exs
+++ /dev/null
@@ -1,54 +0,0 @@
-defmodule Commanded.RegistrationTest do
-  use Commanded.StorageCase
-
-  alias Commanded.Registration
-  alias Commanded.Registration.{RegisteredServer, RegisteredSupervisor}
-
-  setup do
-    {:ok, _pid} = RegisteredSupervisor.start_link()
-
-    :ok
-  end
-
-  describe "`start_child/3`" do
-    test "should return child process PID on success" do
-      assert {:ok, _pid} = RegisteredSupervisor.start_child("child")
-    end
-
-    test "should return existing child process when already started" do
-      assert {:ok, pid} = RegisteredSupervisor.start_child("child")
-      assert {:ok, ^pid} = RegisteredSupervisor.start_child("child")
-    end
-  end
-
-  describe "`start_link/3`" do
-    test "should return process PID on success" do
-      assert {:ok, _pid} = start_link("registered")
-    end
-
-    test "should return existing process when already started" do
-      assert {:ok, pid} = start_link("registered")
-      assert {:ok, ^pid} = start_link("registered")
-    end
-  end
-
-  describe "`whereis_name/1`" do
-    test "should return `:undefined` when not registered" do
-      assert Registration.whereis_name("notregistered") == :undefined
-    end
-
-    test "should return PID when child registered" do
-      assert {:ok, pid} = RegisteredSupervisor.start_child("child")
-      assert Registration.whereis_name("child") == pid
-    end
-
-    test "should return PID when process registered" do
-      assert {:ok, pid} = start_link("registered")
-      assert Registration.whereis_name("registered") == pid
-    end
-  end
-
-  defp start_link(name) do
-    Registration.start_link(name, RegisteredServer, [name])
-  end
-end
diff --git a/test/registration/support/registration_test_case.ex b/test/registration/support/registration_test_case.ex
new file mode 100644
index 0000000..93029e5
--- /dev/null
+++ b/test/registration/support/registration_test_case.ex
@@ -0,0 +1,62 @@
+defmodule Commanded.RegistrationTestCase do
+  import Commanded.SharedTestCase
+
+  define_tests do
+    alias Commanded.Registration
+    alias Commanded.Registration.{RegisteredServer, RegisteredSupervisor}
+
+    setup %{registry: registry} do
+      Application.put_env(:commanded, :registry, registry)
+
+      {:ok, supervisor} = RegisteredSupervisor.start_link()
+
+      on_exit(fn ->
+        Application.delete_env(:commanded, :registry)
+      end)
+
+      [supervisor: supervisor]
+    end
+
+    describe "`start_child/3`" do
+      test "should return child process PID on success" do
+        assert {:ok, _pid} = RegisteredSupervisor.start_child("child")
+      end
+
+      test "should return existing child process when already started" do
+        assert {:ok, pid} = RegisteredSupervisor.start_child("child")
+        assert {:ok, ^pid} = RegisteredSupervisor.start_child("child")
+      end
+    end
+
+    describe "`start_link/3`" do
+      test "should return process PID on success" do
+        assert {:ok, _pid} = start_link("registered")
+      end
+
+      test "should return existing process when already started" do
+        assert {:ok, pid} = start_link("registered")
+        assert {:ok, ^pid} = start_link("registered")
+      end
+    end
+
+    describe "`whereis_name/1`" do
+      test "should return `:undefined` when not registered" do
+        assert Registration.whereis_name("notregistered") == :undefined
+      end
+
+      test "should return `PID` when child registered" do
+        assert {:ok, pid} = RegisteredSupervisor.start_child("child")
+        assert Registration.whereis_name("child") == pid
+      end
+
+      test "should return `PID` when process registered" do
+        assert {:ok, pid} = start_link("registered")
+        assert Registration.whereis_name("registered") == pid
+      end
+    end
+
+    defp start_link(name) do
+      Registration.start_link(name, RegisteredServer, [name])
+    end
+  end
+end