Add subscription ordinal support

Author: bernardd

lib/absinthe/phoenix/channel.ex

@@ -37,7 +37,6 @@ defmodule Absinthe.Phoenix.Channel do
     socket =
       socket
       |> assign(:absinthe, absinthe_config)
-      |> assign(:async_procs, [])
     {:ok, socket}
   end
 
@@ -90,31 +89,56 @@ defmodule Absinthe.Phoenix.Channel do
     {:reply, {:ok, %{subscriptionId: doc_id}}, socket}
   end
 
+  def handle_info(
+    %Phoenix.Socket.Broadcast{payload: %{result: %{ordinal: ordinal}}} = msg,
+    socket
+  ) when not is_nil(ordinal) do
+    absinthe_assigns = Map.get(socket.assigns, :absinthe, %{})
+    last_ordinal = absinthe_assigns[:subscription_ordinals][msg.topic]
+
+    cond do
+      last_ordinal == nil or last_ordinal < ordinal ->
+        send_msg(msg, socket)
+        socket = update_ordinal(socket, msg.topic, ordinal)
+        {:noreply, socket}
+      true ->
+        {:noreply, socket}
+    end
+  end
+
+  def handle_info(msg, socket) do
+    send_msg(msg, socket)
+    {:noreply, socket}
+  end
+
+  defp send_msg(msg, socket) do
+    encoded_msg = socket.serializer.fastlane!(msg)
+    send(socket.transport_pid, encoded_msg)
+  end
+
+  defp update_ordinal(socket, topic, ordinal) do
+    absinthe_assigns = Map.get(socket, :absinthe, %{})
+    ordinals =
+      absinthe_assigns
+      |> Map.get(:subscription_ordinals, %{})
+      |> Map.put(topic, ordinal)
+
+    Phoenix.Socket.assign(socket, :absinthe, Map.put(absinthe_assigns, :subscription_ordinals, ordinals))
+  end
+
   defp run_doc(socket, query, config, opts) do
     case run(query, config[:schema], config[:pipeline], opts) do
       {:ok, %{"subscribed" => topic}, context} ->
-        %{transport_pid: transport_pid, serializer: serializer, pubsub_server: pubsub_server} =
-          socket
-
-        :ok =
-          Phoenix.PubSub.subscribe(
-            pubsub_server,
-            topic,
-            metadata: {:fastlane, transport_pid, serializer, []},
-            link: true
-          )
-
+        pubsub_subscribe(topic, socket)
         socket = Absinthe.Phoenix.Socket.put_options(socket, context: context)
+
         {{:ok, %{subscriptionId: topic}}, socket}
 
       {:more, %{"subscribed" => topic}, continuation, context} ->
-        socket = Absinthe.Phoenix.Socket.put_options(socket, context: context)
         reply(socket_ref(socket), {:ok, %{subscriptionId: topic}})
 
-        :ok = Phoenix.PubSub.subscribe(socket.pubsub_server, topic, [
-          fastlane: {socket.transport_pid, socket.serializer, []},
-          link: true,
-        ])
+        pubsub_subscribe(topic, socket)
+        socket = Absinthe.Phoenix.Socket.put_options(socket, context: context)
 
         handle_subscription_continuation(continuation, topic, socket)
 
@@ -129,10 +153,11 @@ defmodule Absinthe.Phoenix.Channel do
         {{:error, reply}, socket}
 
       {:more, %{data: _} = reply, continuation, context} ->
-        socket = Absinthe.Phoenix.Socket.put_options(socket, context: context)
-
         id = new_query_id()
-        socket = handle_continuation(continuation, id, socket)
+        socket =
+          socket
+          |> Absinthe.Phoenix.Socket.put_options(context: context)
+          |> handle_continuation(continuation, id)
 
         {{:ok, add_query_id(reply, id)}, socket}
 
@@ -154,6 +179,20 @@ defmodule Absinthe.Phoenix.Channel do
     end
   end
 
