Repository: phoenixframework/phoenix
Pull request #4009 (2020-10-01 20:17:32)

Add binary serialization support to default serializers
Initial pass. We need to decide how we want to support the caller push and receive of binary data. For example, we can support `push("foo", <<>>)` or we can require an explicit `push("foo", {:binary, "..."})` since we can't really distinguish between someone pushing a string vs binary and as of now we require maps. Likewise, on the `handle_in` side, either `handle_in("foo", <<>>, socket)` or `handle_in("foo", {:binary, <<>>}, socket)`

shoop2019: Merge
shoop2019: Cool
josevalim: Looks great! We should look into adding an integration test too in websocket_channels_test just to make sure the whole experience works.
chrismccord: Addressed all feedback. I'll push docs next and I believe this will be good to go. Thanks!

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/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