feat: custom dispatcher for Presence handle_diff broadcast

lib/phoenix/presence.ex

@@ -79,6 +79,21 @@ defmodule Phoenix.Presence do
 
   See `c:list/1` for more information on the presence data structure.
 
+  ## Custom dispatcher
+
+  It's possible to customize the dispatcher module used to broadcast.
+  By default, `Phoenix.Channel.Server` is used, which is the same dispatcher
+  used by channels. To customize the dispatcher, pass the `:dispatcher` option
+  when using `Phoenix.Presence`:
+
+      use Phoenix.Presence,
+        otp_app: :my_app,
+        pubsub_server: MyApp.PubSub,
+        dispatcher: MyApp.CustomDispatcher
+
+  See `m:Phoenix.PubSub#module-custom-dispatching` for more information on
+  custom dispatchers.
+
   ## Fetching Presence Information
 
   Presence metadata should be minimized and used to store small,
@@ -417,9 +432,11 @@ defmodule Phoenix.Presence do
       pubsub_server =
         opts[:pubsub_server] || raise "use Phoenix.Presence expects :pubsub_server to be given"
 
+      dispatcher = opts[:dispatcher] || Phoenix.Channel.Server
+
       Phoenix.Tracker.start_link(
         __MODULE__,
-        {module, task_supervisor, pubsub_server},
+        {module, task_supervisor, pubsub_server, dispatcher},
         opts
       )
     end
@@ -455,15 +472,16 @@ defmodule Phoenix.Presence do
   end
 
   @doc false
-  def init({module, task_supervisor, pubsub_server}) do
+  def init({module, task_supervisor, pubsub_server, dispatcher}) do
     state = %{
       module: module,
       task_supervisor: task_supervisor,
       pubsub_server: pubsub_server,
       topics: %{},
       tasks: :queue.new(),
       current_task: nil,
-      client_state: nil
+      client_state: nil,
+      dispatcher: dispatcher
     }
 
     client_state =
@@ -507,12 +525,13 @@ defmodule Phoenix.Presence do
     Task.shutdown(task)
 
     Enum.each(computed_diffs, fn {topic, presence_diff} ->
-      Phoenix.Channel.Server.local_broadcast(
-        state.pubsub_server,
-        topic,
-        "presence_diff",
-        presence_diff
-      )
+      broadcast = %Phoenix.Socket.Broadcast{
+        topic: topic,
+        event: "presence_diff",
+        payload: presence_diff
+      }
+
+      Phoenix.PubSub.local_broadcast(state.pubsub_server, topic, broadcast, state.dispatcher)
     end)
 
     new_state =

test/phoenix/presence_test.exs

@@ -34,7 +34,8 @@ defmodule Phoenix.PresenceTest do
       Phoenix.Presence.init({
         __MODULE__,
         __MODULE__.TaskSupervisor,
-        PresPub
+        PresPub,
+        Phoenix.Channel.Server
       })
     end
 
@@ -43,13 +44,27 @@ defmodule Phoenix.PresenceTest do
     end
   end
 
+  defmodule CustomDispatcher do
+    def dispatch(entries, from, message) do
+      for {pid, _} <- entries, pid != from, do: send(pid, {:custom_dispatcher, message})
+
+      :ok
+    end
+  end
+
+  defmodule CustomDispatcherPresence do
+    use Phoenix.Presence, otp_app: :phoenix, dispatcher: CustomDispatcher
+  end
+
   Application.put_env(:phoenix, MyPresence, pubsub_server: PresPub)
   Application.put_env(:phoenix, MetasPresence, pubsub_server: PresPub)
+  Application.put_env(:phoenix, CustomDispatcherPresence, pubsub_server: PresPub)
 
   setup_all do
     start_supervised!({Phoenix.PubSub, name: PresPub, pool_size: 1})
     start_supervised!(MyPresence)
     start_supervised!(MetasPresence)
+    start_supervised!(CustomDispatcherPresence)
     {:ok, pubsub: PresPub}
   end
 
@@ -174,6 +189,40 @@ defmodule Phoenix.PresenceTest do
     assert MyPresence.list(topic) == %{}
   end
 
+  test "handle_diff with custom dispatcher", %{topic: topic} = config do
+    pid = spawn(fn -> :timer.sleep(:infinity) end)
+    Phoenix.PubSub.subscribe(config.pubsub, topic)
+    start_supervised!({DefaultPresence, pubsub_server: config.pubsub})
+    CustomDispatcherPresence.track(pid, topic, "u1", %{name: "u1"})
+
+    assert_receive {:custom_dispatcher,
+                    %Broadcast{
+                      topic: ^topic,
+                      event: "presence_diff",
+                      payload: %{
+                        joins: %{"u1" => %{metas: [%{name: "u1", phx_ref: u1_ref}]}},
+                        leaves: %{}
+                      }
+                    }}
+
+    assert %{"u1" => %{metas: [%{name: "u1", phx_ref: ^u1_ref}]}} =
+             CustomDispatcherPresence.list(topic)
+
+    Process.exit(pid, :kill)
+
+    assert_receive {:custom_dispatcher,
+                    %Broadcast{
+                      topic: ^topic,
+                      event: "presence_diff",
+                      payload: %{
+                        joins: %{},
+                        leaves: %{"u1" => %{metas: [%{name: "u1", phx_ref: ^u1_ref}]}}
+                      }
+                    }}
+
+    assert CustomDispatcherPresence.list(topic) == %{}
+  end
+
   test "untrack with pid", %{topic: topic} = config do
     Phoenix.PubSub.subscribe(config.pubsub, config.topic)
     MyPresence.track(self(), config.topic, "u1", %{})