+  defp pubsub_subscribe(
+    topic,
+    %{transport_pid: transport_pid, serializer: serializer, pubsub_server: pubsub_server}
+  ) do
+    :ok =
+      Phoenix.PubSub.subscribe(
+        pubsub_server,
+        topic,
+        metadata: {:fastlane, transport_pid, serializer, ["subscription:data"]},
+        link: true
+      )
+  end
+
+
   defp extract_variables(payload) do
     case Map.get(payload, "variables", %{}) do
       nil -> %{}
@@ -167,40 +206,16 @@ defmodule Absinthe.Phoenix.Channel do
     |> Absinthe.Pipeline.for_document(options)
   end
 
-  def handle_info({:DOWN, ref, :process, _pid, _reason}, socket) do
-    procs = List.delete(socket.assigns.async_procs, ref)
-    {:noreply, assign(socket, :async_procs, procs)}
-  end
-
-  def handle_info(_, state) do
-    {:noreply, state}
-  end
-
-  defp handle_continuation(continuation, id, socket) do
-    max_procs = socket.assigns[:max_async_procs] || 0
-
-    case socket.assigns.async_procs do
-      procs when length(procs) < max_procs ->
-        {_pid, ref} = Process.spawn(
-          fn -> do_handle_continuation(continuation, id, socket) end,
-          [:link, :monitor])
-        assign(socket, :async_procs, [ref | procs])
-      _ ->
-        do_handle_continuation(continuation, id, socket)
-        socket
-    end
-  end
-
-  defp do_handle_continuation(continuation, id, socket) do
+  defp handle_continuation(socket, continuation, id) do
     case Absinthe.Pipeline.continue(continuation) do
-      {:ok, %{result: %{continuation: continuation} = result}, _phases} ->
+      {:ok, %{result: %{continuation: next_continuation} = result}, _phases} ->
         result =
           result
           |> Map.delete(:continuation)
           |> add_query_id(id)
 
         push socket, "doc", result
-        do_handle_continuation(continuation, id, socket)
+        handle_continuation(socket, next_continuation, id)
       {:ok, %{result: result}, _phases} ->
         push socket, "doc", add_query_id(result, id)
       {:ok, %{errors: errors}, _phases} ->

mix.lock

