Repository: phoenixframework/phoenix
Pull request #4030 (2020-10-16 20:10:08)

Update directory_structure.md
Fix a typo


Patch:
diff --git a/assets/js/phoenix.js b/assets/js/phoenix.js
index 9e26ac5d7..3dd544962 100644
--- a/assets/js/phoenix.js
+++ b/assets/js/phoenix.js
@@ -51,6 +51,13 @@
  * Successful joins receive an "ok" status, while unsuccessful joins
  * receive "error".
  *
+ * With the default serializers and WebSocket transport, JSON text frames are
+ * used for pushing a JSON object literal. If an `ArrayBuffer` instance is provided,
+ * binary encoding will be used and the message will be sent with the binary
+ * opcode.
+ *
+ * *Note*: binary messages are only supported on the WebSocket transport.
+ *
  * ## Duplicate Join Subscriptions
  *
  * While the client may join any number of topics on any number of channels,
@@ -662,17 +669,110 @@ export class Channel {
 
 /* The default serializer for encoding and decoding messages */
 export let Serializer = {
+  HEADER_LENGTH: 1,
+  META_LENGTH: 4,
+  KINDS: {push: 0, reply: 1, broadcast: 2},
+
   encode(msg, callback){
-    let payload = [
-      msg.join_ref, msg.ref, msg.topic, msg.event, msg.payload
-    ]
-    return callback(JSON.stringify(payload))
+    if(msg.payload.constructor === ArrayBuffer){
+      return callback(this.binaryEncode(msg))
+    } else {
+      let payload = [msg.join_ref, msg.ref, msg.topic, msg.event, msg.payload]
+      return callback(JSON.stringify(payload))
+    }
   },
 
   decode(rawPayload, callback){
-    let [join_ref, ref, topic, event, payload] = JSON.parse(rawPayload)
+    if(rawPayload.constructor === ArrayBuffer){
+      return callback(this.binaryDecode(rawPayload))
+    } else {
+      let [join_ref, ref, topic, event, payload] = JSON.parse(rawPayload)
+      return callback({join_ref, ref, topic, event, payload})
+    }
+  },
+
+  // private
+
+  binaryEncode(message) {
+    let {join_ref, ref, event, topic, payload} = message
+    let metaLength = this.META_LENGTH + join_ref.length + ref.length + topic.length + event.length
+    let header = new ArrayBuffer(this.HEADER_LENGTH + metaLength)
+    let view = new DataView(header)
+    let offset = 0
+
+    view.setUint8(offset++, this.KINDS.push) // kind
+    view.setUint8(offset++, join_ref.length)
+    view.setUint8(offset++, ref.length)
+    view.setUint8(offset++, topic.length)
+    view.setUint8(offset++, event.length)
+    Array.from(join_ref, char => view.setUint8(offset++, char.charCodeAt(0)))
+    Array.from(ref, char => view.setUint8(offset++, char.charCodeAt(0)))
+    Array.from(topic, char => view.setUint8(offset++, char.charCodeAt(0)))
+    Array.from(event, char => view.setUint8(offset++, char.charCodeAt(0)))
+
+    var combined = new Uint8Array(header.byteLength + payload.byteLength)
+    combined.set(new Uint8Array(header), 0)
+    combined.set(new Uint8Array(payload), header.byteLength)
+
+    return combined.buffer
+  },
+
+  binaryDecode(buffer){
+    let view = new DataView(buffer)
+    let kind = view.getUint8(0)
+    let decoder = new TextDecoder()
+    switch(kind){
+      case this.KINDS.push:      return this.decodePush(buffer, view, decoder)
+      case this.KINDS.reply:     return this.decodeReply(buffer, view, decoder)
+      case this.KINDS.broadcast: return this.decodeBroadcast(buffer, view, decoder)
+    }
+  },
+
+  decodePush(buffer, view, decoder){
+    let joinRefSize = view.getUint8(1)
+    let topicSize = view.getUint8(2)
+    let eventSize = view.getUint8(3)
+    let offset = this.HEADER_LENGTH + this.META_LENGTH - 1 // pushes have no ref
+    let joinRef = decoder.decode(buffer.slice(offset, offset + joinRefSize))
+    offset = offset + joinRefSize
+    let topic = decoder.decode(buffer.slice(offset, offset + topicSize))
+    offset = offset + topicSize
+    let event = decoder.decode(buffer.slice(offset, offset + eventSize))
+    offset = offset + eventSize
+    let data = buffer.slice(offset, buffer.byteLength)
+    return {join_ref: joinRef, ref: null, topic: topic, event: event, payload: data}
+  },
+
+  decodeReply(buffer, view, decoder){
+    let joinRefSize = view.getUint8(1)
+    let refSize = view.getUint8(2)
+    let topicSize = view.getUint8(3)
+    let eventSize = view.getUint8(4)
+    let offset = this.HEADER_LENGTH + this.META_LENGTH
+    let joinRef = decoder.decode(buffer.slice(offset, offset + joinRefSize))
+    offset = offset + joinRefSize
+    let ref = decoder.decode(buffer.slice(offset, offset + refSize))
+    offset = offset + refSize
+    let topic = decoder.decode(buffer.slice(offset, offset + topicSize))
+    offset = offset + topicSize
+    let event = decoder.decode(buffer.slice(offset, offset + eventSize))
+    offset = offset + eventSize
+    let data = buffer.slice(offset, buffer.byteLength)
+    let payload = {status: event, response: data}
+    return {join_ref: joinRef, ref: ref, topic: topic, event: CHANNEL_EVENTS.reply, payload: payload}
+  },
+
+  decodeBroadcast(buffer, view, decoder){
+    let topicSize = view.getUint8(1)
+    let eventSize = view.getUint8(2)
+    let offset = this.HEADER_LENGTH + 2
+    let topic = decoder.decode(buffer.slice(offset, offset + topicSize))
+    offset = offset + topicSize
+    let event = decoder.decode(buffer.slice(offset, offset + eventSize))
+    offset = offset + eventSize
+    let data = buffer.slice(offset, buffer.byteLength)
 
-    return callback({join_ref, ref, topic, event, payload})
+    return {join_ref: null, ref: null, topic: topic, event: event, payload: data}
   }
 }
 
@@ -754,8 +854,8 @@ export class Socket {
     this.ref                  = 0
     this.timeout              = opts.timeout || DEFAULT_TIMEOUT
     this.transport            = opts.transport || global.WebSocket || LongPoll
-    this.defaultEncoder       = Serializer.encode
-    this.defaultDecoder       = Serializer.decode
+    this.defaultEncoder       = Serializer.encode.bind(Serializer)
+    this.defaultDecoder       = Serializer.decode.bind(Serializer)
     this.closeWasClean        = false
     this.unloaded             = false
     this.binaryType           = opts.binaryType || "arraybuffer"
@@ -1111,7 +1211,7 @@ export class Socket {
 
   abnormalClose(reason){
     this.closeWasClean = false
-    this.conn.close(WS_CLOSE_NORMAL, reason)
+    if(this.conn.readyState === SOCKET_STATES.open){ this.conn.close(WS_CLOSE_NORMAL, reason) }
   }
 
   flushSendBuffer(){
diff --git a/guides/directory_structure.md b/guides/directory_structure.md
index b348e68b1..8f982f089 100644
--- a/guides/directory_structure.md
+++ b/guides/directory_structure.md
@@ -2,7 +2,7 @@
 
 > **Requirement**: This guide expects that you have gone through the introductory guides and got a Phoenix application up and running.
 
-When we use `mix phx.new` to generates a new Phoenix application, it builds a top-level directory structure like this:
+When we use `mix phx.new` to generate a new Phoenix application, it builds a top-level directory structure like this:
 
 ```console
 ├── _build
diff --git a/lib/phoenix/channel.ex b/lib/phoenix/channel.ex
index a927eaf4e..e9d84fe49 100644
--- a/lib/phoenix/channel.ex
+++ b/lib/phoenix/channel.ex
@@ -57,11 +57,21 @@ defmodule Phoenix.Channel do
         {:noreply, socket}
       end
 
-  You can also push a message directly down the socket:
+  General message payloads are received as maps, and binary data payloads are
+  passed as a `{:binary, data}` tuple:
+
+      def handle_in("file_chunk", {:binary, chunk}, socket) do
+        ...
+        {:reply, :ok, socket}
+      end
+
+  You can also push a message directly down the socket, in the form of a map,
+  or a tagged `{:binary, data}` tuple:
 
       # client asks for their current rank, push sent directly as a new event.
       def handle_in("current_rank", _, socket) do
         push(socket, "current_rank", %{val: Game.get_rank(socket.assigns[:user])})
+        push(socket, "photo", {:binary, File.read!(socket.assigns.photo_path)})
         {:noreply, socket}
       end
 
@@ -100,6 +110,10 @@ defmodule Phoenix.Channel do
         end
       end
 
+  Like binary pushes, binary data is also supported with replies via a `{:binary, data}` tuple:
+
+      {:reply, {:ok, {:binary, bin}}, socket}
+
   ## Intercepting Outgoing Events
 
   When an event is broadcasted with `broadcast/3`, each channel subscriber can
@@ -553,7 +567,7 @@ defmodule Phoenix.Channel do
   """
   def push(socket, event, message) do
     %{transport_pid: transport_pid, topic: topic} = assert_joined!(socket)
-    Server.push(transport_pid, topic, event, message, socket.serializer)
+    Server.push(transport_pid, socket.join_ref, topic, event, message, socket.serializer)
   end
 
   @doc """
diff --git a/lib/phoenix/channel/server.ex b/lib/phoenix/channel/server.ex
index 3c3c5a906..5feabd883 100644
--- a/lib/phoenix/channel/server.ex
+++ b/lib/phoenix/channel/server.ex
@@ -16,10 +16,9 @@ defmodule Phoenix.Channel.Server do
   """
   @spec join(Socket.t(), module, Message.t(), keyword) :: {:ok, term, pid} | {:error, term}
   def join(socket, channel, message, opts) do
-    %{topic: topic, payload: payload, ref: join_ref} = message
+    %{topic: topic, payload: payload, ref: ref, join_ref: join_ref} = message
     assigns = Map.merge(socket.assigns, Keyword.get(opts, :assigns, %{}))
-    socket = %{socket | topic: topic, channel: channel, join_ref: join_ref, assigns: assigns}
-
+    socket = %{socket | topic: topic, channel: channel, join_ref: join_ref || ref, assigns: assigns}
     ref = make_ref()
     from = {self(), ref}
     child_spec = channel.child_spec({socket.endpoint, from})
@@ -136,7 +135,7 @@ defmodule Phoenix.Channel.Server do
   The message is encoded as `Phoenix.Socket.Broadcast`.
   """
   def broadcast(pubsub_server, topic, event, payload)
-      when is_binary(topic) and is_binary(event) and is_map(payload) do
+      when is_binary(topic) and is_binary(event) do
     broadcast = %Broadcast{
       topic: topic,
       event: event,
@@ -153,7 +152,7 @@ defmodule Phoenix.Channel.Server do
   Raises in case of crashes.
   """
   def broadcast!(pubsub_server, topic, event, payload)
-      when is_binary(topic) and is_binary(event) and is_map(payload) do
+      when is_binary(topic) and is_binary(event) do
     broadcast = %Broadcast{
       topic: topic,
       event: event,
@@ -170,7 +169,7 @@ defmodule Phoenix.Channel.Server do
   The message is encoded as `Phoenix.Socket.Broadcast`.
   """
   def broadcast_from(pubsub_server, from, topic, event, payload)
-      when is_binary(topic) and is_binary(event) and is_map(payload) do
+      when is_binary(topic) and is_binary(event) do
     broadcast = %Broadcast{
       topic: topic,
       event: event,
@@ -187,7 +186,7 @@ defmodule Phoenix.Channel.Server do
   Raises in case of crashes.
   """
   def broadcast_from!(pubsub_server, from, topic, event, payload)
-      when is_binary(topic) and is_binary(event) and is_map(payload) do
+      when is_binary(topic) and is_binary(event) do
     broadcast = %Broadcast{
       topic: topic,
       event: event,
@@ -204,7 +203,7 @@ defmodule Phoenix.Channel.Server do
   The message is encoded as `Phoenix.Socket.Broadcast`.
   """
   def local_broadcast(pubsub_server, topic, event, payload)
-      when is_binary(topic) and is_binary(event) and is_map(payload) do
+      when is_binary(topic) and is_binary(event) do
     broadcast = %Broadcast{
       topic: topic,
       event: event,
@@ -221,7 +220,7 @@ defmodule Phoenix.Channel.Server do
   The message is encoded as `Phoenix.Socket.Broadcast`.
   """
   def local_broadcast_from(pubsub_server, from, topic, event, payload)
-      when is_binary(topic) and is_binary(event) and is_map(payload) do
+      when is_binary(topic) and is_binary(event) do
     broadcast = %Broadcast{
       topic: topic,
       event: event,
@@ -235,9 +234,9 @@ defmodule Phoenix.Channel.Server do
   Pushes a message with the given topic, event and payload
   to the given process.
   """
-  def push(pid, topic, event, payload, serializer)
-      when is_binary(topic) and is_binary(event) and is_map(payload) do
-    message = %Message{topic: topic, event: event, payload: payload}
+  def push(pid, join_ref, topic, event, payload, serializer)
+      when is_binary(topic) and is_binary(event) do
+    message = %Message{join_ref: join_ref, topic: topic, event: event, payload: payload}
     send(pid, serializer.encode!(message))
     :ok
   end
@@ -246,7 +245,7 @@ defmodule Phoenix.Channel.Server do
   Replies to a given ref to the transport process.
   """
   def reply(pid, join_ref, ref, topic, {status, payload}, serializer)
-      when is_binary(topic) and is_map(payload) do
+      when is_binary(topic) do
     reply = %Reply{topic: topic, join_ref: join_ref, ref: ref, status: status, payload: payload}
     send(pid, serializer.encode!(reply))
     :ok
@@ -500,7 +499,7 @@ defmodule Phoenix.Channel.Server do
     handle_result(other, :handle_in)
   end
 
-  defp handle_reply(socket, {status, payload}) when is_atom(status) and is_map(payload) do
+  defp handle_reply(socket, {status, payload}) when is_atom(status) do
     reply(
       socket.transport_pid,
       socket.join_ref,
diff --git a/lib/phoenix/socket/serializers/v2_json_serializer.ex b/lib/phoenix/socket/serializers/v2_json_serializer.ex
index 71d052f8a..8727622e7 100644
--- a/lib/phoenix/socket/serializers/v2_json_serializer.ex
+++ b/lib/phoenix/socket/serializers/v2_json_serializer.ex
@@ -2,15 +2,62 @@ defmodule Phoenix.Socket.V2.JSONSerializer do
   @moduledoc false
   @behaviour Phoenix.Socket.Serializer
 
+  @push 0
+  @reply 1
+  @broadcast 2
+
   alias Phoenix.Socket.{Broadcast, Message, Reply}
 
   @impl true
-  def fastlane!(%Broadcast{} = msg) do
+  def fastlane!(%Broadcast{payload: {:binary, data}} = msg) do
+    topic_size = byte_size!(msg.topic, :topic, 255)
+    event_size = byte_size!(msg.event, :event, 255)
+
+    bin = <<
+      @broadcast::size(8),
+      topic_size::size(8),
+      event_size::size(8),
+      msg.topic::binary-size(topic_size),
+      msg.event::binary-size(event_size),
+      data::binary
+    >>
+
+    {:socket_push, :binary, bin}
+  end
+
+  def fastlane!(%Broadcast{payload: %{}} = msg) do
     data = Phoenix.json_library().encode_to_iodata!([nil, nil, msg.topic, msg.event, msg.payload])
     {:socket_push, :text, data}
   end
 
+  def fastlane!(%Broadcast{payload: invalid}) do
+    raise ArgumentError, "expected broadcasted payload to be a map, got: #{inspect(invalid)}"
+  end
+
   @impl true
+  def encode!(%Reply{payload: {:binary, data}} = reply) do
+    status = to_string(reply.status)
+    join_ref_size = byte_size!(reply.join_ref, :join_ref, 255)
+    ref_size = byte_size!(reply.ref, :ref, 255)
+    topic_size = byte_size!(reply.topic, :topic, 255)
+    status_size = byte_size!(status, :status, 255)
+
+    bin = <<
+      @reply::size(8),
+      join_ref_size::size(8),
+      ref_size::size(8),
+      topic_size::size(8),
+      status_size::size(8),
+      reply.join_ref::binary-size(join_ref_size),
+      reply.ref::binary-size(ref_size),
+      reply.topic::binary-size(topic_size),
+      status::binary-size(status_size),
+      data::binary
+    >>
+
+    {:socket_push, :binary, bin}
+  end
+
   def encode!(%Reply{} = reply) do
     data = [
       reply.join_ref,
@@ -23,13 +70,44 @@ defmodule Phoenix.Socket.V2.JSONSerializer do
     {:socket_push, :text, Phoenix.json_library().encode_to_iodata!(data)}
   end
 
-  def encode!(%Message{} = msg) do
+  def encode!(%Message{payload: {:binary, data}} = msg) do
+    join_ref = to_string(msg.join_ref)
+    join_ref_size = byte_size!(join_ref, :join_ref, 255)
+    topic_size = byte_size!(msg.topic, :topic, 255)
+    event_size = byte_size!(msg.event, :event, 255)
+
+    bin = <<
+      @push::size(8),
+      join_ref_size::size(8),
+      topic_size::size(8),
+      event_size::size(8),
+      join_ref::binary-size(join_ref_size),
+      msg.topic::binary-size(topic_size),
+      msg.event::binary-size(event_size),
+      data::binary
+    >>
+
+    {:socket_push, :binary, bin}
+  end
+
+  def encode!(%Message{payload: %{}} = msg) do
     data = [msg.join_ref, msg.ref, msg.topic, msg.event, msg.payload]
     {:socket_push, :text, Phoenix.json_library().encode_to_iodata!(data)}
   end
 
+  def encode!(%Message{payload: invalid}) do
+    raise ArgumentError, "expected payload to be a map, got: #{inspect(invalid)}"
+  end
+
   @impl true
-  def decode!(raw_message, _opts) do
+  def decode!(raw_message, opts) do
+    case Keyword.fetch(opts, :opcode) do
+      {:ok, :text} -> decode_text(raw_message)
+      {:ok, :binary} -> decode_binary(raw_message)
+    end
+  end
+
+  defp decode_text(raw_message) do
     [join_ref, ref, topic, event, payload | _] = Phoenix.json_library().decode!(raw_message)
 
     %Message{
@@ -40,4 +118,41 @@ defmodule Phoenix.Socket.V2.JSONSerializer do
       join_ref: join_ref
     }
   end
+
+  defp decode_binary(<<
+         @push::size(8),
+         join_ref_size::size(8),
+         ref_size::size(8),
+         topic_size::size(8),
+         event_size::size(8),
+         join_ref::binary-size(join_ref_size),
+         ref::binary-size(ref_size),
+         topic::binary-size(topic_size),
+         event::binary-size(event_size),
+         data::binary
+       >>) do
+    %Message{
+      topic: topic,
+      event: event,
+      payload: {:binary, data},
+      ref: ref,
+      join_ref: join_ref
+    }
+  end
+
+  defp byte_size!(bin, kind, max) do
+    case byte_size(bin) do
+      size when size <= max ->
+        size
+
+      oversized ->
+        raise ArgumentError, """
+        unable to convert #{kind} to binary.
+
+            #{inspect(bin)}
+
+        must be less than or equal to #{max} bytes, but is #{oversized} bytes.
+        """
+    end
+  end
 end


Test patch:
diff --git a/assets/test/serializer_test.js b/assets/test/serializer_test.js
new file mode 100644
index 000000000..e82a25b3b
--- /dev/null
+++ b/assets/test/serializer_test.js
@@ -0,0 +1,96 @@
+import assert from "assert"
+
+import {Serializer} from "../js/phoenix"
+
+let exampleMsg = {join_ref: "0", ref: "1", topic: "t", event: "e", payload: {foo: 1}}
+
+let binPayload = () => {
+  let buffer = new ArrayBuffer(1)
+  new DataView(buffer).setUint8(0, 1)
+  return buffer
+}
+
+describe("JSON", () => {
+  it("encodes general pushes", done => {
+    Serializer.encode(exampleMsg, result => {
+      assert.equal(result, `["0","1","t","e",{"foo":1}]`)
+      done()
+    })
+  })
+
+  it("decodes", done => {
+    Serializer.decode(`["0","1","t","e",{"foo":1}]`, result => {
+      assert.deepEqual(result, exampleMsg)
+      done()
+    })
+  })
+})
+
+describe("binary", () => {
+  it("encodes", done => {
+    let buffer = binPayload()
+    let bin = "\0\x01\x01\x01\x0101te\x01"
+    let decoder = new TextDecoder()
+    Serializer.encode({join_ref: "0", ref: "1", topic: "t", event: "e", payload: buffer}, result => {
+      assert.equal(decoder.decode(result), bin)
+      done()
+    })
+  })
+
+  it("encodes variable length segments", done => {
+    let buffer = binPayload()
+    let bin = "\0\x02\x01\x03\x02101topev\x01"
+    let decoder = new TextDecoder()
+    Serializer.encode({join_ref: "10", ref: "1", topic: "top", event: "ev", payload: buffer}, result => {
+      assert.equal(decoder.decode(result), bin)
+      done()
+    })
+  })
+
+  it("decodes push", done => {
+    let bin = "\0\x03\x03\n123topsome-event\x01\x01"
+    let buffer = (new TextEncoder()).encode(bin).buffer
+    let decoder = new TextDecoder()
+    Serializer.decode(buffer, result => {
+      assert.equal(result.join_ref, "123")
+      assert.equal(result.ref, null)
+      assert.equal(result.topic, "top")
+      assert.equal(result.event, "some-event")
+      assert.equal(result.payload.constructor, ArrayBuffer)
+      assert.equal(decoder.decode(result.payload), "\x01\x01")
+      done()
+    })
+  })
+
+  it("decodes reply", done => {
+    let bin = "\x01\x03\x02\x03\x0210012topok\x01\x01"
+    let buffer = (new TextEncoder()).encode(bin).buffer
+    let decoder = new TextDecoder()
+    Serializer.decode(buffer, result => {
+      assert.equal(result.join_ref, "100")
+      assert.equal(result.ref, "12")
+      assert.equal(result.topic, "top")
+      assert.equal(result.event, "phx_reply")
+      assert.equal(result.payload.status, "ok")
+      assert.equal(result.payload.response.constructor, ArrayBuffer)
+      assert.equal(decoder.decode(result.payload.response), "\x01\x01")
+      done()
+    })
+  })
+
+  it("decodes broadcast", done => {
+    let bin = "\x02\x03\ntopsome-event\x01\x01"
+    let buffer = (new TextEncoder()).encode(bin).buffer
+    let decoder = new TextDecoder()
+    Serializer.decode(buffer, result => {
+      assert.equal(result.join_ref, null)
+      assert.equal(result.ref, null)
+      assert.equal(result.topic, "top")
+      assert.equal(result.event, "some-event")
+      assert.equal(result.payload.constructor, ArrayBuffer)
+      assert.equal(decoder.decode(result.payload), "\x01\x01")
+      done()
+    })
+  })
+})
+
diff --git a/test/phoenix/integration/websocket_channels_test.exs b/test/phoenix/integration/websocket_channels_test.exs
index a2e677074..10c5264c0 100644
--- a/test/phoenix/integration/websocket_channels_test.exs
+++ b/test/phoenix/integration/websocket_channels_test.exs
@@ -50,6 +50,11 @@ defmodule Phoenix.Integration.WebSocketChannelsTest do
       raise "boom"
     end
 
+    def handle_in("binary_event", {:binary, data}, socket) do
+      push(socket, "binary_event", {:binary, <<0, 1>>})
+      {:reply, {:ok, {:binary, <<data::binary, 3, 4>>}}, socket}
+    end
+
     def handle_out("new_msg", payload, socket) do
       push socket, "new_msg", Map.put(payload, "transport", inspect(socket.transport))
       {:noreply, socket}
@@ -208,7 +213,7 @@ defmodule Phoenix.Integration.WebSocketChannelsTest do
 
   @endpoint Endpoint
 
-  for {serializer, vsn, join_ref} <- [{V1.JSONSerializer, "1.0.0", nil}, {V2.JSONSerializer, "2.0.0", 11}] do
+  for {serializer, vsn, join_ref} <- [{V1.JSONSerializer, "1.0.0", nil}, {V2.JSONSerializer, "2.0.0", "11"}] do
     @serializer serializer
     @vsn vsn
     @vsn_path "ws://127.0.0.1:#{@port}/ws/websocket?vsn=#{@vsn}"
@@ -235,9 +240,11 @@ defmodule Phoenix.Integration.WebSocketChannelsTest do
                                 ref: "1", topic: ^lobby}
 
         assert_receive %Message{event: "joined",
+                                join_ref: @join_ref,
                                 payload: %{"status" => "connected", "user_id" => nil}}
         assert_receive %Message{event: "user_entered",
                                 payload: %{"user" => nil},
+                                join_ref: nil,
                                 ref: nil, topic: ^lobby}
 
         channel_pid = Process.whereis(String.to_atom(lobby))
@@ -603,7 +610,7 @@ defmodule Phoenix.Integration.WebSocketChannelsTest do
       WebsocketClient.join(sock, "custom:ignore", %{"action" => "ignore"})
 
       assert_receive %Message{event: "phx_reply",
-                              join_ref: 11,
+                              join_ref: "11",
                               payload: %{"response" => %{"action" => "ignore"}, "status" => "error"},
                               ref: "1",
                               topic: "custom:ignore"}
@@ -611,7 +618,7 @@ defmodule Phoenix.Integration.WebSocketChannelsTest do
       WebsocketClient.join(sock, "custom:error", %{"action" => "error"})
 
       assert_receive %Message{event: "phx_reply",
-                              join_ref: 12,
+                              join_ref: "12",
                               payload: %{"response" => %{"reason" => "join crashed"}, "status" => "error"},
                               ref: "2",
                               topic: "custom:error"}
@@ -619,7 +626,7 @@ defmodule Phoenix.Integration.WebSocketChannelsTest do
       WebsocketClient.join(sock, "custom:ok", %{"action" => "ok"})
 
       assert_receive %Message{event: "phx_reply",
-                              join_ref: 13,
+                              join_ref: "13",
                               payload: %{"response" => %{"action" => "ok"}, "status" => "ok"},
                               ref: "3",
                               topic: "custom:ok"}
@@ -628,4 +635,41 @@ defmodule Phoenix.Integration.WebSocketChannelsTest do
       assert_receive %Message{event: "phx_close", payload: %{}}
     end
   end
+
+  describe "binary" do
+    @serializer V2.JSONSerializer
+    @vsn "2.0.0"
+    @join_ref "11"
+
+    test "messages can be pushed and received" do
+      topic = "room:bin"
+
+      {:ok, socket} =
+        WebsocketClient.start_link(
+          self(),
+          "ws://127.0.0.1:#{@port}/ws/websocket?vsn=#{@vsn}",
+          @serializer
+        )
+
+      WebsocketClient.join(socket, topic, %{})
+
+      assert_receive %Message{
+        event: "phx_reply",
+        payload: %{"response" => %{}, "status" => "ok"},
+        join_ref: @join_ref,
+        ref: "1",
+        topic: ^topic
+      }
+
+      WebsocketClient.send_event(socket, topic, "binary_event", {:binary, <<1, 2>>})
+      assert_receive %Message{
+        event: "phx_reply",
+        payload: %{"response" => {:binary, <<1, 2, 3, 4>>}, "status" => "ok"},
+        join_ref: @join_ref,
+        ref: "2",
+        topic: ^topic
+      }
+      assert_receive %Message{event: "binary_event", join_ref: @join_ref, payload: {:binary, <<0, 1>>}}
+    end
+  end
 end
diff --git a/test/phoenix/socket/v2_json_serializer_test.exs b/test/phoenix/socket/v2_json_serializer_test.exs
index 808638e21..4723eb727 100644
--- a/test/phoenix/socket/v2_json_serializer_test.exs
+++ b/test/phoenix/socket/v2_json_serializer_test.exs
@@ -3,14 +3,74 @@ defmodule Phoenix.Socket.V2.JSONSerializerTest do
   alias Phoenix.Socket.{Broadcast, Message, Reply, V2}
 
   @serializer V2.JSONSerializer
-  @v2_fastlane_json "[null,null,\"t\",\"e\",\"m\"]"
-  @v2_reply_json "[null,null,\"t\",\"phx_reply\",{\"response\":\"m\",\"status\":null}]"
-  @v2_msg_json "[null,null,\"t\",\"e\",\"m\"]"
+  @v2_fastlane_json "[null,null,\"t\",\"e\",{\"m\":1}]"
+  @v2_reply_json "[null,null,\"t\",\"phx_reply\",{\"response\":{\"m\":1},\"status\":null}]"
+  @v2_msg_json "[null,null,\"t\",\"e\",{\"m\":1}]"
+
+  @client_push <<
+    # push
+    0::size(8),
+    # join_ref_size
+    2,
+    # ref_size
+    3,
+    # topic_size
+    5,
+    # event_size
+    5,
+    "12",
+    "123",
+    "topic",
+    "event",
+    101,
+    102,
+    103
+  >>
+
+  @reply <<
+    # reply
+    1::size(8),
+    # join_ref_size
+    2,
+    # ref_size
+    3,
+    # topic_size
+    5,
+    # status_size
+    2,
+    "12",
+    "123",
+    "topic",
+    "ok",
+    101,
+    102,
+    103
+  >>
+
+  @broadcast <<
+    # broadcast
+    2::size(8),
+    # topic_size
+    5,
+    # event_size
+    5,
+    "topic",
+    "event",
+    101,
+    102,
+    103
+  >>
 
   def encode!(serializer, msg) do
-    {:socket_push, :text, encoded} = serializer.encode!(msg)
-    assert is_list(encoded)
-    IO.iodata_to_binary(encoded)
+    case serializer.encode!(msg) do
+      {:socket_push, :text, encoded} ->
+        assert is_list(encoded)
+        IO.iodata_to_binary(encoded)
+
+      {:socket_push, :binary, encoded} ->
+        assert is_binary(encoded)
+        encoded
+    end
   end
 
   def decode!(serializer, msg, opts \\ []) do
@@ -18,28 +78,166 @@ defmodule Phoenix.Socket.V2.JSONSerializerTest do
   end
 
   def fastlane!(serializer, msg) do
-    {:socket_push, :text, encoded} = serializer.fastlane!(msg)
-    assert is_list(encoded)
-    IO.iodata_to_binary(encoded)
+    case serializer.fastlane!(msg) do
+      {:socket_push, :text, encoded} ->
+        assert is_list(encoded)
+        IO.iodata_to_binary(encoded)
+
+      {:socket_push, :binary, encoded} ->
+        assert is_binary(encoded)
+        encoded
+    end
   end
 
   test "encode!/1 encodes `Phoenix.Socket.Message` as JSON" do
-    msg = %Message{topic: "t", event: "e", payload: "m"}
+    msg = %Message{topic: "t", event: "e", payload: %{m: 1}}
     assert encode!(@serializer, msg) == @v2_msg_json
   end
 
+  test "encode!/1 raises when payload is not a map" do
+    msg = %Message{topic: "t", event: "e", payload: "invalid"}
+    assert_raise ArgumentError, fn -> encode!(@serializer, msg) end
+  end
+
   test "encode!/1 encodes `Phoenix.Socket.Reply` as JSON" do
-    msg = %Reply{topic: "t", payload: "m"}
+    msg = %Reply{topic: "t", payload: %{m: 1}}
     assert encode!(@serializer, msg) == @v2_reply_json
   end
 
   test "decode!/2 decodes `Phoenix.Socket.Message` from JSON" do
-    assert %Message{topic: "t", event: "e", payload: "m"} ==
-      decode!(@serializer, @v2_msg_json)
+    assert %Message{topic: "t", event: "e", payload: %{"m" => 1}} ==
+             decode!(@serializer, @v2_msg_json, opcode: :text)
   end
 
   test "fastlane!/1 encodes a broadcast into a message as JSON" do
-    msg = %Broadcast{topic: "t", event: "e", payload: "m"}
+    msg = %Broadcast{topic: "t", event: "e", payload: %{m: 1}}
     assert fastlane!(@serializer, msg) == @v2_fastlane_json
   end
+
+  test "fastlane!/1 raises when payload is not a map" do
+    msg = %Broadcast{topic: "t", event: "e", payload: "invalid"}
+    assert_raise ArgumentError, fn -> fastlane!(@serializer, msg) end
+  end
+
+  describe "binary encode" do
+    test "general pushed message" do
+      push = <<
+        # push
+        0::size(8),
+        # join_ref_size
+        2,
+        # topic_size
+        5,
+        # event_size
+        5,
+        "12",
+        "topic",
+        "event",
+        101,
+        102,
+        103
+      >>
+
+      assert encode!(@serializer, %Phoenix.Socket.Message{
+               join_ref: "12",
+               ref: nil,
+               topic: "topic",
+               event: "event",
+               payload: {:binary, <<101, 102, 103>>}
+             }) == push
+    end
+
+    test "encode with oversized headers" do
+      assert_raise ArgumentError, ~r/unable to convert topic to binary/, fn ->
+        encode!(@serializer, %Phoenix.Socket.Message{
+          join_ref: "12",
+          ref: nil,
+          topic: String.duplicate("t", 256),
+          event: "event",
+          payload: {:binary, <<101, 102, 103>>}
+        })
+      end
+
+      assert_raise ArgumentError, ~r/unable to convert event to binary/, fn ->
+        encode!(@serializer, %Phoenix.Socket.Message{
+          join_ref: "12",
+          ref: nil,
+          topic: "topic",
+          event: String.duplicate("e", 256),
+          payload: {:binary, <<101, 102, 103>>}
+        })
+      end
+
+      assert_raise ArgumentError, ~r/unable to convert join_ref to binary/, fn ->
+        encode!(@serializer, %Phoenix.Socket.Message{
+          join_ref: String.duplicate("j", 256),
+          ref: nil,
+          topic: "topic",
+          event: "event",
+          payload: {:binary, <<101, 102, 103>>}
+        })
+      end
+    end
+
+    test "reply" do
+      assert encode!(@serializer, %Phoenix.Socket.Reply{
+               join_ref: "12",
+               ref: "123",
+               topic: "topic",
+               status: :ok,
+               payload: {:binary, <<101, 102, 103>>}
+             }) == @reply
+    end
+
+    test "reply with oversized headers" do
+      assert_raise ArgumentError, ~r/unable to convert ref to binary/, fn ->
+        encode!(@serializer, %Phoenix.Socket.Reply{
+          join_ref: "12",
+          ref: String.duplicate("r", 256),
+          topic: "topic",
+          status: :ok,
+          payload: {:binary, <<101, 102, 103>>}
+        })
+      end
+    end
+
+    test "fastlane" do
+      assert fastlane!(@serializer, %Phoenix.Socket.Broadcast{
+               topic: "topic",
+               event: "event",
+               payload: {:binary, <<101, 102, 103>>}
+             }) == @broadcast
+    end
+
+    test "fastlane with oversized headers" do
+      assert_raise ArgumentError, ~r/unable to convert topic to binary/, fn ->
+        fastlane!(@serializer, %Phoenix.Socket.Broadcast{
+          topic: String.duplicate("t", 256),
+          event: "event",
+          payload: {:binary, <<101, 102, 103>>}
+        })
+      end
+
+      assert_raise ArgumentError, ~r/unable to convert event to binary/, fn ->
+        fastlane!(@serializer, %Phoenix.Socket.Broadcast{
+          topic: "topic",
+          event: String.duplicate("e", 256),
+          payload: {:binary, <<101, 102, 103>>}
+        })
+      end
+
+    end
+  end
+
+  describe "binary decode" do
+    test "pushed message" do
+      assert decode!(@serializer, @client_push, opcode: :binary) == %Phoenix.Socket.Message{
+               join_ref: "12",
+               ref: "123",
+               topic: "topic",
+               event: "event",
+               payload: {:binary, <<101, 102, 103>>}
+             }
+    end
+  end
 end
diff --git a/test/support/websocket_client.exs b/test/support/websocket_client.exs
index 2438abd52..a1a204fec 100644
--- a/test/support/websocket_client.exs
+++ b/test/support/websocket_client.exs
@@ -84,6 +84,11 @@ defmodule Phoenix.Integration.WebsocketClient do
     {:ok, state}
   end
 
+  def websocket_handle({:binary, data}, _conn_state, state) do
+    send(state.sender, binary_decode(data))
+    {:ok, state}
+  end
+
   # The websocket client always sends a payload, even when none is explicitly set
   # on the frame.
   def websocket_handle({opcode, msg}, _conn_state, state) when opcode in [:ping, :pong] do
@@ -103,9 +108,15 @@ defmodule Phoenix.Integration.WebsocketClient do
     end
   end
 
+  def websocket_info({:send, %Message{payload: {:binary, _}} = msg}, _conn_state, %{ref: ref} = state) do
+    {join_ref, state} = join_ref_for(msg, state)
+    msg = Map.merge(msg, %{ref: to_string(ref), join_ref: to_string(join_ref)})
+    {:reply, {:binary, binary_encode_push!(msg)}, put_in(state.ref, ref + 1)}
+  end
+
   def websocket_info({:send, %Message{} = msg}, _conn_state, %{ref: ref} = state) do
     {join_ref, state} = join_ref_for(msg, state)
-    msg = Map.merge(msg, %{ref: to_string(ref), join_ref: join_ref})
+    msg = Map.merge(msg, %{ref: to_string(ref), join_ref: to_string(join_ref)})
     {:reply, {:text, encode!(msg, state)}, put_in(state.ref, ref + 1)}
   end
 
@@ -131,4 +142,57 @@ defmodule Phoenix.Integration.WebsocketClient do
     {:socket_push, :text, chardata} = state.serializer.encode!(map)
     IO.chardata_to_string(chardata)
   end
+
+  defp binary_encode_push!(%Message{payload: {:binary, data}} = msg) do
+    ref = to_string(msg.ref)
+    join_ref = to_string(msg.join_ref)
+    join_ref_size = byte_size(join_ref)
+    ref_size = byte_size(ref)
+    topic_size = byte_size(msg.topic)
+    event_size = byte_size(msg.event)
+
+    <<
+      0::size(8),
+      join_ref_size::size(8),
+      ref_size::size(8),
+      topic_size::size(8),
+      event_size::size(8),
+      join_ref::binary-size(join_ref_size),
+      ref::binary-size(ref_size),
+      msg.topic::binary-size(topic_size),
+      msg.event::binary-size(event_size),
+      data::binary
+    >>
+  end
+
+  # push
+  defp binary_decode(<<
+         0::size(8),
+         join_ref_size::size(8),
+         topic_size::size(8),
+         event_size::size(8),
+         join_ref::binary-size(join_ref_size),
+         topic::binary-size(topic_size),
+         event::binary-size(event_size),
+         data::binary
+       >>) do
+    %Message{join_ref: join_ref, topic: topic, event: event, payload: {:binary, data}}
+  end
+
+  # reply
+  defp binary_decode(<<
+         1::size(8),
+         join_ref_size::size(8),
+         ref_size::size(8),
+         topic_size::size(8),
+         status_size::size(8),
+         join_ref::binary-size(join_ref_size),
+         ref::binary-size(ref_size),
+         topic::binary-size(topic_size),
+         status::binary-size(status_size),
+         data::binary
+       >>) do
+    payload = %{"status" => status, "response" => {:binary, data}}
+    %Message{join_ref: join_ref, ref: ref, topic: topic, event: "phx_reply", payload: payload}
+  end
 end