Add configurable default dispatcher (#209)

Adds a :dispatcher option to child_spec/start_link that sets the
default dispatcher module for all broadcast functions. Defaults to
Phoenix.PubSub (preserving existing behavior). Can be overridden
per-call by passing a dispatcher to broadcast/4 and friends.

    {Phoenix.PubSub, name: :my_pubsub, dispatcher: MyApp.Dispatcher}

The internal meta :pubsub value is extended from {adapter, name}
to {adapter, name, dispatcher}. Single Registry.meta lookup per
broadcast call, same as before.

Author: twinn

lib/phoenix/pubsub.ex

@@ -171,6 +171,9 @@ defmodule Phoenix.PubSub do
     * `:broadcast_pool_size` - number of pubsub partitions used for broadcasting messages
       (defaults to `:pool_size`). This option is used during pool size migrations to ensure
       no messages are lost. See the "Safe Pool Size Migration" section in the module documentation.
+    * `:dispatcher` - the default dispatcher module for broadcasts
+      (defaults to `Phoenix.PubSub`). Can be overridden per-call by
+      passing a dispatcher to `broadcast/4` and friends.
 
   """
   @spec child_spec(keyword) :: Supervisor.child_spec()
@@ -249,9 +252,10 @@ defmodule Phoenix.PubSub do
   See the "Custom dispatching" section in the module documentation.
   """
   @spec broadcast(t, topic, message, dispatcher) :: :ok | {:error, term}
-  def broadcast(pubsub, topic, message, dispatcher \\ __MODULE__)
+  def broadcast(pubsub, topic, message, dispatcher \\ nil)
       when is_atom(pubsub) and is_binary(topic) and is_atom(dispatcher) do
-    {:ok, {adapter, name}} = Registry.meta(pubsub, :pubsub)
+    {:ok, {adapter, name, default_dispatcher}} = Registry.meta(pubsub, :pubsub)
+    dispatcher = dispatcher || default_dispatcher
 
     with :ok <- adapter.broadcast(name, topic, message, dispatcher) do
       dispatch(pubsub, :none, topic, message, dispatcher)
@@ -273,9 +277,10 @@ defmodule Phoenix.PubSub do
   See the "Custom dispatching" section in the module documentation.
   """
   @spec broadcast_from(t, pid, topic, message, dispatcher) :: :ok | {:error, term}
-  def broadcast_from(pubsub, from, topic, message, dispatcher \\ __MODULE__)
+  def broadcast_from(pubsub, from, topic, message, dispatcher \\ nil)
       when is_atom(pubsub) and is_pid(from) and is_binary(topic) and is_atom(dispatcher) do
-    {:ok, {adapter, name}} = Registry.meta(pubsub, :pubsub)
+    {:ok, {adapter, name, default_dispatcher}} = Registry.meta(pubsub, :pubsub)
+    dispatcher = dispatcher || default_dispatcher
 
     with :ok <- adapter.broadcast(name, topic, message, dispatcher) do
       dispatch(pubsub, from, topic, message, dispatcher)
@@ -293,9 +298,10 @@ defmodule Phoenix.PubSub do
   See the "Custom dispatching" section in the module documentation.
   """
   @spec local_broadcast(t, topic, message, dispatcher) :: :ok
-  def local_broadcast(pubsub, topic, message, dispatcher \\ __MODULE__)
+  def local_broadcast(pubsub, topic, message, dispatcher \\ nil)
       when is_atom(pubsub) and is_binary(topic) and is_atom(dispatcher) do
-    dispatch(pubsub, :none, topic, message, dispatcher)
+    {:ok, {_adapter, _name, default_dispatcher}} = Registry.meta(pubsub, :pubsub)
+    dispatch(pubsub, :none, topic, message, dispatcher || default_dispatcher)
   end
 
   @doc """
@@ -313,9 +319,10 @@ defmodule Phoenix.PubSub do
   See the "Custom dispatching" section in the module documentation.
   """
   @spec local_broadcast_from(t, pid, topic, message, dispatcher) :: :ok
-  def local_broadcast_from(pubsub, from, topic, message, dispatcher \\ __MODULE__)
+  def local_broadcast_from(pubsub, from, topic, message, dispatcher \\ nil)
       when is_atom(pubsub) and is_pid(from) and is_binary(topic) and is_atom(dispatcher) do
-    dispatch(pubsub, from, topic, message, dispatcher)
+    {:ok, {_adapter, _name, default_dispatcher}} = Registry.meta(pubsub, :pubsub)
+    dispatch(pubsub, from, topic, message, dispatcher || default_dispatcher)
   end
 
   @doc """
@@ -333,17 +340,17 @@ defmodule Phoenix.PubSub do
   See the "Custom dispatching" section in the module documentation.
   """
   @spec direct_broadcast(node_name, t, topic, message, dispatcher) :: :ok | {:error, term}
-  def direct_broadcast(node_name, pubsub, topic, message, dispatcher \\ __MODULE__)
+  def direct_broadcast(node_name, pubsub, topic, message, dispatcher \\ nil)
       when is_atom(pubsub) and is_binary(topic) and is_atom(dispatcher) do
-    {:ok, {adapter, name}} = Registry.meta(pubsub, :pubsub)
-    adapter.direct_broadcast(name, node_name, topic, message, dispatcher)
+    {:ok, {adapter, name, default_dispatcher}} = Registry.meta(pubsub, :pubsub)
+    adapter.direct_broadcast(name, node_name, topic, message, dispatcher || default_dispatcher)
   end
 
   @doc """
   Raising version of `broadcast/4`.
   """
   @spec broadcast!(t, topic, message, dispatcher) :: :ok
-  def broadcast!(pubsub, topic, message, dispatcher \\ __MODULE__) do
+  def broadcast!(pubsub, topic, message, dispatcher \\ nil) do
     case broadcast(pubsub, topic, message, dispatcher) do
       :ok -> :ok
       {:error, error} -> raise BroadcastError, "broadcast failed: #{inspect(error)}"
@@ -354,7 +361,7 @@ defmodule Phoenix.PubSub do
   Raising version of `broadcast_from/5`.
   """
   @spec broadcast_from!(t, pid, topic, message, dispatcher) :: :ok
-  def broadcast_from!(pubsub, from, topic, message, dispatcher \\ __MODULE__) do
+  def broadcast_from!(pubsub, from, topic, message, dispatcher \\ nil) do
     case broadcast_from(pubsub, from, topic, message, dispatcher) do
       :ok -> :ok
       {:error, error} -> raise BroadcastError, "broadcast failed: #{inspect(error)}"
@@ -365,7 +372,7 @@ defmodule Phoenix.PubSub do
   Raising version of `direct_broadcast/5`.
   """
   @spec direct_broadcast!(node_name, t, topic, message, dispatcher) :: :ok
-  def direct_broadcast!(node_name, pubsub, topic, message, dispatcher \\ __MODULE__) do
+  def direct_broadcast!(node_name, pubsub, topic, message, dispatcher \\ nil) do
     case direct_broadcast(node_name, pubsub, topic, message, dispatcher) do
       :ok -> :ok
       {:error, error} -> raise BroadcastError, "broadcast failed: #{inspect(error)}"
@@ -377,7 +384,7 @@ defmodule Phoenix.PubSub do
   """
   @spec node_name(t) :: node_name
   def node_name(pubsub) do
-    {:ok, {adapter, name}} = Registry.meta(pubsub, :pubsub)
+    {:ok, {adapter, name, _dispatcher}} = Registry.meta(pubsub, :pubsub)
     adapter.node_name(name)
   end
 

lib/phoenix/pubsub/supervisor.ex

@@ -25,8 +25,10 @@ defmodule Phoenix.PubSub.Supervisor do
       opts[:registry_size] || opts[:pool_size] ||
         System.schedulers_online() |> Kernel./(4) |> Float.ceil() |> trunc()
 
+    dispatcher = Keyword.get(opts, :dispatcher, Phoenix.PubSub)
+
     registry = [
-      meta: [pubsub: {adapter, adapter_name}],
+      meta: [pubsub: {adapter, adapter_name, dispatcher}],
       partitions: partitions,
       keys: :duplicate,
       name: name

test/phoenix/pubsub_test.exs

@@ -1,9 +1,11 @@
 defmodule Phoenix.PubSub.UnitTest do
   use ExUnit.Case, async: true
 
+  alias Phoenix.PubSub
+
   describe "child_spec/1" do
     test "expects a name" do
-      {:error, {{:EXIT, {exception, _}}, _}} = start_supervised({Phoenix.PubSub, []})
+      {:error, {{:EXIT, {exception, _}}, _}} = start_supervised({PubSub, []})
 
       assert Exception.message(exception) ==
                "the :name option is required when starting Phoenix.PubSub"
@@ -22,4 +24,82 @@ defmodule Phoenix.PubSub.UnitTest do
       :"#{__MODULE__}_#{:crypto.strong_rand_bytes(8) |> Base.encode16()}"
     end
   end
+
+  describe "default dispatcher" do
+    defmodule TestDispatcher do
+      def dispatch(entries, :none, message) do
+        for {pid, _} <- entries do
+          send(pid, {:custom_dispatched, message})
+        end
+
+        :ok
+      end
+
+      def dispatch(entries, from, message) do
+        for {pid, _} <- entries, pid != from do
+          send(pid, {:custom_dispatched, message})
+        end
+
+        :ok
+      end
+    end
+
+    test "defaults to Phoenix.PubSub when no dispatcher configured" do
+      name = :"ps_default_#{:erlang.unique_integer([:positive])}"
+      start_supervised!({PubSub, name: name})
+
+      PubSub.subscribe(name, "topic")
+      PubSub.broadcast(name, "topic", :hello)
+      assert_receive :hello
+    end
+
+    test "uses configured dispatcher for broadcast/3" do
+      name = :"ps_custom_#{:erlang.unique_integer([:positive])}"
+      start_supervised!({PubSub, name: name, dispatcher: TestDispatcher})
+
+      PubSub.subscribe(name, "topic")
+      PubSub.broadcast(name, "topic", :hello)
+      assert_receive {:custom_dispatched, :hello}
+      refute_received :hello
+    end
+
+    test "uses configured dispatcher for local_broadcast/3" do
+      name = :"ps_local_#{:erlang.unique_integer([:positive])}"
+      start_supervised!({PubSub, name: name, dispatcher: TestDispatcher})
+
+      PubSub.subscribe(name, "topic")
+      PubSub.local_broadcast(name, "topic", :hello)
+      assert_receive {:custom_dispatched, :hello}
+    end
+
+    test "uses configured dispatcher for broadcast_from/4" do
+      name = :"ps_from_#{:erlang.unique_integer([:positive])}"
+      start_supervised!({PubSub, name: name, dispatcher: TestDispatcher})
+
+      PubSub.subscribe(name, "topic")
+      other = spawn(fn -> Process.sleep(:infinity) end)
+      PubSub.broadcast_from(name, other, "topic", :hello)
+      assert_receive {:custom_dispatched, :hello}
+    end
+
+    test "explicit dispatcher overrides the configured default" do
+      name = :"ps_override_#{:erlang.unique_integer([:positive])}"
+      start_supervised!({PubSub, name: name, dispatcher: TestDispatcher})
+
+      PubSub.subscribe(name, "topic")
+      # Pass Phoenix.PubSub explicitly to override the configured TestDispatcher
+      PubSub.broadcast(name, "topic", :hello, PubSub)
+      assert_receive :hello
+      refute_received {:custom_dispatched, :hello}
+    end
+
+    test "bang variants use configured dispatcher" do
+      name = :"ps_bang_#{:erlang.unique_integer([:positive])}"
+      start_supervised!({PubSub, name: name, dispatcher: TestDispatcher})
+
+      PubSub.subscribe(name, "topic")
+      PubSub.broadcast!(name, "topic", :hello)
+      assert_receive {:custom_dispatched, :hello}
+    end
+  end
 end