@@ -1,20 +1,21 @@
 %{
-  "absinthe": {:git, "https://github.com/circles-learning-labs/absinthe.git", "a88e7c2fc645b489815eafbd423f93e01754bbaf", [branch: "subscription-prime"]},
+  "absinthe": {:git, "https://github.com/circles-learning-labs/absinthe.git", "d7bddaa17f01b475932ea1e69866a09f478d4726", [branch: "subscription-prime"]},
   "absinthe_plug": {:hex, :absinthe_plug, "1.5.0", "018ef544cf577339018d1f482404b4bed762e1b530c78be9de4bbb88a6f3a805", [:mix], [{:absinthe, "~> 1.5.0", [hex: :absinthe, repo: "hexpm", optional: false]}, {:plug, "~> 1.3.2 or ~> 1.4", [hex: :plug, repo: "hexpm", optional: false]}], "hexpm", "4c160f4ce9a1233a4219a42de946e4e05d0e8733537cd5d8d20e7d4ef8d4b7c7"},
-  "decimal": {:hex, :decimal, "1.9.0", "83e8daf59631d632b171faabafb4a9f4242c514b0a06ba3df493951c08f64d07", [:mix], [], "hexpm", "b1f2343568eed6928f3e751cf2dffde95bfaa19dd95d09e8a9ea92ccfd6f7d85"},
+  "decimal": {:hex, :decimal, "2.0.0", "a78296e617b0f5dd4c6caf57c714431347912ffb1d0842e998e9792b5642d697", [:mix], [], "hexpm", "34666e9c55dea81013e77d9d87370fe6cb6291d1ef32f46a1600230b1d44f577"},
   "earmark": {:hex, :earmark, "1.4.4", "4821b8d05cda507189d51f2caeef370cf1e18ca5d7dfb7d31e9cafe6688106a4", [:mix], [], "hexpm", "1f93aba7340574847c0f609da787f0d79efcab51b044bb6e242cae5aca9d264d"},
   "earmark_parser": {:hex, :earmark_parser, "1.4.12", "b245e875ec0a311a342320da0551da407d9d2b65d98f7a9597ae078615af3449", [:mix], [], "hexpm", "711e2cc4d64abb7d566d43f54b78f7dc129308a63bc103fbd88550d2174b3160"},
   "ex_doc": {:hex, :ex_doc, "0.24.1", "15673de99154f93ca7f05900e4e4155ced1ee0cd34e0caeee567900a616871a4", [:mix], [{:earmark_parser, "~> 1.4.0", [hex: :earmark_parser, repo: "hexpm", optional: false]}, {:makeup_elixir, "~> 0.14", [hex: :makeup_elixir, repo: "hexpm", optional: false]}, {:makeup_erlang, "~> 0.1", [hex: :makeup_erlang, repo: "hexpm", optional: false]}], "hexpm", "07972f17bdf7dc7b5bd76ec97b556b26178ed3f056e7ec9288eb7cea7f91cce2"},
-  "jason": {:hex, :jason, "1.2.1", "12b22825e22f468c02eb3e4b9985f3d0cb8dc40b9bd704730efa11abd2708c44", [:mix], [{:decimal, "~> 1.0", [hex: :decimal, repo: "hexpm", optional: true]}], "hexpm", "b659b8571deedf60f79c5a608e15414085fa141344e2716fbd6988a084b5f993"},
+  "jason": {:hex, :jason, "1.3.0", "fa6b82a934feb176263ad2df0dbd91bf633d4a46ebfdffea0c8ae82953714946", [:mix], [{:decimal, "~> 1.0 or ~> 2.0", [hex: :decimal, repo: "hexpm", optional: true]}], "hexpm", "53fc1f51255390e0ec7e50f9cb41e751c260d065dcba2bf0d08dc51a4002c2ac"},
   "makeup": {:hex, :makeup, "1.0.5", "d5a830bc42c9800ce07dd97fa94669dfb93d3bf5fcf6ea7a0c67b2e0e4a7f26c", [:mix], [{:nimble_parsec, "~> 0.5 or ~> 1.0", [hex: :nimble_parsec, repo: "hexpm", optional: false]}], "hexpm", "cfa158c02d3f5c0c665d0af11512fed3fba0144cf1aadee0f2ce17747fba2ca9"},
   "makeup_elixir": {:hex, :makeup_elixir, "0.14.1", "4f0e96847c63c17841d42c08107405a005a2680eb9c7ccadfd757bd31dabccfb", [:mix], [{:makeup, "~> 1.0", [hex: :makeup, repo: "hexpm", optional: false]}], "hexpm", "f2438b1a80eaec9ede832b5c41cd4f373b38fd7aa33e3b22d9db79e640cbde11"},
   "makeup_erlang": {:hex, :makeup_erlang, "0.1.1", "3fcb7f09eb9d98dc4d208f49cc955a34218fc41ff6b84df7c75b3e6e533cc65f", [:mix], [{:makeup, "~> 1.0", [hex: :makeup, repo: "hexpm", optional: false]}], "hexpm", "174d0809e98a4ef0b3309256cbf97101c6ec01c4ab0b23e926a9e17df2077cbb"},
-  "mime": {:hex, :mime, "1.3.1", "30ce04ab3175b6ad0bdce0035cba77bba68b813d523d1aac73d9781b4d193cf8", [:mix], [], "hexpm", "6cbe761d6a0ca5a31a0931bf4c63204bceb64538e664a8ecf784a9a6f3b875f1"},
+  "mime": {:hex, :mime, "2.0.2", "0b9e1a4c840eafb68d820b0e2158ef5c49385d17fb36855ac6e7e087d4b1dcc5", [:mix], [], "hexpm", "e6a3f76b4c277739e36c2e21a2c640778ba4c3846189d5ab19f97f126df5f9b7"},
   "nimble_parsec": {:hex, :nimble_parsec, "1.2.3", "244836e6e3f1200c7f30cb56733fd808744eca61fd182f731eac4af635cc6d0b", [:mix], [], "hexpm", "c8d789e39b9131acf7b99291e93dae60ab48ef14a7ee9d58c6964f59efb570b0"},
   "phoenix": {:hex, :phoenix, "1.5.1", "95156589879dc69201d5fc0ebdbfdfc7901a09a3616ea611ec297f81340275a2", [:mix], [{:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: true]}, {:phoenix_html, "~> 2.13", [hex: :phoenix_html, repo: "hexpm", optional: true]}, {:phoenix_pubsub, "~> 2.0", [hex: :phoenix_pubsub, repo: "hexpm", optional: false]}, {:plug, "~> 1.10", [hex: :plug, repo: "hexpm", optional: false]}, {:plug_cowboy, "~> 1.0 or ~> 2.2", [hex: :plug_cowboy, repo: "hexpm", optional: true]}, {:plug_crypto, "~> 1.1.2 or ~> 1.2", [hex: :plug_crypto, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "fc272b38e79d2881790fccae6f67a9fbe9b790103d6878175ea03d23003152eb"},
-  "phoenix_html": {:hex, :phoenix_html, "2.14.2", "b8a3899a72050f3f48a36430da507dd99caf0ac2d06c77529b1646964f3d563e", [:mix], [{:plug, "~> 1.5", [hex: :plug, repo: "hexpm", optional: false]}], "hexpm", "58061c8dfd25da5df1ea0ca47c972f161beb6c875cd293917045b92ffe1bf617"},
-  "phoenix_pubsub": {:hex, :phoenix_pubsub, "2.0.0", "a1ae76717bb168cdeb10ec9d92d1480fec99e3080f011402c0a2d68d47395ffb", [:mix], [], "hexpm", "c52d948c4f261577b9c6fa804be91884b381a7f8f18450c5045975435350f771"},
-  "plug": {:hex, :plug, "1.10.0", "6508295cbeb4c654860845fb95260737e4a8838d34d115ad76cd487584e2fc4d", [:mix], [{:mime, "~> 1.0", [hex: :mime, repo: "hexpm", optional: false]}, {:plug_crypto, "~> 1.1.1 or ~> 1.2", [hex: :plug_crypto, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4", [hex: :telemetry, repo: "hexpm", optional: true]}], "hexpm", "422a9727e667be1bf5ab1de03be6fa0ad67b775b2d84ed908f3264415ef29d4a"},
-  "plug_crypto": {:hex, :plug_crypto, "1.1.2", "bdd187572cc26dbd95b87136290425f2b580a116d3fb1f564216918c9730d227", [:mix], [], "hexpm", "6b8b608f895b6ffcfad49c37c7883e8df98ae19c6a28113b02aa1e9c5b22d6b5"},
+  "phoenix_html": {:hex, :phoenix_html, "2.14.3", "51f720d0d543e4e157ff06b65de38e13303d5778a7919bcc696599e5934271b8", [:mix], [{:plug, "~> 1.5", [hex: :plug, repo: "hexpm", optional: false]}], "hexpm", "efd697a7fff35a13eeeb6b43db884705cba353a1a41d127d118fda5f90c8e80f"},
+  "phoenix_pubsub": {:hex, :phoenix_pubsub, "2.1.1", "ba04e489ef03763bf28a17eb2eaddc2c20c6d217e2150a61e3298b0f4c2012b5", [:mix], [], "hexpm", "81367c6d1eea5878ad726be80808eb5a787a23dee699f96e72b1109c57cdd8d9"},
+  "phoenix_view": {:hex, :phoenix_view, "1.1.2", "1b82764a065fb41051637872c7bd07ed2fdb6f5c3bd89684d4dca6e10115c95a", [:mix], [{:phoenix_html, "~> 2.14.2 or ~> 3.0", [hex: :phoenix_html, repo: "hexpm", optional: true]}], "hexpm", "7ae90ad27b09091266f6adbb61e1d2516a7c3d7062c6789d46a7554ec40f3a56"},
+  "plug": {:hex, :plug, "1.13.6", "187beb6b67c6cec50503e940f0434ea4692b19384d47e5fdfd701e93cadb4cc2", [:mix], [{:mime, "~> 1.0 or ~> 2.0", [hex: :mime, repo: "hexpm", optional: false]}, {:plug_crypto, "~> 1.1.1 or ~> 1.2", [hex: :plug_crypto, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4.3 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "02b9c6b9955bce92c829f31d6284bf53c591ca63c4fb9ff81dfd0418667a34ff"},
+  "plug_crypto": {:hex, :plug_crypto, "1.2.2", "05654514ac717ff3a1843204b424477d9e60c143406aa94daf2274fdd280794d", [:mix], [], "hexpm", "87631c7ad914a5a445f0a3809f99b079113ae4ed4b867348dd9eec288cecb6db"},
   "telemetry": {:hex, :telemetry, "0.4.3", "a06428a514bdbc63293cd9a6263aad00ddeb66f608163bdec7c8995784080818", [:rebar3], [], "hexpm", "eb72b8365ffda5bed68a620d1da88525e326cb82a75ee61354fc24b844768041"},
 }

test/absinthe/phoenix_test.exs

@@ -100,7 +100,7 @@ defmodule Absinthe.PhoenixTest do
     assert_push("subscription:data", push)
 
     expected = %{
-      result: %{data: %{"commentAdded" => %{"contents" => "hello world"}}},
+      result: %{data: %{"commentAdded" => %{"contents" => "hello world"}}, ordinal: nil},
       subscriptionId: subscription_ref
     }
 
@@ -145,6 +145,27 @@ defmodule Absinthe.PhoenixTest do
     assert expected == push
   end
 
+  test "subscription with ordinal", %{socket: socket} do
+    ref = push socket, "doc", %{"query" => "subscription { ordinal }"}
+
+    assert_reply ref, :ok, %{subscriptionId: subscription_ref}
+
+    Absinthe.Subscription.publish(TestEndpoint, 1, ordinal: "ordinal_topic")
+
+    assert_push "subscription:data", push
+    expected = %{result: %{data: %{"ordinal" => 1}, ordinal: 1}, subscriptionId: subscription_ref}
+    assert expected == push
+
+    Absinthe.Subscription.publish(TestEndpoint, 0, ordinal: "ordinal_topic")
+    # This message should not generate a notification because it has a lower ordinal
+
+    Absinthe.Subscription.publish(TestEndpoint, 2, ordinal: "ordinal_topic")
+
+    assert_push "subscription:data", push
+    expected = %{result: %{data: %{"ordinal" => 2}, ordinal: 2}, subscriptionId: subscription_ref}
+    assert expected == push
+  end
+
   test "context changes are persisted across documents", %{socket: socket} do
     ref =
       push(socket, "doc", %{

test/support/schema.ex

@@ -99,11 +99,17 @@ defmodule Schema do
     end
 
     field :prime, :string do
-      config fn _, %{context: %{pubsub: pubsub}} ->
-        {:ok, topic: "prime_topic", prime: fn ->
+      config fn _, _ ->
+        {:ok, topic: "prime_topic", prime: fn _ ->
           {:ok, ["prime1", "prime2"]}
         end}
       end
     end
+
+    field :ordinal, :integer do
+      config fn _, _ ->
+        {:ok, topic: "ordinal_topic", ordinal: fn value -> value end}
+      end
+    end
   end
 end