Repository: commanded/commanded
Pull request #180 (2018-07-24 11:39:49)

Fix error when subscription process already tracked
`Phoenix.Tracker` returns `{:error, {:already_tracked, _pid, _topic, _key}}` if the process is already tracked for a given topic/key pair. This is now handled as ok, rather than erroring.


Patch:
diff --git a/lib/commanded/process_managers/process_router.ex b/lib/commanded/process_managers/process_router.ex
index 6430e6f..47eada0 100644
--- a/lib/commanded/process_managers/process_router.ex
+++ b/lib/commanded/process_managers/process_router.ex
@@ -142,7 +142,14 @@ defmodule Commanded.ProcessManagers.ProcessRouter do
   def handle_info({:subscribed, subscription}, %State{subscription: subscription} = state) do
     Logger.debug(fn -> describe(state) <> " has successfully subscribed to event store" end)
 
-    %State{command_dispatcher: command_dispatcher} = state
+    %State{
+      consistency: consistency,
+      command_dispatcher: command_dispatcher,
+      process_manager_name: process_manager_name
+    } = state
+
+    # Register this process manager as a subscription with the given consistency
+    :ok = Subscriptions.register(process_manager_name, consistency)
 
     {:ok, supervisor} = Supervisor.start_link(command_dispatcher)
 
@@ -185,11 +192,10 @@ defmodule Commanded.ProcessManagers.ProcessRouter do
     {:stop, reason, state}
   end
 
-  defp subscribe_to_all_streams(%State{consistency: consistency, process_manager_name: process_manager_name, subscribe_from: subscribe_from} = state) do
-    {:ok, subscription} = EventStore.subscribe_to_all_streams(process_manager_name, self(), subscribe_from)
+  defp subscribe_to_all_streams(%State{} = state) do
+    %State{process_manager_name: process_manager_name, subscribe_from: subscribe_from} = state
 
-    # register this event handler as a subscription with the given consistency
-    :ok = Subscriptions.register(process_manager_name, consistency)
+    {:ok, subscription} = EventStore.subscribe_to_all_streams(process_manager_name, self(), subscribe_from)
 
     %State{state | subscription: subscription}
   end
diff --git a/lib/commanded/pubsub/local_pubsub.ex b/lib/commanded/pubsub/local_pubsub.ex
index 203d566..8c2ea28 100644
--- a/lib/commanded/pubsub/local_pubsub.ex
+++ b/lib/commanded/pubsub/local_pubsub.ex
@@ -60,9 +60,14 @@ defmodule Commanded.PubSub.LocalPubSub do
   @spec track(String.t(), term) :: :ok
   @impl Commanded.PubSub
   def track(topic, key) when is_binary(topic) do
-    {:ok, _} = Registry.register(LocalPubSub.Tracker, topic, key)
+    case Registry.match(LocalPubSub.Tracker, topic, key) do
+      [] ->
+        {:ok, _pid} = Registry.register(LocalPubSub.Tracker, topic, key)
+        :ok
 
-    :ok
+      _matches ->
+        :ok
+    end
   end
 
   @doc """
diff --git a/lib/commanded/pubsub/phoenix_pubsub.ex b/lib/commanded/pubsub/phoenix_pubsub.ex
index 3e4f61d..d301db4 100644
--- a/lib/commanded/pubsub/phoenix_pubsub.ex
+++ b/lib/commanded/pubsub/phoenix_pubsub.ex
@@ -35,8 +35,6 @@ defmodule Commanded.PubSub.PhoenixPubSub do
 
   @behaviour Commanded.PubSub
 
-  alias Phoenix.PubSub
-
   defmodule Tracker do
     @behaviour Phoenix.Tracker
 
@@ -88,7 +86,7 @@ defmodule Commanded.PubSub.PhoenixPubSub do
   @spec subscribe(atom) :: :ok | {:error, term}
   @impl Commanded.PubSub
   def subscribe(topic) when is_binary(topic) do
-    PubSub.subscribe(__MODULE__, topic)
+    Phoenix.PubSub.subscribe(__MODULE__, topic)
   end
 
   @doc """
@@ -97,7 +95,7 @@ defmodule Commanded.PubSub.PhoenixPubSub do
   @spec broadcast(String.t(), term) :: :ok | {:error, term}
   @impl Commanded.PubSub
   def broadcast(topic, message) when is_binary(topic) do
-    PubSub.broadcast(__MODULE__, topic, message)
+    Phoenix.PubSub.broadcast(__MODULE__, topic, message)
   end
 
   @doc """
@@ -107,8 +105,11 @@ defmodule Commanded.PubSub.PhoenixPubSub do
   @spec track(String.t(), term) :: :ok
   @impl Commanded.PubSub
   def track(topic, key) when is_binary(topic) do
-    {:ok, _ref} = Phoenix.Tracker.track(Tracker, self(), topic, key, %{pid: self()})
-    :ok
+    case Phoenix.Tracker.track(Tracker, self(), topic, key, %{pid: self()}) do
+      {:ok, _ref} -> :ok
+      {:error, {:already_tracked, _pid, _topic, _key}} -> :ok
+      reply -> reply
+    end
   end
 
   @doc """
diff --git a/mix.exs b/mix.exs
index f4c6b79..212ce2d 100644
--- a/mix.exs
+++ b/mix.exs
@@ -39,6 +39,7 @@ defmodule Commanded.Mixfile do
       "test/example_domain",
       "test/helpers",
       "test/process_managers/support",
