Use the PubSub.Sender behavior for fastlaning

lib/phoenix/channel/server.ex

@@ -2,6 +2,8 @@ defmodule Phoenix.Channel.Server do
   @moduledoc false
   use GenServer, restart: :temporary
 
+  @behaviour Phoenix.PubSub.Sender
+
   require Logger
 
   alias Phoenix.PubSub
@@ -89,52 +91,44 @@ defmodule Phoenix.Channel.Server do
   ## Channel API
 
   @doc """
-  Hook invoked by Phoenix.PubSub dispatch.
+  Hook invoked by Phoenix.PubSub to deliver a message to a channel subscription.
+
+  Broadcasts not intercepted by the channel are encoded once per
+  serializer and sent directly to the transport process.
   """
-  def dispatch(subscribers, from, %Broadcast{event: event} = msg) do
-    Enum.reduce(subscribers, %{}, fn
-      {pid, _}, cache when pid == from ->
-        cache
-
-      {pid, {:fastlane, fastlane_pid, serializer, event_intercepts}}, cache ->
-        if event in event_intercepts do
-          send(pid, msg)
+  def send(pid, {transport_pid, serializer, intercepts}, %Broadcast{event: event} = msg, cache) do
+    if event in intercepts do
+      send(pid, msg)
+      cache
+    else
+      case cache do
+        %{^serializer => encoded_msg} ->
+          send(transport_pid, encoded_msg)
           cache
-        else
-          case cache do
-            %{^serializer => encoded_msg} ->
-              send(fastlane_pid, encoded_msg)
-              cache
-
-            %{} ->
-              encoded_msg = serializer.fastlane!(msg)
-              send(fastlane_pid, encoded_msg)
-              Map.put(cache, serializer, encoded_msg)
-          end
-        end
-
-      {pid, _}, cache ->
-        send(pid, msg)
-        cache
-    end)
-
-    :ok
-  end
 
-  def dispatch(entries, :none, message) do
-    for {pid, _} <- entries do
-      send(pid, message)
+        _ ->
+          encoded_msg = serializer.fastlane!(msg)
+          send(transport_pid, encoded_msg)
+          Map.put(cache || %{}, serializer, encoded_msg)
+      end
     end
+  end
 
-    :ok
+  def send(pid, _meta, msg, cache) do
+    send(pid, msg)
+    cache
   end
 
+  # TODO: Remove in Phoenix 1.10
+  #
+  # Nodes running Phoenix < 1.9 expect cluster-wide channel broadcasts to use
+  # this module as dispatcher, otherwise their channels receive broadcasts
+  # they cannot handle. Therefore we still broadcast with it, and it is also
+  # invoked for broadcasts coming from older nodes. Subscriptions carry their
+  # own sender, so we delegate to the default dispatching.
+  @doc false
   def dispatch(entries, from, message) do
-    for {pid, _} <- entries, pid != from do
-      send(pid, message)
-    end
-
-    :ok
+    PubSub.dispatch(entries, from, message)
   end
 
   @doc """
@@ -151,6 +145,7 @@ defmodule Phoenix.Channel.Server do
       payload: payload
     }
 
+    # TODO: Stop passing the dispatcher in 1.10, see the comment on dispatch/3
     PubSub.broadcast(pubsub_server, topic, broadcast, __MODULE__)
   end
 
@@ -168,6 +163,7 @@ defmodule Phoenix.Channel.Server do
       payload: payload
     }
 
+    # TODO: Stop passing the dispatcher in 1.10, see the comment on dispatch/3
     PubSub.broadcast!(pubsub_server, topic, broadcast, __MODULE__)
   end
 
@@ -185,6 +181,7 @@ defmodule Phoenix.Channel.Server do
       payload: payload
     }
 
+    # TODO: Stop passing the dispatcher in 1.10, see the comment on dispatch/3
     PubSub.broadcast_from(pubsub_server, from, topic, broadcast, __MODULE__)
   end
 
@@ -202,6 +199,7 @@ defmodule Phoenix.Channel.Server do
       payload: payload
     }
 
+    # TODO: Stop passing the dispatcher in 1.10, see the comment on dispatch/3
     PubSub.broadcast_from!(pubsub_server, from, topic, broadcast, __MODULE__)
   end
 
@@ -219,7 +217,7 @@ defmodule Phoenix.Channel.Server do
       payload: payload
     }
 
-    PubSub.local_broadcast(pubsub_server, topic, broadcast, __MODULE__)
+    PubSub.local_broadcast(pubsub_server, topic, broadcast)
   end
 
   @doc """
@@ -236,7 +234,7 @@ defmodule Phoenix.Channel.Server do
       payload: payload
     }
 
