Repository: commanded/commanded
Pull request #245 (2019-01-22 16:02:23)

Add `delete_subscription/2` to `Commanded.EventStore` behaviour
Unsubscribing an event store subscription should not delete the subscription. This pull request adds `Commanded.EventStore.delete_subscription/2` to allow an existing subscription to be completely removed.

### Example usage

Create a subscription to all events: 

```elixir
{:ok, subscription} = Commanded.EventStore.subscribe_to(:all, "Example", self())
```

Unsubscribe the subscription to stop receiving events, but still remember the subscription's last acknowledged event pointer:

```elixir
:ok = Commanded.EventStore.unsubscribe(subscription)
```

Finally, delete the subscription:

```elixir
:ok = Commanded.EventStore.delete_subscription(:all, "Example")
```


Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index e510a60..9605f06 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -16,6 +16,7 @@
 - Allow `:infinity` timeout on command dispatch ([#227](https://github.com/commanded/commanded/pull/227))
 - Strict process manager routing ([#243](https://github.com/commanded/commanded/pull/243)).
 - Allow `Commanded.Aggregate.Multi` to be nested ([#244](https://github.com/commanded/commanded/pull/244)).
+- Add `delete_subscription/2` to `Commanded.EventStore` behaviour ([#245](https://github.com/commanded/commanded/pull/245)).
 
 ### Bug fixes
 
diff --git a/lib/commanded/assertions/event_assertions.ex b/lib/commanded/assertions/event_assertions.ex
index abd5019..aa2b692 100644
--- a/lib/commanded/assertions/event_assertions.ex
+++ b/lib/commanded/assertions/event_assertions.ex
@@ -29,11 +29,14 @@ defmodule Commanded.Assertions.EventAssertions do
   end
 
   @doc """
-  Wait for an event of the given event type, matching the predicate, to be published.
+  Wait for an event of the given event type, matching the predicate, to be
+  published.
 
   ## Examples
 
-      wait_for_event BankAccountOpened, fn opened -> opened.account_number == "ACC123" end
+      wait_for_event(BankAccountOpened, fn opened ->
+        opened.account_number == "ACC123"
+      end)
 
   """
   def wait_for_event(event_type, predicate_fn) when is_function(predicate_fn) do
@@ -43,17 +46,18 @@ defmodule Commanded.Assertions.EventAssertions do
   end
 
   @doc """
-  Assert that events matching their respective predicates have a matching correlation id.
+  Assert that events matching their respective predicates have a matching
+  correlation id.
+
   Useful when there is a chain of events that is connected through event handlers.
 
   ## Examples
 
-      id_one = 1
-      id_two = 2
       assert_correlated(
-        BankAccountOpened, fn opened -> opened.id == id_one end,
-        InitialAmountDeposited, fn deposited -> deposited.id == id_two end
+        BankAccountOpened, fn opened -> opened.id == 1 end,
+        InitialAmountDeposited, fn deposited -> deposited.id == 2 end
       )
+
   """
   def assert_correlated(event_type_a, predicate_a, event_type_b, predicate_b) do
     assert_receive_event(event_type_a, predicate_a, fn _event_a, metadata_a ->
@@ -64,13 +68,14 @@ defmodule Commanded.Assertions.EventAssertions do
   end
 
   @doc """
-  Assert that an event of the given event type is published. Verify that event using the assertion function.
+  Assert that an event of the given event type is published.
+  Verify that event using the assertion function.
 
   ## Examples
 
-      assert_receive_event BankAccountOpened, fn opened ->
+      assert_receive_event(BankAccountOpened, fn opened ->
         assert opened.account_number == "ACC123"
-      end
+      end)
 
   """
   def assert_receive_event(event_type, assertion_fn) do
@@ -78,16 +83,16 @@ defmodule Commanded.Assertions.EventAssertions do
   end
 
   @doc """
-  Assert that an event of the given event type, matching the predicate, is published.
-  Verify that event using the assertion function.
+  Assert that an event of the given event type, matching the predicate, is
+  published. Verify that event using the assertion function.
 
   ## Examples
 
-      assert_receive_event BankAccountOpened,
+      assert_receive_event(BankAccountOpened,
         fn opened -> opened.account_number == "ACC123" end,
         fn opened ->
           assert opened.balance == 1_000
-        end
+        end)
 
   """
   def assert_receive_event(event_type, predicate_fn, assertion_fn) do
@@ -100,20 +105,18 @@ defmodule Commanded.Assertions.EventAssertions do
     end)
   end
 
-  defp default_receive_timeout,
-    do: Application.get_env(:commanded, :assert_receive_event_timeout, 1_000)
-
-  defp with_subscription(callback_fn) do
+  defp with_subscription(callback_fun) when is_function(callback_fun, 1) do
     subscription_name = UUID.uuid4()
 
-    {:ok, subscription} = create_subscription(subscription_name)
+    {:ok, subscription} = EventStore.subscribe_to(:all, subscription_name, self(), :origin)
 
     assert_receive {:subscribed, ^subscription}, default_receive_timeout()
 
     try do
-      apply(callback_fn, [subscription])
+      apply(callback_fun, [subscription])
     after
-      remove_subscription(subscription)
+      :ok = EventStore.unsubscribe(subscription)
+      :ok = EventStore.delete_subscription(:all, subscription_name)
     end
   end
 
@@ -162,9 +165,14 @@ defmodule Commanded.Assertions.EventAssertions do
     expected_event =
       Enum.find(received_events, fn received_event ->
         case received_event.event_type do
-          ^expected_type when is_function(predicate_fn, 2) -> apply(predicate_fn, [received_event.data, received_event])
-          ^expected_type when is_function(predicate_fn, 1) -> apply(predicate_fn, [received_event.data])
-          _ -> false
+          ^expected_type when is_function(predicate_fn, 2) ->
+            apply(predicate_fn, [received_event.data, received_event])
+
+          ^expected_type when is_function(predicate_fn, 1) ->
+            apply(predicate_fn, [received_event.data])
+
+          _ ->
+            false
         end
       end)
 
@@ -174,12 +182,8 @@ defmodule Commanded.Assertions.EventAssertions do
     end
   end
 
-  defp create_subscription(subscription_name),
-    do: EventStore.subscribe_to(:all, subscription_name, self(), :origin)
-
-  defp remove_subscription(subscription),
-    do: EventStore.unsubscribe(subscription)
+  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))
+  defp default_receive_timeout,
+    do: Application.get_env(:commanded, :assert_receive_event_timeout, 1_000)
 end
diff --git a/lib/commanded/event_store/adapters/in_memory.ex b/lib/commanded/event_store/adapters/in_memory.ex
index 84d266c..3544771 100644
--- a/lib/commanded/event_store/adapters/in_memory.ex
+++ b/lib/commanded/event_store/adapters/in_memory.ex
@@ -82,6 +82,11 @@ defmodule Commanded.EventStore.Adapters.InMemory do
     GenServer.call(__MODULE__, {:unsubscribe, subscription})
   end
 
+  @impl Commanded.EventStore
+  def delete_subscription(stream_uuid, subscription_name) do
+    GenServer.call(__MODULE__, {:delete_subscription, stream_uuid, subscription_name})
+  end
+
   @impl Commanded.EventStore
   def read_snapshot(source_uuid) do
     GenServer.call(__MODULE__, {:read_snapshot, source_uuid})
@@ -216,9 +221,11 @@ defmodule Commanded.EventStore.Adapters.InMemory do
           persistent_subscription(subscription, state)
 
         %Subscription{subscriber: nil} = subscription ->
-          persistent_subscription(%Subscription{subscription | subscriber: subscriber}, state)
+          subscription = %Subscription{subscription | subscriber: subscriber}
+
+          persistent_subscription(subscription, state)
 
-        _subscription ->
+        %Subscription{} ->
           {{:error, :subscription_already_exists}, state}
       end
 
@@ -226,17 +233,52 @@ defmodule Commanded.EventStore.Adapters.InMemory do
   end
 
   @impl GenServer
-  def handle_call({:unsubscribe, subscription}, _from, %State{} = state) do
+  def handle_call({:unsubscribe, pid}, _from, %State{} = state) do
     %State{persistent_subscriptions: subscriptions} = state
 
-    :ok = stop_subscription(subscription)
+    {reply, state} =
+      case Enum.find(subscriptions, fn
+             {_name, %Subscription{subscriber: ^pid}} -> true
+             {_name, %Subscription{}} -> false
+           end) do
+        {subscription_name, %Subscription{} = subscription} ->
+          :ok = stop_subscription(pid)
 
-    state = %State{
-      state
-      | persistent_subscriptions: remove_subscriber_by_pid(subscriptions, subscription)
-    }
+          subscription = %Subscription{subscription | subscriber: nil, ref: nil}
 
-    {:reply, :ok, state}
+          state = %State{
+            state
+            | persistent_subscriptions: Map.put(subscriptions, subscription_name, subscription)
+          }
+
+          {:ok, state}
+
+        nil ->
+          {:ok, state}
+      end
+
+    {:reply, reply, state}
+  end
+
+  @impl GenServer
+  def handle_call({:delete_subscription, stream_uuid, subscription_name}, _from, %State{} = state) do
+    %State{persistent_subscriptions: subscriptions} = state
+
+    {reply, state} =
+      case Map.get(subscriptions, subscription_name) do
+        %Subscription{stream_uuid: ^stream_uuid, subscriber: nil} ->
+          state = %State{
+            state
+            | persistent_subscriptions: Map.delete(subscriptions, subscription_name)
+          }
+
+          {:ok, state}
+
+        nil ->
+          {{:error, :subscription_not_found}, state}
+      end
+
+    {:reply, reply, state}
   end
 
   @impl GenServer
@@ -412,7 +454,7 @@ defmodule Commanded.EventStore.Adapters.InMemory do
 
   defp remove_subscriber_by_pid(subscriptions, pid) do
     Enum.reduce(subscriptions, subscriptions, fn
-      {name, %Subscription{subscriber: subscriber} = subscription}, acc when subscriber == pid ->
+      {name, %Subscription{subscriber: ^pid} = subscription}, acc ->
         Map.put(acc, name, %Subscription{subscription | subscriber: nil})
 
       _, acc ->
diff --git a/lib/commanded/event_store/event_store.ex b/lib/commanded/event_store/event_store.ex
index 5996c47..af7303e 100644
--- a/lib/commanded/event_store/event_store.ex
+++ b/lib/commanded/event_store/event_store.ex
@@ -100,9 +100,27 @@ defmodule Commanded.EventStore do
 
   @doc """
   Unsubscribe an existing subscriber from event notifications.
+
+  This will not delete the subscription.
+
+  ## Example
+
+      :ok = Commanded.EventStore.unsubscribe(subscription)
+
   """
   @callback unsubscribe(subscription) :: :ok
 
+  @doc """
+  Delete an existing subscription.
+
+  ## Example
+
+      :ok = Commanded.EventStore.delete_subscription(:all, "Example")
+
+  """
+  @callback delete_subscription(stream_uuid | :all, subscription_name) ::
+              :ok | {:error, :subscription_not_found} | {:error, error}
+
   @doc """
   Read a snapshot, if available, for a given source.
   """
@@ -192,6 +210,15 @@ defmodule Commanded.EventStore do
     event_store_adapter().unsubscribe(subscription)
   end
 
+  @doc """
+  Delete an existing subscription.
+  """
+  @spec delete_subscription(stream_uuid | :all, subscription_name) ::
+          :ok | {:error, :subscription_not_found} | {:error, error}
+  def delete_subscription(stream_uuid, subscription_name) do
+    event_store_adapter().delete_subscription(stream_uuid, subscription_name)
+  end
+
   @doc """
   Read a snapshot, if available, for a given source.
   """


Test patch:
diff --git a/test/event_store/support/subscription_test_case.ex b/test/event_store/support/subscription_test_case.ex
index d1e4a23..44ba2f0 100644
--- a/test/event_store/support/subscription_test_case.ex
+++ b/test/event_store/support/subscription_test_case.ex
@@ -253,6 +253,51 @@ defmodule Commanded.EventStore.SubscriptionTestCase do
 
         refute_receive {:events, _received_events}
       end
+
+      test "should resume subscription when subscribing again" do
+        {:ok, subscription1} = EventStore.subscribe_to(:all, "subscriber", self(), :origin)
+
+        assert_receive {:subscribed, ^subscription1}
+
+        :ok = EventStore.append_to_stream("stream1", 0, build_events(1))
+
+        assert_receive_events(subscription1, 1, from: 1)
+
+        :ok = EventStore.unsubscribe(subscription1)
+
+        {:ok, subscription2} = EventStore.subscribe_to(:all, "subscriber", self(), :origin)
+
+        :ok = EventStore.append_to_stream("stream2", 0, build_events(2))
+
+        assert_receive {:subscribed, ^subscription2}
+        assert_receive_events(subscription2, 2, from: 2)
+      end
+    end
+
+    describe "delete subscription" do
+      test "should be deleted" do
+        {:ok, subscription1} = EventStore.subscribe_to(:all, "subscriber", self(), :origin)
+
+        assert_receive {:subscribed, ^subscription1}
+
+        :ok = EventStore.append_to_stream("stream1", 0, build_events(1))
+
+        assert_receive_events(subscription1, 1, from: 1)
+
+        :ok = EventStore.unsubscribe(subscription1)
+        :ok = EventStore.delete_subscription(:all, "subscriber")
+
+        :ok = EventStore.append_to_stream("stream2", 0, build_events(2))
+
+        refute_receive {:events, _received_events}
+
+        {:ok, subscription2} = EventStore.subscribe_to(:all, "subscriber", self(), :origin)
+
+        # Should receive all events as subscription has been recreated from `:origin`
+        assert_receive {:subscribed, ^subscription2}
+        assert_receive_events(subscription2, 1, from: 1)
+        assert_receive_events(subscription2, 2, from: 2)
+      end
     end
 
     describe "resume subscription" do