+      "test/pubsub/support",
       "test/registration/support",
       "test/support"
     ]


Test patch:
diff --git a/test/pubsub/local_pubsub_test.exs b/test/pubsub/local_pubsub_test.exs
index a76f08a..9f3d88b 100644
--- a/test/pubsub/local_pubsub_test.exs
+++ b/test/pubsub/local_pubsub_test.exs
@@ -1,35 +1,17 @@
 defmodule Commanded.PubSub.LocalPubSubTest do
-  use ExUnit.Case
+  alias Commanded.PubSub.{LocalPubSub, PubSubTestCase}
 
-  alias Commanded.PubSub.LocalPubSub
-
-  @topic "test"
+  use PubSubTestCase, pubsub: LocalPubSub
 
   setup do
     Application.put_env(:commanded, :pubsub, :local)
 
-    {:ok, _pid} = Supervisor.start_link(LocalPubSub.child_spec(), strategy: :one_for_one)
+    if Process.whereis(Commanded.PubSub.LocalPubSub) == nil do
+      {:ok, _pid} = Supervisor.start_link(LocalPubSub.child_spec(), strategy: :one_for_one)
+    end
 
     on_exit(fn ->
       Application.delete_env(:commanded, :pubsub)
     end)
   end
-
-  describe "pub/sub" do
-    test "should receive broadcast message" do
-      assert :ok = LocalPubSub.subscribe(@topic)
-      assert :ok = LocalPubSub.broadcast(@topic, :message)
-
-      assert_receive(:message)
-    end
-  end
-
-  describe "tracker" do
-    test "should list tracked processes" do
-      self = self()
-
-      assert :ok = LocalPubSub.track(@topic, :example)
-      assert [{:example, ^self}] = LocalPubSub.list(@topic)
-    end
-  end
 end
diff --git a/test/pubsub/phoenix_pubsub_test.exs b/test/pubsub/phoenix_pubsub_test.exs
index 3c50120..8ca92eb 100644
--- a/test/pubsub/phoenix_pubsub_test.exs
+++ b/test/pubsub/phoenix_pubsub_test.exs
@@ -1,9 +1,7 @@
 defmodule Commanded.PubSub.PhoenixPubSubTest do
-  use ExUnit.Case
+  alias Commanded.PubSub.{PhoenixPubSub, PubSubTestCase}
 
-  alias Commanded.PubSub.PhoenixPubSub
-
-  @topic "test"
+  use PubSubTestCase, pubsub: PhoenixPubSub
 
   setup do
     Application.put_env(
@@ -21,22 +19,4 @@ defmodule Commanded.PubSub.PhoenixPubSubTest do
       Application.delete_env(:commanded, :pubsub)
     end)
   end
-
-  describe "pub/sub" do
-    test "should receive broadcast message" do
-      assert :ok = PhoenixPubSub.subscribe(@topic)
-      assert :ok = PhoenixPubSub.broadcast(@topic, :message)
-
-      assert_receive(:message)
-    end
-  end
-
-  describe "tracker" do
-    test "should list tracked processes" do
-      self = self()
-
-      assert :ok = PhoenixPubSub.track(@topic, :example)
-      assert [{:example, ^self}] = PhoenixPubSub.list(@topic)
-    end
-  end
 end
diff --git a/test/pubsub/support/pubsub_test_case.ex b/test/pubsub/support/pubsub_test_case.ex
new file mode 100644
index 0000000..5fe4b2f
--- /dev/null
+++ b/test/pubsub/support/pubsub_test_case.ex
@@ -0,0 +1,36 @@
+defmodule Commanded.PubSub.PubSubTestCase do
+  import Commanded.SharedTestCase
+
+  define_tests do
+    describe "pub/sub" do
+      test "should receive broadcast message", %{pubsub: pubsub} do
+        :ok = pubsub.subscribe("test")
+        :ok = pubsub.broadcast("test", :message)
+
+        assert_receive(:message)
+        refute_receive(:message)
+      end
+    end
+
+    describe "tracker" do
+      test "should list tracked processes", %{pubsub: pubsub} do
+        self = self()
+
+        :ok = pubsub.track("test", :example1)
+        :ok = pubsub.track("test", :example2)
+
+        assert [{:example1, ^self}, {:example2, ^self}] = pubsub.list("test")
+      end
+
+      test "should ignore duplicate tracks", %{pubsub: pubsub} do
+        self = self()
+
+        :ok = pubsub.track("test", :example)
+        :ok = pubsub.track("test", :example)
+        :ok = pubsub.track("test", :example)
+
+        assert [{:example, ^self}] = pubsub.list("test")
+      end
+    end
+  end
+end
diff --git a/test/support/shared_test_case.ex b/test/support/shared_test_case.ex
new file mode 100644
index 0000000..3e8bc54
--- /dev/null
+++ b/test/support/shared_test_case.ex
@@ -0,0 +1,16 @@
+defmodule Commanded.SharedTestCase do
+  defmacro define_tests(do: block) do
+    quote do
+      defmacro __using__(options) do
+        block = unquote(Macro.escape(block))
+
+        quote do
+          use ExUnit.Case
+
+          @moduletag unquote(options)
+          unquote(block)
+        end
+      end
+    end
+  end
+end