-    PubSub.local_broadcast_from(pubsub_server, from, topic, broadcast, __MODULE__)
+    PubSub.local_broadcast_from(pubsub_server, from, topic, broadcast)
   end
 
   @doc """
@@ -440,8 +438,8 @@ defmodule Phoenix.Channel.Server do
     end
 
     Process.monitor(transport_pid)
-    fastlane = {:fastlane, transport_pid, serializer, channel.__intercepts__()}
-    PubSub.subscribe(pubsub_server, topic, metadata: fastlane)
+    sender = {__MODULE__, {transport_pid, serializer, channel.__intercepts__()}}
+    PubSub.subscribe(pubsub_server, topic, sender: sender)
 
     {:noreply, %{socket | joined: true}}
   end

lib/phoenix/presence.ex

@@ -82,16 +82,18 @@ defmodule Phoenix.Presence do
   ## Custom dispatcher
 
   It's possible to customize the dispatcher module used to broadcast.
-  By default, presence uses the same dispatcher as channels. To customize the
+  By default, presence uses the default `Phoenix.PubSub` dispatching,
+  which respects the `:sender` given when subscribing. 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.
+  Custom dispatchers are deprecated in `Phoenix.PubSub`. Prefer customizing
+  delivery per subscription with the `:sender` option of `Phoenix.PubSub.subscribe/3`
+  instead. See `Phoenix.PubSub.Sender` for more information.
 
   ## Fetching Presence Information
 
@@ -431,7 +433,7 @@ 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
+      dispatcher = opts[:dispatcher]
 
       Phoenix.Tracker.start_link(
         __MODULE__,
@@ -530,7 +532,11 @@ defmodule Phoenix.Presence do
         payload: presence_diff
       }
 
-      Phoenix.PubSub.local_broadcast(state.pubsub_server, topic, broadcast, state.dispatcher)
+      if dispatcher = state.dispatcher do
+        Phoenix.PubSub.local_broadcast(state.pubsub_server, topic, broadcast, dispatcher)
+      else
+        Phoenix.PubSub.local_broadcast(state.pubsub_server, topic, broadcast)
+      end
     end)
 
     new_state =

mix.exs

@@ -82,7 +82,7 @@ defmodule Phoenix.MixProject do
       {:plug, "~> 1.14"},
       {:plug_crypto, "~> 2.2"},
       {:telemetry, "~> 0.4 or ~> 1.0"},
-      {:phoenix_pubsub, "~> 2.1"},
+      {:phoenix_pubsub, github: "phoenixframework/phoenix_pubsub", branch: "sd-sender"},
       {:phoenix_template, "~> 1.0"},
       {:websock_adapter, "~> 0.5"},
 

mix.lock

@@ -25,7 +25,7 @@
   "mint_web_socket": {:hex, :mint_web_socket, "1.0.5", "60354efeb49b1eccf95dfb75f55b08d692e211970fe735a5eb3188b328be2a90", [:mix], [{:mint, ">= 1.4.1 and < 2.0.0-0", [hex: :mint, repo: "hexpm", optional: false]}], "hexpm", "04b35663448fc758f3356cce4d6ac067ca418bbafe6972a3805df984b5f12e61"},
   "nimble_parsec": {:hex, :nimble_parsec, "1.4.2", "8efba0122db06df95bfaa78f791344a89352ba04baedd3849593bfce4d0dc1c6", [:mix], [], "hexpm", "4b21398942dda052b403bbe1da991ccd03a053668d147d53fb8c4e0efe09c973"},
   "phoenix_html": {:hex, :phoenix_html, "4.3.0", "d3577a5df4b6954cd7890c84d955c470b5310bb49647f0a114a6eeecc850f7ad", [:mix], [], "hexpm", "3eaa290a78bab0f075f791a46a981bbe769d94bc776869f4f3063a14f30497ad"},
-  "phoenix_pubsub": {:hex, :phoenix_pubsub, "2.2.0", "ff3a5616e1bed6804de7773b92cbccfc0b0f473faf1f63d7daf1206c7aeaaa6f", [:mix], [], "hexpm", "adc313a5bf7136039f63cfd9668fde73bba0765e0614cba80c06ac9460ff3e96"},
+  "phoenix_pubsub": {:git, "https://github.com/phoenixframework/phoenix_pubsub.git", "3128ce1d5093ced118e0c63e47802fe17734e423", [branch: "sd-sender"]},
   "phoenix_template": {:hex, :phoenix_template, "1.0.4", "e2092c132f3b5e5b2d49c96695342eb36d0ed514c5b252a77048d5969330d639", [:mix], [{:phoenix_html, "~> 2.14.2 or ~> 3.0 or ~> 4.0", [hex: :phoenix_html, repo: "hexpm", optional: true]}], "hexpm", "2c0c81f0e5c6753faf5cca2f229c9709919aba34fab866d3bc05060c9c444206"},
   "phoenix_view": {:hex, :phoenix_view, "2.0.4", "b45c9d9cf15b3a1af5fb555c674b525391b6a1fe975f040fb4d913397b31abf4", [:mix], [{:phoenix_html, "~> 2.14.2 or ~> 3.0 or ~> 4.0", [hex: :phoenix_html, repo: "hexpm", optional: true]}, {:phoenix_template, "~> 1.0", [hex: :phoenix_template, repo: "hexpm", optional: false]}], "hexpm", "4e992022ce14f31fe57335db27a28154afcc94e9983266835bb3040243eb620b"},
   "plug": {:hex, :plug, "1.20.3", "56c480c633ec2ce10140e236e15233bf576e1d323887d7c96711bd02ab5160db", [:mix], [{:mime, "~> 1.0 or ~> 2.0", [hex: :mime, repo: "hexpm", optional: false]}, {:plug_crypto, "~> 1.1.1 or ~> 1.2 or ~> 2.0", [hex: :plug_crypto, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4.3 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "be266aee1b8536ef6409d58cf39a3121319f0ec47cfa1b24024485aa0e76ad76"},

test/phoenix/channel_test.exs

@@ -4,6 +4,28 @@ defmodule Phoenix.Channel.ChannelTest do
   @pubsub __MODULE__.PubSub
   import Phoenix.Channel
 
+  defmodule ClusterAdapter do
+    # Sends what would be broadcast to other nodes to the test process
+    @behaviour Phoenix.PubSub.Adapter
+
+    def node_name(_adapter_name), do: node()
+
+    def child_spec(opts) do
+      test_pid = Keyword.fetch!(opts, :test_pid)
+      agent_opts = [name: opts[:adapter_name]]
+      %{id: __MODULE__, start: {Agent, :start_link, [fn -> test_pid end, agent_opts]}}
+    end
+
+    def broadcast(adapter_name, topic, message, dispatcher) do
+      send(Agent.get(adapter_name, & &1), {:cluster_broadcast, topic, message, dispatcher})
+      :ok
+    end
+
+    def direct_broadcast(adapter_name, _node_name, topic, message, dispatcher) do
+      broadcast(adapter_name, topic, message, dispatcher)
+    end
+  end
+
   setup_all do
     start_supervised! {Phoenix.PubSub, name: @pubsub, pool_size: 1}
     :ok
@@ -95,6 +117,39 @@ defmodule Phoenix.Channel.ChannelTest do
     }
   end
 
+  # TODO: Remove in Phoenix 1.10
+  test "broadcasts to other nodes with Phoenix.Channel.Server as dispatcher" do
+    pubsub = __MODULE__.ClusterPubSub
+    start_supervised!({Phoenix.PubSub, name: pubsub, adapter: ClusterAdapter, test_pid: self()})
+
+    socket = %Phoenix.Socket{
+      pubsub_server: pubsub,
+      topic: "sometopic",
+      channel_pid: spawn_link(fn -> :ok end),
+      joined: true
+    }
+
+    broadcast(socket, "event1", %{key: :val})
+
+    assert_receive {:cluster_broadcast, "sometopic", %Phoenix.Socket.Broadcast{event: "event1"},
+                    Phoenix.Channel.Server}
+
+    broadcast!(socket, "event2", %{key: :val})
+
+    assert_receive {:cluster_broadcast, "sometopic", %Phoenix.Socket.Broadcast{event: "event2"},
+                    Phoenix.Channel.Server}
+
+    broadcast_from(socket, "event3", %{key: :val})
+
+    assert_receive {:cluster_broadcast, "sometopic", %Phoenix.Socket.Broadcast{event: "event3"},
+                    Phoenix.Channel.Server}
+
+    broadcast_from!(socket, "event4", %{key: :val})
+
+    assert_receive {:cluster_broadcast, "sometopic", %Phoenix.Socket.Broadcast{event: "event4"},
+                    Phoenix.Channel.Server}
+  end
+
   test "pushing to transport" do
     socket = %Phoenix.Socket{
       serializer: Phoenix.ChannelTest.NoopSerializer,

test/phoenix/endpoint/endpoint_test.exs

@@ -42,6 +42,15 @@ defmodule Phoenix.Endpoint.EndpointTest do
     use Phoenix.Endpoint, otp_app: :phoenix
   end
 
+  defmodule TaggingSender do
+    @behaviour Phoenix.PubSub.Sender
+
+    def send(pid, tag, message, state) do
+      Kernel.send(pid, {tag, message})
+      state
+    end
+  end
+
   setup_all do
     ExUnit.CaptureLog.capture_log(fn -> start_supervised!(Endpoint) end)
     start_supervised!({Phoenix.PubSub, name: :endpoint_pub})
@@ -307,6 +316,22 @@ defmodule Phoenix.Endpoint.EndpointTest do
     }
   end
 
+  test "pubsub broadcasts are delivered through the subscription sender" do
+    Endpoint.subscribe("sendertopic", sender: {TaggingSender, :tagged})
+    some = spawn(fn -> :ok end)
+
+    Endpoint.broadcast("sendertopic", "event1", %{key: :val})
+    assert_receive {:tagged, %Phoenix.Socket.Broadcast{event: "event1", topic: "sendertopic"}}
+
+    Endpoint.broadcast_from!(some, "sendertopic", "event2", %{key: :val})
+    assert_receive {:tagged, %Phoenix.Socket.Broadcast{event: "event2", topic: "sendertopic"}}
+
+    Endpoint.local_broadcast("sendertopic", "event3", %{key: :val})
+    assert_receive {:tagged, %Phoenix.Socket.Broadcast{event: "event3", topic: "sendertopic"}}
+
+    refute_received %Phoenix.Socket.Broadcast{}
+  end
+
   test "loads cache manifest from specified application" do
     config =
       put_in(

test/phoenix/presence_test.exs

@@ -35,7 +35,7 @@ defmodule Phoenix.PresenceTest do
         __MODULE__,
         __MODULE__.TaskSupervisor,
         PresPub,
-        Phoenix.Channel.Server
+        nil
       })
     end
 
@@ -52,6 +52,15 @@ defmodule Phoenix.PresenceTest do
     end
   end
 
+  defmodule TaggingSender do
+    @behaviour Phoenix.PubSub.Sender
+
+    def send(pid, tag, message, state) do
+      Kernel.send(pid, {tag, message})
+      state
+    end
+  end
+
   defmodule CustomDispatcherPresence do
     use Phoenix.Presence, otp_app: :phoenix, dispatcher: CustomDispatcher
   end
@@ -189,6 +198,21 @@ defmodule Phoenix.PresenceTest do
     assert MyPresence.list(topic) == %{}
   end
 
+  test "handle_diff broadcasts through the subscription sender", %{topic: topic} = config do
+    pid = spawn(fn -> :timer.sleep(:infinity) end)
+    Phoenix.PubSub.subscribe(config.pubsub, topic, sender: {TaggingSender, :tagged})
+    MyPresence.track(pid, topic, "u1", %{name: "u1"})
+
+    assert_receive {:tagged,
+                    %Broadcast{
+                      topic: ^topic,
+                      event: "presence_diff",
+                      payload: %{joins: %{"u1" => _}, leaves: %{}}
+                    }}
+
+    refute_received %Broadcast{}
+  end
+
   test "handle_diff with custom dispatcher", %{topic: topic} = config do
     pid = spawn(fn -> :timer.sleep(:infinity) end)
     Phoenix.PubSub.subscribe(config.pubsub, topic)

test/phoenix/test/channel_test.exs

@@ -454,6 +454,28 @@ defmodule Phoenix.Test.ChannelTest do
     assert_graceful_exit(pid)
   end
 
+  test "push broadcasts sent directly through pubsub" do
+    subscribe_and_join!(socket(UserSocket), Channel, "foo:ok")
+    broadcast = %Broadcast{topic: "foo:ok", event: "default", payload: %{"foo" => "bar"}}
+    Phoenix.PubSub.broadcast!(Phoenix.Test.ChannelTest.PubSub, "foo:ok", broadcast)
+    assert_push "default", %{"foo" => "bar"}
+  end
+
+  test "handles broadcasts dispatched by Phoenix.Channel.Server from older nodes" do
+    Process.flag(:trap_exit, true)
+    {:ok, _, socket} = subscribe_and_join(socket(UserSocket), Channel, "foo:ok")
+    pubsub = Phoenix.Test.ChannelTest.PubSub
+
+    broadcast = %Broadcast{topic: "foo:ok", event: "default", payload: %{"foo" => "bar"}}
+    Phoenix.PubSub.local_broadcast(pubsub, "foo:ok", broadcast, Phoenix.Channel.Server)
+    assert_push "default", %{"foo" => "bar"}
+
+    broadcast = %Broadcast{topic: "foo:ok", event: "stop", payload: %{"foo" => "bar"}}
+    Phoenix.PubSub.local_broadcast(pubsub, "foo:ok", broadcast, Phoenix.Channel.Server)
+    assert_receive {:terminate, :shutdown}
+    assert_graceful_exit(socket.channel_pid)
+  end
+
   ## handle_info
 
   test "handles messages and stops" do