Repository: phoenixframework/phoenix
Pull request #5280 (2023-02-16 11:27:33)

Fix typespec for Phoenix.put_new_layout
Relates to: https://github.com/phoenixframework/phoenix/issues/5278#issuecomment-1432896755

```elixir
@spec put_new_layout(Plug.Conn.t(), [{format :: atom, layout}] | layout) :: Plug.Conn.t()
  def put_new_layout(%Plug.Conn{state: state} = conn, layout)
      when (is_tuple(layout) and tuple_size(layout) == 2) or is_list(layout) or layout == false do
```

The typespec is now in sync with the guards and allows to pass a `layout()` without a `format`.

# Reason
The problem starts when [default layouts](https://github.com/phoenixframework/phoenix/blob/v1.7.0-rc.3/lib/phoenix/controller.ex#L230) are set in the Phoenix.Controller and the error case of [__view__](https://github.com/phoenixframework/phoenix/blob/v1.7.0-rc.3/lib/phoenix/controller.ex#L1823)
and [__layout__ ](https://github.com/phoenixframework/phoenix/blob/v1.7.0-rc.3/lib/phoenix/controller.ex#L1860), don't return a `format`. This makes the return incompatible with new layout definition.

I guess all those function will be removed very soon.




Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index aa8b910f0..c99776178 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -35,6 +35,11 @@ This feature replaces the `Helpers` module generated in your Phoenix router, but
 will continue to work and be generated. You can disable router helpers by passing the
 `helpers: false` option to `use Phoenix.Router`.
 
+## 1.7.0-rc.4
+
+### Bug Fixes
+  * Fix race conditions in the longpoll trasnport by batching messages
+
 ## 1.7.0-rc.3 (2023-02-15)
 
 ### Enhancements
diff --git a/assets/js/phoenix/longpoll.js b/assets/js/phoenix/longpoll.js
index 18c15a2fc..ff3c57a9f 100644
--- a/assets/js/phoenix/longpoll.js
+++ b/assets/js/phoenix/longpoll.js
@@ -12,6 +12,10 @@ export default class LongPoll {
     this.token = null
     this.skipHeartbeat = true
     this.reqs = new Set()
+    this.awaitingBatchAck = false
+    this.currentBatch = null
+    this.currentBatchTimer = null
+    this.batchBuffer = []
     this.onopen = function (){ } // noop
     this.onerror = function (){ } // noop
     this.onmessage = function (){ } // noop
@@ -45,7 +49,7 @@ export default class LongPoll {
   isActive(){ return this.readyState === SOCKET_STATES.open || this.readyState === SOCKET_STATES.connecting }
 
   poll(){
-    this.ajax("GET", null, () => this.ontimeout(), resp => {
+    this.ajax("GET", "application/json", null, () => this.ontimeout(), resp => {
       if(resp){
         var {status, token, messages} = resp
         this.token = token
@@ -100,11 +104,33 @@ export default class LongPoll {
     })
   }
 
+  // we collect all pushes within the current event loop by
+  // setTimeout 0, which optimizes back-to-back procedural
+  // pushes against an empty buffer
   send(body){
-    this.ajax("POST", body, () => this.onerror("timeout"), resp => {
+    if(this.currentBatch){
+      this.currentBatch.push(body)
+    } else if(this.awaitingBatchAck){
+      this.batchBuffer.push(body)
+    } else {
+      this.currentBatch = [body]
+      this.currentBatchTimer = setTimeout(() => {
+        this.batchSend(this.currentBatch)
+        this.currentBatch = null
+      }, 0)
+    }
+  }
+
+  batchSend(messages){
+    this.awaitingBatchAck = true
+    this.ajax("POST", "application/x-ndjson", messages.join("\n"), () => this.onerror("timeout"), resp => {
+      this.awaitingBatchAck = false
       if(!resp || resp.status !== 200){
         this.onerror(resp && resp.status)
         this.closeAndRetry(1011, "internal server error", false)
+      } else if(this.batchBuffer.length > 0){
+        this.batchSend(this.batchBuffer)
+        this.batchBuffer = []
       }
     })
   }
@@ -113,6 +139,9 @@ export default class LongPoll {
     for(let req of this.reqs){ req.abort() }
     this.readyState = SOCKET_STATES.closed
     let opts = Object.assign({code: 1000, reason: undefined, wasClean: true}, {code, reason, wasClean})
+    this.batchBuffer = []
+    clearTimeout(this.currentBatchTimer)
+    this.currentBatchTimer = null
     if(typeof(CloseEvent) !== "undefined"){
       this.onclose(new CloseEvent("close", opts))
     } else {
@@ -120,13 +149,13 @@ export default class LongPoll {
     }
   }
 
-  ajax(method, body, onCallerTimeout, callback){
+  ajax(method, contentType, body, onCallerTimeout, callback){
     let req
     let ontimeout = () => {
       this.reqs.delete(req)
       onCallerTimeout()
     }
-    req = Ajax.request(method, this.endpointURL(), "application/json", body, this.timeout, ontimeout, resp => {
+    req = Ajax.request(method, this.endpointURL(), contentType, body, this.timeout, ontimeout, resp => {
       this.reqs.delete(req)
       if(this.isActive()){ callback(resp) }
     })
diff --git a/guides/components.md b/guides/components.md
index 31e1ce95a..af07635bc 100644
--- a/guides/components.md
+++ b/guides/components.md
@@ -34,7 +34,7 @@ That's simple enough. There's only two lines, `use HelloWeb, :html`. This line c
 
 All of the imports and aliases we make in our module will also be available in our templates. That's because templates are effectively compiled into functions inside their respective module. For example, if you define a function in your module, you will be able to invoke it directly from the template. Let's see this in practice.
 
-Imagine we want to refactor our `show.html.heex` to move the rendering of `<h2>Hello World, from <%= @messenger %>!</h2>` to its own function. We can move it to a function component inside `HelloHTML`:
+Imagine we want to refactor our `show.html.heex` to move the rendering of `<h2>Hello World, from <%= @messenger %>!</h2>` to its own function. We can move it to a function component inside `HelloHTML`, let's do so:
 
 ```elixir
 defmodule HelloWeb.HelloHTML do
@@ -52,7 +52,7 @@ defmodule HelloWeb.HelloHTML do
 end
 ```
 
-We declared the attributes we accept via `attr` provided by `Phoenix.Component`, then we defined our `title/1` function which returns the HEEx template. When we reload our home page, we should see our new title. Since templates are compiled inside the view, we can invoke the view function simply as `<.greet messenger="..." />`, but we can also type `<HelloWeb.HelloHTML.greet messenger="..." />` if the component was defined elsewhere.
+In the example above, we defined a `greet/1` function which returns the HEEx template. Above the function, we called `attr`, provided by `Phoenix.Component`, which defines the attributes/assigns that function expects. Since templates are embedded inside the `HelloHTML` module, we can invoke the our component simply as `<.greet messenger="..." />`, but we can also type `<HelloWeb.HelloHTML.greet messenger="..." />` if the component was defined elsewhere.
 
 By declaring attributes, Phoenix will warn if we call the `<.greet />` component without passing attributes. If an attribute is optional, you can specify the `:default` option with a value:
 
@@ -183,13 +183,13 @@ In other words, after rendering your page, the result is placed in the `@inner_c
 
 Phoenix provides all kinds of conveniences to control which layout should be rendered. For example, the `Phoenix.Controller` module provides the `put_root_layout/2` function for us to switch _root layouts_. This takes `conn` as its first argument and a keyword list of formats and their layouts. You can set it to `false` to disable the layout altogether.
 
-You can edit the `index` action of `PageController` in `lib/hello_web/controllers/page_controller.ex` to look like this.
+You can edit the `home` action of `PageController` in `lib/hello_web/controllers/page_controller.ex` to look like this.
 
 ```elixir
-def index(conn, _params) do
+def home(conn, _params) do
   conn
   |> put_root_layout(html: false)
-  |> render(:index)
+  |> render(:home)
 end
 ```
 
@@ -197,13 +197,13 @@ After reloading [http://localhost:4000/](http://localhost:4000/), we should see
 
 To customize the application layout, we invoke a similar function named `put_layout/2`. Let's actually create another layout and render the index template into it. As an example, let's say we had a different layout for the admin section of our application which didn't have the logo image. To do this, copy the existing `app.html.heex` to a new file `admin.html.heex` in the same directory `lib/hello_web/components/layouts`. Then remove everything inside the `<header>...</header>` tags (or change it to whatever you desire) in the new file.
 
-Now, in the `index` action of the controller of `lib/hello_web/controllers/page_controller.ex`, add the following:
+Now, in the `home` action of the controller of `lib/hello_web/controllers/page_controller.ex`, add the following:
 
 ```elixir
-def index(conn, _params) do
+def home(conn, _params) do
   conn
   |> put_layout(html: :admin)
-  |> render(:index)
+  |> render(:home)
 end
 ```
 
diff --git a/guides/contexts.md b/guides/contexts.md
index 765b7f666..50c0cb49b 100644
--- a/guides/contexts.md
+++ b/guides/contexts.md
@@ -493,6 +493,8 @@ Next, let's expose our new feature to the web by adding the category input to ou
 defmodule HelloWeb.ProductHTML do
   use HelloWeb, :html
 
+  import Phoenix.HTML.Form
+
   def category_select(f, changeset) do
     existing_ids =
       changeset
@@ -508,7 +510,7 @@ defmodule HelloWeb.ProductHTML do
 end
 ```
 
-We added a new `category_select/2` function which uses `Phoenix.HTML`'s `multiple_select/3` to generate a multiple select tag. We calculated the existing category IDs from our changeset, then used those values when we generate the select options for the input tag. We did this by enumerating over all of our categories and returning the appropriate `key`, `value`, and `selected` values. We marked an option as selected if the category ID was found in those category IDs in our changeset.
+We added a new `category_select/2` function which uses `Phoenix.HTML.Form`'s `multiple_select/3` to generate a multiple select tag. We calculated the existing category IDs from our changeset, then used those values when we generate the select options for the input tag. We did this by enumerating over all of our categories and returning the appropriate `key`, `value`, and `selected` values. We marked an option as selected if the category ID was found in those category IDs in our changeset.
 
 With our `category_select` function in place, we can open up `lib/hello_web/controllers/product_html/form.html.heex` and add:
 
diff --git a/installer/lib/phx_new/generator.ex b/installer/lib/phx_new/generator.ex
index 015f8c52e..64f201a7a 100644
--- a/installer/lib/phx_new/generator.ex
+++ b/installer/lib/phx_new/generator.ex
@@ -356,7 +356,8 @@ defmodule Phx.New.Generator do
           For example: ecto://USER:PASS@HOST/DATABASE
           \"""
 
-      maybe_ipv6 = if System.get_env("ECTO_IPV6"), do: [:inet6], else: []
+      maybe_ipv6 = if System.get_env("ECTO_IPV6") in ~w(true 1), do: [:inet6], else: []
+
       """,
       prod_config: """
       # ssl: true,
diff --git a/lib/mix/tasks/phx.gen.release.ex b/lib/mix/tasks/phx.gen.release.ex
index e0cf66944..dc97c567e 100644
--- a/lib/mix/tasks/phx.gen.release.ex
+++ b/lib/mix/tasks/phx.gen.release.ex
@@ -113,7 +113,7 @@ defmodule Mix.Tasks.Phx.Gen.Release do
 
       Add the following to your config/runtime.exs:
 
-          maybe_ipv6 = if System.get_env("ECTO_IPV6"), do: [:inet6], else: []
+          maybe_ipv6 = if System.get_env("ECTO_IPV6") in ~w(true 1), do: [:inet6], else: []
 
           config :#{app}, #{app_namespace}.Repo,
             ...,
diff --git a/lib/phoenix/controller.ex b/lib/phoenix/controller.ex
index 802a2c915..2b15e8e44 100644
--- a/lib/phoenix/controller.ex
+++ b/lib/phoenix/controller.ex
@@ -690,7 +690,7 @@ defmodule Phoenix.Controller do
 
   Raises `Plug.Conn.AlreadySentError` if `conn` is already sent.
   """
-  @spec put_new_layout(Plug.Conn.t(), [{format :: atom, layout}]) :: Plug.Conn.t()
+  @spec put_new_layout(Plug.Conn.t(), [{format :: atom, layout}] | layout) :: Plug.Conn.t()
   def put_new_layout(%Plug.Conn{state: state} = conn, layout)
       when (is_tuple(layout) and tuple_size(layout) == 2) or is_list(layout) or layout == false do
     unless state in @unsent, do: raise(AlreadySentError)
diff --git a/lib/phoenix/socket.ex b/lib/phoenix/socket.ex
index 796a7cf74..7de5bfa47 100644
--- a/lib/phoenix/socket.ex
+++ b/lib/phoenix/socket.ex
@@ -636,9 +636,24 @@ defmodule Phoenix.Socket do
     handle_in(nil, message, new_state, new_socket)
   end
 
+  defp handle_in({pid, _ref, _status}, %{event: "phx_leave"} = msg, state, socket) do
+    %{topic: topic, join_ref: join_ref} = msg
+
+    case state.channels_inverse do
+      # we need to match on nil to handle v1 protocol
+      %{^pid => {^topic, existing_join_ref}} when existing_join_ref in [join_ref, nil] ->
+        send(pid, msg)
+        {:ok, {update_channel_status(state, pid, topic, :leaving), socket}}
+
+      # the client has raced a server close. No need to reply since we already sent close
+      %{^pid => {^topic, _old_join_ref}} ->
+        {:ok, {state, socket}}
+    end
+  end
+
   defp handle_in({pid, _ref, _status}, message, state, socket) do
     send(pid, message)
-    {:ok, {maybe_put_status(state, pid, message), socket}}
+    {:ok, {state, socket}}
   end
 
   defp handle_in(nil, %{event: "phx_leave", ref: ref, topic: topic, join_ref: join_ref}, state, socket) do
@@ -726,14 +741,6 @@ defmodule Phoenix.Socket do
     end
   end
 
-  defp maybe_put_status(state, pid, %{event: "phx_leave", topic: topic}) do
-    update_channel_status(state, pid, topic, :leaving)
-  end
-
-  defp maybe_put_status(state, _pid, %{} = _msg) do
-    state
-  end
-
   defp update_channel_status(state, pid, topic, status) do
     new_channels = Map.update!(state.channels, topic, fn {^pid, ref, _} -> {pid, ref, status} end)
     %{state | channels: new_channels}
diff --git a/lib/phoenix/transports/long_poll.ex b/lib/phoenix/transports/long_poll.ex
index 135df8137..6628d6bbf 100644
--- a/lib/phoenix/transports/long_poll.ex
+++ b/lib/phoenix/transports/long_poll.ex
@@ -72,7 +72,21 @@ defmodule Phoenix.Transports.LongPoll do
   defp publish(conn, server_ref, endpoint, opts) do
     case read_body(conn, []) do
       {:ok, body, conn} ->
-        status = transport_dispatch(endpoint, server_ref, body, opts)
+        # we need to match on both v1 and v2 protocol, as well as wrap for backwards compat
+        batch =
+          case get_req_header(conn, "content-type") do
+            ["application/x-ndjson"] -> String.split(body, ["\n", "\r\n"])
+            _ -> [body]
+          end
+
+        {conn, status} =
+          Enum.reduce_while(batch, {conn, nil}, fn msg, {conn, _status} ->
+            case transport_dispatch(endpoint, server_ref, msg, opts) do
+              :ok -> {:cont, {conn, :ok}}
+              :request_timeout = timeout -> {:halt, {conn, timeout}}
+            end
+          end)
+
         conn |> put_status(status) |> status_json()
 
       _ ->
@@ -168,8 +182,12 @@ defmodule Phoenix.Transports.LongPoll do
 
   ## Helpers
 
-  defp server_ref(endpoint_id, id, pid, topic) do
-    if endpoint_id == id and Process.alive?(pid), do: pid, else: topic
+  defp server_ref(endpoint_id, id, pid, topic) when is_pid(pid) do
+    cond do
+      node(pid) in Node.list() -> pid
+      endpoint_id == id and Process.alive?(pid) -> pid
+      true -> topic
+    end
   end
 
   defp client_ref(topic) when is_binary(topic), do: topic
diff --git a/priv/templates/phx.gen.auth/auth.ex b/priv/templates/phx.gen.auth/auth.ex
index 6dcf8ff68..9aa6a753e 100644
--- a/priv/templates/phx.gen.auth/auth.ex
+++ b/priv/templates/phx.gen.auth/auth.ex
@@ -173,15 +173,11 @@ defmodule <%= inspect auth_module %> do
   end
 
   defp mount_current_<%= schema.singular %>(session, socket) do
-    case session do
-      %{"<%= schema.singular %>_token" => <%= schema.singular %>_token} ->
-        Phoenix.Component.assign_new(socket, :current_<%= schema.singular %>, fn ->
-          <%= inspect context.alias %>.get_<%= schema.singular %>_by_session_token(<%= schema.singular %>_token)
-        end)
-
-      %{} ->
-        Phoenix.Component.assign_new(socket, :current_<%= schema.singular %>, fn -> nil end)
-    end
+    Phoenix.Component.assign_new(socket, :current_<%= schema.singular %>, fn ->
+      if <%= schema.singular %>_token = session["<%= schema.singular %>_token"] do
+        <%= inspect context.alias %>.get_<%= schema.singular %>_by_session_token(<%= schema.singular %>_token)
+      end
+    end)
   end
 
   @doc """
diff --git a/priv/templates/phx.gen.auth/schema.ex b/priv/templates/phx.gen.auth/schema.ex
index 81a33e83a..2624e96f3 100644
--- a/priv/templates/phx.gen.auth/schema.ex
+++ b/priv/templates/phx.gen.auth/schema.ex
@@ -54,6 +54,7 @@ defmodule <%= inspect schema.module %> do
     changeset
     |> validate_required([:password])
     |> validate_length(:password, min: 12, max: 72)
+    # Examples of additional password validation:
     # |> validate_format(:password, ~r/[a-z]/, message: "at least one lower case character")
     # |> validate_format(:password, ~r/[A-Z]/, message: "at least one upper case character")
     # |> validate_format(:password, ~r/[!?@#$%^&*_0-9]/, message: "at least one digit or punctuation character")
@@ -68,6 +69,8 @@ defmodule <%= inspect schema.module %> do
       changeset<%= if hashing_library.name == :bcrypt do %>
       # If using Bcrypt, then further validate it is at most 72 bytes long
       |> validate_length(:password, max: 72, count: :bytes)<% end %>
+      # Hashing could be done with `Ecto.Changeset.prepare_changes/2`, but that
+      # would keep the database transaction open longer and hurt performance.
       |> put_change(:hashed_password, <%= inspect hashing_library.module %>.hash_pwd_salt(password))
       |> delete_change(:password)
     else


Test patch:
diff --git a/installer/test/phx_new_test.exs b/installer/test/phx_new_test.exs
index f73d11fe9..9600d3320 100644
--- a/installer/test/phx_new_test.exs
+++ b/installer/test/phx_new_test.exs
@@ -178,7 +178,10 @@ defmodule Mix.Tasks.Phx.NewTest do
 
       assert_file("phx_blog/config/runtime.exs", fn file ->
         assert file =~ config
-        assert file =~ ~S|maybe_ipv6 = if System.get_env("ECTO_IPV6"), do: [:inet6], else: []|
+
+        assert file =~
+                 ~S|maybe_ipv6 = if System.get_env("ECTO_IPV6") in ~w(true 1), do: [:inet6], else: []|
+
         assert file =~ ~S|socket_options: maybe_ipv6|
 
         assert file =~ """
diff --git a/test/phoenix/integration/long_poll_channels_test.exs b/test/phoenix/integration/long_poll_channels_test.exs
index 5c1104460..92e5e35d2 100644
--- a/test/phoenix/integration/long_poll_channels_test.exs
+++ b/test/phoenix/integration/long_poll_channels_test.exs
@@ -158,8 +158,14 @@ defmodule Phoenix.Integration.LongPollChannelsTest do
   Returns a response with body decoded into JSON map.
   """
   def poll(method, path, vsn, params, json \\ nil, headers \\ %{}) do
-    serializer = serializer(vsn)
-    headers = Map.merge(%{"content-type" => "application/json"}, headers)
+    {serializer, json} = serializer(vsn, json)
+    headers =
+      if is_list(json) do
+        Map.merge(%{"content-type" => "application/x-ndjson"}, headers)
+      else
+        Map.merge(%{"content-type" => "application/json"}, headers)
+      end
+
     body = encode(serializer, json)
     query_string = params |> Map.put("vsn", vsn) |> URI.encode_query()
     url = "http://127.0.0.1:#{@port}#{path}/longpoll?" <> query_string
@@ -167,8 +173,14 @@ defmodule Phoenix.Integration.LongPollChannelsTest do
     decode_body(serializer, resp)
   end
 
-  defp serializer("2." <> _), do: V2.JSONSerializer
-  defp serializer(_), do: V1.JSONSerializer
+  defp serializer("2." <> _, json), do: {V2.JSONSerializer, json}
+  defp serializer(_, nil), do: {V1.JSONSerializer, nil}
+  defp serializer(_, batch) when is_list(batch) do
+    {V1.JSONSerializer, for(msg <- batch, do: Map.delete(msg, "join_ref"))}
+  end
+  defp serializer(_, %{} = json) do
+    {V1.JSONSerializer, json}
+  end
 
   defp decode_body(serializer, %{} = resp) do
     resp
@@ -182,13 +194,19 @@ defmodule Phoenix.Integration.LongPollChannelsTest do
 
   defp encode(_vsn, nil), do: ""
 
-  defp encode(V2.JSONSerializer, map) do
+  defp encode(V2.JSONSerializer = serializer, batch) when is_list(batch) do
+    batch
+    |> Enum.map(&encode(serializer, &1))
+    |> Enum.join("\n")
+  end
+
+  defp encode(V2.JSONSerializer, %{} = map) do
     Phoenix.json_library().encode!(
       [map["join_ref"], map["ref"], map["topic"], map["event"], map["payload"]]
     )
   end
 
-  defp encode(V1.JSONSerializer, map), do: Phoenix.json_library().encode!(map)
+  defp encode(V1.JSONSerializer, %{} = map), do: Phoenix.json_library().encode!(map)
 
   @doc """
   Joins a long poll socket.
@@ -199,9 +217,9 @@ defmodule Phoenix.Integration.LongPollChannelsTest do
   process. If the mode is pubsub, the session will use the
   pubsub system.
   """
-  def join(path, topic, vsn, mode \\ :local, payload \\ %{}, params \\ %{}, headers \\ %{})
+  def join(path, topic, vsn, join_ref, mode \\ :local, payload \\ %{}, params \\ %{}, headers \\ %{})
 
-  def join(path, topic, vsn, :local, payload, params, headers) do
+  def join(path, topic, vsn, join_ref, :local, payload, params, headers) do
     resp = poll :get, path, vsn, params, %{}, headers
     assert resp.body["token"]
     assert resp.body["status"] == 410
@@ -212,7 +230,7 @@ defmodule Phoenix.Integration.LongPollChannelsTest do
       "topic" => topic,
       "event" => "phx_join",
       "ref" => "1",
-      "join_ref" => "1",
+      "join_ref" => join_ref,
       "payload" => payload
     }, headers
 
@@ -220,8 +238,8 @@ defmodule Phoenix.Integration.LongPollChannelsTest do
     session
   end
 
-  def join(path, topic, vsn, :pubsub, payload, params, headers) do
-    session = join(path, topic, vsn, :local, payload, params, headers)
+  def join(path, topic, vsn, join_ref, :pubsub, payload, params, headers) do
+    session = join(path, topic, vsn, join_ref, :local, payload, params, headers)
 
     {:ok, {:v1, _id, pid, topic}} =
       Phoenix.Token.verify(Endpoint, Atom.to_string(__MODULE__), session["token"])
@@ -235,7 +253,7 @@ defmodule Phoenix.Integration.LongPollChannelsTest do
     @vsn "1.0.0"
 
     test "#{@mode}: joins and poll messages" do
-      session = join("/ws", "room:lobby", @vsn, @mode)
+      session = join("/ws", "room:lobby", @vsn, "1", @mode)
 
       # pull messages
       resp = poll(:get, "/ws", @vsn, session)
@@ -272,7 +290,7 @@ defmodule Phoenix.Integration.LongPollChannelsTest do
     end
 
     test "#{@mode}: transport x_headers are extracted to the socket connect_info" do
-      session = join("/ws/connect_info", "room:lobby", @vsn, @mode, %{}, %{}, %{"x-application" => "Phoenix"})
+      session = join("/ws/connect_info", "room:lobby", @vsn, "1", @mode, %{}, %{}, %{"x-application" => "Phoenix"})
 
       # pull messages
       resp = poll(:get, "/ws/connect_info", @vsn, session)
@@ -289,7 +307,7 @@ defmodule Phoenix.Integration.LongPollChannelsTest do
       ctx_headers =
         %{"traceparent" => "00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01",
         "tracestate" => "congo=t61rcWkgMz"}
-      session = join("/ws/connect_info", "room:lobby", @vsn, @mode, %{}, %{}, ctx_headers)
+      session = join("/ws/connect_info", "room:lobby", @vsn, "1", @mode, %{}, %{}, ctx_headers)
 
       # pull messages
       resp = poll(:get, "/ws/connect_info", @vsn, session)
@@ -303,7 +321,7 @@ defmodule Phoenix.Integration.LongPollChannelsTest do
     end
 
     test "#{@mode}: transport peer_data is extracted to the socket connect_info" do
-      session = join("/ws/connect_info", "room:lobby", @vsn, @mode, %{}, %{}, %{"x-application" => "Phoenix"})
+      session = join("/ws/connect_info", "room:lobby", @vsn, "1", @mode, %{}, %{}, %{"x-application" => "Phoenix"})
 
       # pull messages
       resp = poll(:get, "/ws/connect_info", @vsn, session)
@@ -317,7 +335,7 @@ defmodule Phoenix.Integration.LongPollChannelsTest do
     end
 
     test "#{@mode}: transport uri is extracted to the socket connect_info" do
-      session = join("/ws/connect_info", "room:lobby", @vsn, @mode, %{}, %{}, %{"x-application" => "Phoenix"})
+      session = join("/ws/connect_info", "room:lobby", @vsn, "1", @mode, %{}, %{}, %{"x-application" => "Phoenix"})
 
       # pull messages
       resp = poll(:get, "/ws/connect_info", @vsn, session)
@@ -335,7 +353,7 @@ defmodule Phoenix.Integration.LongPollChannelsTest do
 
     test "#{@mode}: publishing events" do
       Phoenix.PubSub.subscribe(__MODULE__, "room:lobby")
-      session = join("/ws", "room:lobby", @vsn, @mode)
+      session = join("/ws", "room:lobby", @vsn, "1", @mode)
 
       # Publish successfully
       resp = poll :post, "/ws", @vsn, session, %{
@@ -383,8 +401,61 @@ defmodule Phoenix.Integration.LongPollChannelsTest do
       end
     end
 
+    test "#{@mode}: lonpoll publishing batch events on v2 protocol" do
+      vsn = "2.0.0"
+      Phoenix.PubSub.subscribe(__MODULE__, "room:lobby")
+      session = join("/ws", "room:lobby", vsn, "1", @mode)
+      # Publish successfully
+      resp =
+        poll(:post, "/ws", vsn, session, [
+          %{
+            "topic" => "room:lobby",
+            "event" => "new_msg",
+            "ref" => "2",
+            "join_ref" => "1",
+            "payload" => %{"body" => "hi1"}
+          },
+          %{
+            "topic" => "room:lobby",
+            "event" => "new_msg",
+            "ref" => "3",
+            "join_ref" => "1",
+            "payload" => %{"body" => "hi2"}
+          }
+        ])
+
+
+      assert resp.body["status"] == 200
+      assert_receive %Broadcast{event: "new_msg", payload: %{"body" => "hi1"}}
+      assert_receive %Broadcast{event: "new_msg", payload: %{"body" => "hi2"}}
+
+      # Get published message
+      resp = poll(:get, "/ws", vsn, session)
+      assert resp.body["status"] == 200
+
+      assert [
+               _phx_reply,
+               _user_entered,
+               _joined,
+               %Message{
+                 topic: "room:lobby",
+                 event: "new_msg",
+                 payload: %{"body" => "hi1", "transport" => ":longpoll"},
+                 ref: nil,
+                 join_ref: "1"
+               },
+               %Message{
+                 topic: "room:lobby",
+                 event: "new_msg",
+                 payload: %{"body" => "hi2", "transport" => ":longpoll"},
+                 ref: nil,
+                 join_ref: "1"
+               }
+             ] = resp.body["messages"]
+    end
+
     test "#{@mode}: shuts down after timeout" do
-      session = join("/ws", "room:lobby", @vsn, @mode)
+      session = join("/ws", "room:lobby", @vsn, "1", @mode)
 
       channel = Process.whereis(:"room:lobby")
       assert channel
@@ -421,13 +492,13 @@ defmodule Phoenix.Integration.LongPollChannelsTest do
 
       test "filter params on join" do
         log = capture_log fn ->
-          join("/ws", "room:lobby", @vsn, :local, %{"foo" => "bar", "password" => "shouldnotshow"}, %{"logging" => "enabled"})
+          join("/ws", "room:lobby", @vsn, @join_ref, :local, %{"foo" => "bar", "password" => "shouldnotshow"}, %{"logging" => "enabled"})
         end
         assert log =~ "Parameters: %{\"foo\" => \"bar\", \"password\" => \"[FILTERED]\"}"
       end
 
       test "sends phx_error if a channel server abnormally exits", %{topic: topic} do
-        session = join("/ws", topic, @vsn)
+        session = join("/ws", topic, @vsn, @join_ref)
 
         capture_log fn ->
           resp = poll :post, "/ws", @vsn, session, %{
@@ -449,20 +520,23 @@ defmodule Phoenix.Integration.LongPollChannelsTest do
           event: "phx_error",
           payload: %{},
           topic: topic,
-          ref: "1",
+          ref: @join_ref,
           join_ref: @join_ref
         }
       end
 
       test "sends phx_close if a channel server normally exits" do
-        session = join("/ws", "room:lobby", @vsn)
+        session = join("/ws", "room:lobby", @vsn, @join_ref)
+
+        resp =
+          poll :post, "/ws", @vsn, session, %{
+            "topic" => "room:lobby",
+            "event" => "phx_leave",
+            "join_ref" => @join_ref,
+            "ref" => "2",
+            "payload" => %{}
+          }
 
-        resp = poll :post, "/ws", @vsn, session, %{
-          "topic" => "room:lobby",
-          "event" => "phx_leave",
-          "ref" => "2",
-          "payload" => %{}
-        }
         assert resp.body["status"] == 200
         assert resp.status == 200
 
@@ -472,7 +546,7 @@ defmodule Phoenix.Integration.LongPollChannelsTest do
         assert phx_close == %Message{
           event: "phx_close",
           payload: %{},
-          ref: "1",
+          ref: @join_ref,
           join_ref: @join_ref,
           topic: "room:lobby"
         }
@@ -516,12 +590,13 @@ defmodule Phoenix.Integration.LongPollChannelsTest do
       end
 
       test "forces application/json content-type" do
-        session = join("/ws", "room:lobby", @vsn)
+        session = join("/ws", "room:lobby", @vsn, @join_ref)
 
         resp = poll :post, "/ws", @vsn, session, %{
           "topic" => "room:lobby",
           "event" => "phx_leave",
           "ref" => "2",
+          "join_ref" => @join_ref,
           "payload" => %{}
         }, %{"content-type" => ""}
         assert resp.body["status"] == 200