diff --git a/assets/js/phoenix/channel.js b/assets/js/phoenix/channel.js index 76bcb33285..28f71a264a 100644 --- a/assets/js/phoenix/channel.js +++ b/assets/js/phoenix/channel.js @@ -23,6 +23,8 @@ export default class Channel { this.bindingRef = 0 this.timeout = this.socket.timeout this.joinedOnce = false + // the ref of the join the server sent for us, see awaitJoin + this.awaitedJoinRef = null this.joinPush = new Push(this, CHANNEL_EVENTS.join, this.params, this.timeout) this.pushBuffer = [] this.stateChangeRefs = [] @@ -87,6 +89,39 @@ export default class Channel { } } + /** + * Waits for the reply to a channel join initiated by the server. See `Socket.attachSession`. + * + * Call this method before the socket finishes attaching. If the socket is not attaching, + * the method calls `join()` instead. If attachment fails, the channel joins again through + * a new connection. + * + * @param {string} joinRef - The ref of the join, given by the server + * @param {integer} timeout + * @returns {Push} + */ + awaitJoin(joinRef, timeout = this.timeout){ + if(this.joinedOnce){ + throw new Error("tried to join multiple times. 'join' can only be called a single time per channel instance") + } + if(!this.socket.isAttaching()){ return this.join(timeout) } + this.timeout = timeout + this.joinedOnce = true + this.socket.leaveOpenTopic(this.topic) + this.state = CHANNEL_STATES.joining + this.joinPush.awaitReply(joinRef, timeout) + this.awaitedJoinRef = joinRef + this.socket.carryAwaitedJoin(this) + return this.joinPush + } + + /** + * @private + * + * Whether the channel waits for the reply to a join the server sent for us. + */ + awaitsJoin(){ return this.isJoining() && this.awaitedJoinRef !== null && this.joinRef() === this.awaitedJoinRef } + /** * Hook into channel close * @param {Function} callback diff --git a/assets/js/phoenix/longpoll.js b/assets/js/phoenix/longpoll.js index 5eb81c500c..390791cbb9 100644 --- a/assets/js/phoenix/longpoll.js +++ b/assets/js/phoenix/longpoll.js @@ -28,7 +28,14 @@ let arrayBufferToBase64 = (buffer) => { export default class LongPoll { - constructor(endPoint, protocols){ + /** + * @param {string} endPoint + * @param {string[]} [protocols] + * @param {Object} [opts] + * @param {string} [opts.session] - The token of a session the server started for us, + * which we attach to like we resume a session, see `Socket.attachSession` + */ + constructor(endPoint, protocols, opts = {}){ // the URL and subprotocols of a poll that resumes the session, which the socket // replaces so that we resume with its current params and auth token this.connectionInfo = () => ({endPoint, protocols}) @@ -74,6 +81,9 @@ export default class LongPoll { this.onsuspend = function (){ } // noop this.onresume = function (){ } // noop this.readyState = SOCKET_STATES.connecting + if(opts.session){ + this.resumeSession({token: opts.session, received: 0, confirmed: 0, unconfirmed: []}) + } // we must wait for the caller to finish setting up our callbacks and timeout properties, // and it might close us before that, e.g. when disconnecting right after connecting setTimeout(() => this.poll(), 0) diff --git a/assets/js/phoenix/push.js b/assets/js/phoenix/push.js index e243efba43..409b8e6278 100644 --- a/assets/js/phoenix/push.js +++ b/assets/js/phoenix/push.js @@ -43,6 +43,18 @@ export default class Push { }, this.channel) } + /** + * @private + * + * Waits for the reply to a push the server sent for us with the given ref. + */ + awaitReply(ref, timeout){ + this.timeout = timeout + this.reset() + this.startTimeout(ref) + this.sent = true + } + /** * * @param {*} status @@ -95,10 +107,10 @@ export default class Push { /** * @private */ - startTimeout(){ + startTimeout(ref = this.channel.socket.makeRef()){ if(this.timeoutTimer){ this.cancelTimeout() } this.cancelRefEvent() - this.ref = this.channel.socket.makeRef() + this.ref = ref this.refEvent = this.channel.replyEventName(this.ref) this.channel.on(this.refEvent, payload => { diff --git a/assets/js/phoenix/resumable_websocket.js b/assets/js/phoenix/resumable_websocket.js index b18c2ebbc1..572c943696 100644 --- a/assets/js/phoenix/resumable_websocket.js +++ b/assets/js/phoenix/resumable_websocket.js @@ -19,6 +19,12 @@ const CONTROL_PREFIX = "[null,null,\"phoenix\",\"phx_" let control = (event, payload) => JSON.stringify([null, null, "phoenix", event, payload]) +// Heartbeats and their replies are V2 messages on the "phoenix" topic with a ref but without +// join ref. They only check the current WebSocket, so neither side numbers or replays them. +const HEARTBEAT_PATTERN = /^\[null,"[^"]*","phoenix","heartbeat",/ +const HEARTBEAT_REPLY_PATTERN = /^\[null,"[^"]*","phoenix","phx_reply",/ +let isHeartbeat = (data, pattern) => typeof(data) === "string" && pattern.test(data) + /** * A WebSocket transport that resumes its session over a new WebSocket when the * connection fails, without rejoining channels. @@ -35,7 +41,14 @@ let control = (event, payload) => JSON.stringify([null, null, "phoenix", event, * Servers that don't keep the socket for us make it behave like a WebSocket. */ export default class ResumableWebSocket { - constructor(endPoint, protocols){ + /** + * @param {string} endPoint + * @param {string[]} [protocols] + * @param {Object} [opts] + * @param {string} [opts.session] - The token of a session the server started for us, + * which we attach to like we resume a session, see `Socket.attachSession` + */ + constructor(endPoint, protocols, opts = {}){ // the URL and subprotocols of a new WebSocket, which the socket replaces // so that we resume with its current params and auth token this.connectionInfo = () => ({endPoint, protocols}) @@ -78,7 +91,13 @@ export default class ResumableWebSocket { this.onclose = function (){ } // noop this.onsuspend = function (){ } // noop this.onresume = function (){ } // noop - this.connect(false) + if(opts.session){ + // we attach to the session like we resume one, and open once it resumed + this.token = opts.session + this.connect(true) + } else { + this.connect(false) + } } get binaryType(){ return this.binaryTypeValue } @@ -97,7 +116,7 @@ export default class ResumableWebSocket { return } // we number the messages implicitly, in the order we send them - if(!this.unsupported){ this.unconfirmed.push(data) } + if(!this.unsupported && !isHeartbeat(data, HEARTBEAT_PATTERN)){ this.unconfirmed.push(data) } if(this.ready){ this.ws.send(data) } } @@ -147,7 +166,12 @@ export default class ResumableWebSocket { // a server that does not answer is as good as a failed connection this.resumeTimer = setTimeout(() => { this.dropWebSocket() - this.retry() + // a session we could not attach to is gone for us, so the socket connects again + if(this.readyState === SOCKET_STATES.connecting){ + this.end() + } else { + this.retry() + } }, this.resumeTimeout) } ws.onopen = () => { @@ -161,8 +185,9 @@ export default class ResumableWebSocket { } } ws.onerror = error => { - // failed attempts to resume the session are no errors of the socket - if(this.ws === ws && !this.token){ this.onerror(error) } + // Retry failed resume attempts without reporting a socket error. + // Report attachment failures because the transport has not opened yet. + if(this.ws === ws && (!this.token || this.readyState === SOCKET_STATES.connecting)){ this.onerror(error) } } ws.onmessage = event => { if(this.ws === ws){ this.handleMessage(event) } } ws.onclose = event => { if(this.ws === ws){ this.handleClose(event) } } @@ -192,6 +217,10 @@ export default class ResumableWebSocket { this.unsupported = true this.unconfirmed = [] } + if(!this.unsupported && isHeartbeat(data, HEARTBEAT_REPLY_PATTERN)){ + this.onmessage(event) + return + } this.received++ this.scheduleAck() this.onmessage(event) @@ -244,6 +273,11 @@ export default class ResumableWebSocket { // The WebSocket is usable now; replay must not trigger a long-poll fallback. clearTimeout(this.fallbackTimer) this.fallbackTimer = null + if(this.readyState === SOCKET_STATES.connecting){ + // we attached to a session the server started for us + this.readyState = SOCKET_STATES.open + this.onopen() + } this.scheduleRecovery() } diff --git a/assets/js/phoenix/socket.js b/assets/js/phoenix/socket.js index 7564b7c8fd..aa23da625d 100644 --- a/assets/js/phoenix/socket.js +++ b/assets/js/phoenix/socket.js @@ -148,6 +148,10 @@ export default class Socket { this.fallbackErrorRef = null // sends the health check of the current fallback attempt, see connectWithFallback this.fallbackHealthCheck = null + // the token of the session the next connection attaches to, see attachSession, + // and the connection that attaches to it + this.attachToken = null + this.attachConnection = null this.timeout = opts.timeout || DEFAULT_TIMEOUT this.transport = opts.transport || (global.WebSocket && opts.resumable ? ResumableWebSocket : global.WebSocket) || LongPoll this.primaryPassedHealthCheck = false @@ -362,6 +366,53 @@ export default class Socket { this.teardown(callback, code, reason) } + /** + * Configures the next connection to attach to an existing session. + * Pass the token returned by `Phoenix.Socket.start_detached/3`. + * + * The server can join channels in the session before the client attaches, with + * `Phoenix.Socket.join_detached/3`, which returns the join ref. Join them with + * `channel.awaitJoin(joinRef)` before connecting, so they receive the replies to these joins + * and the messages the server sent in the meantime. + * + * Attaching requires the ResumableWebSocket or LongPoll transport. A socket that uses the + * browser's WebSocket switches to ResumableWebSocket, as the server keeps sessions of this + * socket. When the client cannot attach, for example because the session expired, it connects + * with a new session and its channels join again. + * + * @param {string} token - The token of the session + * @returns {boolean} Whether the transport supports attachment and the token was stored + * for the next connection + */ + attachSession(token){ + if(global.WebSocket && this.transport === global.WebSocket){ this.transport = ResumableWebSocket } + if(this.transport !== ResumableWebSocket && this.transport !== LongPoll){ return false } + this.attachToken = token + return true + } + + /** + * @private + * + * Whether the socket attaches to a session and did not receive its messages yet. + */ + isAttaching(){ + if(this.attachToken !== null){ return true } + const connection = this.attachConnection + return connection !== null && connection.isCurrent() && connection.isConnecting() + } + + /** + * @private + * + * The reply to the awaited join of the channel arrives over the connection that attaches, + * so the channel joins again when that connection fails, as if it had sent the join. + */ + carryAwaitedJoin(channel){ + const connection = this.attachConnection + if(connection && connection.isCurrent()){ connection.joins.set(channel, channel.joinRef()) } + } + /** * * @param {Object} params - The params to send when connecting, for example `{user_id: userToken}` @@ -511,7 +562,12 @@ export default class Socket { // its tries, which only reset once a connection opens clearTimeout(this.reconnectTimer.timer) const {endPoint, protocols} = this.connectionInfo() - const transport = new this.transport(endPoint, protocols) + // only the first connection after attachSession attaches, later ones start a new session + const attachToken = this.attachToken + this.attachToken = null + const transport = attachToken ? + new this.transport(endPoint, protocols, {session: attachToken}) : + new this.transport(endPoint, protocols) transport.binaryType = this.binaryType transport.timeout = this.longpollerTimeout transport.resumeAfterMs = this.resumeAfterMs @@ -520,6 +576,8 @@ export default class Socket { // a transport that keeps its session falls back to long polling to continue it transport.longPollFallbackMs = this.longPollFallbackMs this.connection = new Connection(this, transport) + this.attachConnection = attachToken ? this.connection : null + if(attachToken){ this.channels.forEach(channel => channel.awaitsJoin() && this.carryAwaitedJoin(channel)) } // connect can replace an ended connection that is still open while its buffer drains, but the // channels joined over it need to rejoin on the new connection instead of waiting for it to close if(previous && previous.isOpen()){ this.triggerChanError("connection_closed", previous) } diff --git a/assets/test/attach_test.js b/assets/test/attach_test.js new file mode 100644 index 0000000000..c7d0fc4ee0 --- /dev/null +++ b/assets/test/attach_test.js @@ -0,0 +1,250 @@ +import {jest} from "@jest/globals" +import {LongPoll, ResumableWebSocket, Socket} from "../js/phoenix" +import {SOCKET_STATES} from "../js/phoenix/constants" +import Ajax from "../js/phoenix/ajax" + +const control = (event, payload) => JSON.stringify([null, null, "phoenix", event, payload]) +const reply = (joinRef, ref, topic, status, response) => + JSON.stringify([joinRef, ref, topic, "phx_reply", {status, response}]) + +class FakeWebSocket { + constructor(url, protocols){ + this.url = url + this.protocols = protocols + this.readyState = SOCKET_STATES.connecting + this.sent = [] + this.closed = null + FakeWebSocket.instances.push(this) + } + + get params(){ return Object.fromEntries(new URL(this.url).searchParams) } + + send(data){ this.sent.push(data) } + + close(code, reason){ + this.closed = {code, reason} + this.readyState = SOCKET_STATES.closed + } + + open(){ + this.readyState = SOCKET_STATES.open + this.onopen() + } + + receive(data){ this.onmessage({data}) } + + drop(code = 1006){ + this.readyState = SOCKET_STATES.closed + this.onclose({code, reason: "", wasClean: false}) + } +} + +FakeWebSocket.instances = [] + +const OriginalWebSocket = global.WebSocket +const last = () => FakeWebSocket.instances.at(-1) +const events = ws => ws.sent.filter(data => !data.startsWith("[null,null,\"phoenix\",\"phx_")).map(data => JSON.parse(data)) + +describe("Socket.attachSession", () => { + let socket + + beforeEach(() => { + jest.useFakeTimers() + FakeWebSocket.instances = [] + global.WebSocket = FakeWebSocket + }) + + afterEach(() => { + socket.disconnect() + global.WebSocket = OriginalWebSocket + jest.useRealTimers() + }) + + it("switches a socket with the browser's WebSocket to ResumableWebSocket", () => { + socket = new Socket("ws://localhost/socket") + expect(socket.attachSession("token")).toBe(true) + expect(socket.transport).toBe(ResumableWebSocket) + }) + + it("does not attach with a transport that cannot resume sessions", () => { + class CustomTransport extends FakeWebSocket {} + socket = new Socket("ws://localhost/socket", {transport: CustomTransport}) + expect(socket.attachSession("token")).toBe(false) + expect(socket.transport).toBe(CustomTransport) + }) + + it("attaches to the session and receives the replies of the joins the server sent", () => { + socket = new Socket("ws://localhost/socket") + socket.attachSession("token") + const channel = socket.channel("room:a", {a: 1}) + const replies = [] + channel.awaitJoin("phx-1").receive("ok", response => replies.push(response)) + socket.connect() + + const ws = last() + expect(ws.params).toEqual({vsn: "2.0.0", "phx:resumable": "1", "phx:resume": "1"}) + ws.open() + expect(ws.sent).toEqual([control("phx_resume", {token: "token", ack: 0})]) + expect(socket.isConnected()).toBe(false) + + ws.receive(control("phx_resumed", {seq: 0, last: 2})) + expect(socket.isConnected()).toBe(true) + ws.receive(reply("phx-1", "phx-1", "room:a", "ok", {rendered: true})) + expect(replies).toEqual([{rendered: true}]) + expect(channel.isJoined()).toBe(true) + + // the client does not join, and pushes with the join ref of the server + channel.push("event", {}) + expect(events(ws).map(([joinRef, , topic, event]) => [joinRef, topic, event])) + .toEqual([["phx-1", "room:a", "event"]]) + }) + + it("delivers the messages the channel received before the client attached", () => { + socket = new Socket("ws://localhost/socket") + socket.attachSession("token") + const channel = socket.channel("room:a") + const pushes = [] + channel.on("pushed", payload => pushes.push(payload)) + channel.awaitJoin("phx-1") + socket.connect() + last().open() + last().receive(control("phx_resumed", {seq: 0, last: 2})) + last().receive(reply("phx-1", "phx-1", "room:a", "ok", {})) + last().receive(JSON.stringify(["phx-1", null, "room:a", "pushed", {n: 1}])) + expect(pushes).toEqual([{n: 1}]) + }) + + it("can await joins after connecting, until the socket attached", () => { + socket = new Socket("ws://localhost/socket") + socket.attachSession("token") + socket.connect() + const channel = socket.channel("room:a") + channel.awaitJoin("phx-1") + last().open() + last().receive(control("phx_resumed", {seq: 0, last: 1})) + last().receive(reply("phx-1", "phx-1", "room:a", "ok", {})) + expect(channel.isJoined()).toBe(true) + + // the replies of the server are gone, so later channels join + const later = socket.channel("room:b") + later.awaitJoin("phx-2") + expect(events(last()).map(([, , topic, event]) => [topic, event])).toEqual([["room:b", "phx_join"]]) + }) + + it("joins when the socket does not attach", () => { + socket = new Socket("ws://localhost/socket") + const channel = socket.channel("room:a") + channel.awaitJoin("phx-1") + socket.connect() + last().open() + expect(events(last()).map(([, , topic, event]) => [topic, event])).toEqual([["room:a", "phx_join"]]) + expect(channel.joinRef()).not.toBe("phx-1") + }) + + it("connects with a new session and joins when the session is gone", () => { + socket = new Socket("ws://localhost/socket") + socket.attachSession("token") + const channel = socket.channel("room:a", {a: 1}) + channel.awaitJoin("phx-1") + socket.connect() + last().open() + last().receive(control("phx_gone", {})) + expect(socket.isConnected()).toBe(false) + + jest.advanceTimersByTime(10) + const ws = last() + expect(ws.params).toEqual({vsn: "2.0.0", "phx:resumable": "1"}) + ws.open() + ws.receive(control("phx_session", {token: "new"})) + expect(events(ws).map(([, , topic, event, payload]) => [topic, event, payload])) + .toEqual([["room:a", "phx_join", {a: 1}]]) + }) + + it("connects with a new session when the server does not answer", () => { + socket = new Socket("ws://localhost/socket") + socket.attachSession("token") + const channel = socket.channel("room:a") + channel.awaitJoin("phx-1") + socket.connect() + const ws = last() + ws.open() + + jest.advanceTimersByTime(10_000) + expect(ws.closed).not.toBeNull() + jest.advanceTimersByTime(10) + expect(last()).not.toBe(ws) + expect(last().params["phx:resume"]).toBeUndefined() + last().open() + expect(events(last()).map(([, , topic, event]) => [topic, event])).toEqual([["room:a", "phx_join"]]) + }) + + it("reports an error when the WebSocket fails before it attached", () => { + socket = new Socket("ws://localhost/socket") + const errors = [] + socket.onError(error => errors.push(error)) + socket.attachSession("token") + socket.connect() + last().onerror("failed") + expect(errors).toEqual(["failed"]) + }) + + describe("with long polling", () => { + let polls + + beforeEach(() => { + polls = [] + jest.spyOn(Ajax, "request").mockImplementation((method, url, headers, body, timeout, ontimeout, callback) => { + polls.push({method, params: Object.fromEntries(new URL(url).searchParams), headers, body, callback}) + return {abort: jest.fn()} + }) + }) + + afterEach(() => jest.restoreAllMocks()) + + it("attaches to the session with the first poll", () => { + socket = new Socket("ws://localhost/socket", {transport: LongPoll}) + expect(socket.attachSession("token")).toBe(true) + const channel = socket.channel("room:a") + const replies = [] + channel.awaitJoin("phx-1").receive("ok", response => replies.push(response)) + socket.connect() + jest.advanceTimersByTime(0) + + expect(polls).toHaveLength(1) + expect(polls[0].params).toEqual({vsn: "2.0.0", "phx:ack": "0", "phx:resume": "1"}) + expect(polls[0].headers["X-Phoenix-Longpoll-Token"]).toBe("token") + + polls[0].callback({ + status: 200, + token: "token", + messages: [reply("phx-1", "phx-1", "room:a", "ok", {rendered: true})], + last: 1 + }) + expect(socket.isConnected()).toBe(true) + jest.advanceTimersByTime(0) + expect(replies).toEqual([{rendered: true}]) + expect(polls.at(-1).params).toEqual({vsn: "2.0.0", "phx:ack": "1"}) + }) + + it("connects with a new session and joins when the session is gone", () => { + socket = new Socket("ws://localhost/socket", {transport: LongPoll}) + socket.attachSession("token") + const channel = socket.channel("room:a") + channel.awaitJoin("phx-1") + socket.connect() + jest.advanceTimersByTime(0) + polls[0].callback({status: 410}) + + // the socket connects again, and the new LongPoll polls on the next tick + jest.advanceTimersByTime(20) + const poll = polls.at(-1) + expect(poll.params).toEqual({vsn: "2.0.0"}) + expect(poll.headers["X-Phoenix-Longpoll-Token"]).toBeUndefined() + poll.callback({status: 410, token: "new", messages: [], last: 0}) + jest.advanceTimersByTime(0) + + const post = polls.find(({method}) => method === "POST") + expect(JSON.parse(post.body).slice(2, 4)).toEqual(["room:a", "phx_join"]) + }) + }) +}) diff --git a/assets/test/resumable_websocket_test.js b/assets/test/resumable_websocket_test.js index bfd37892cc..54b60984ba 100644 --- a/assets/test/resumable_websocket_test.js +++ b/assets/test/resumable_websocket_test.js @@ -138,6 +138,30 @@ describe("ResumableWebSocket", () => { expect(last().sent.slice(1)).toEqual(["c"]) }) + it("neither numbers nor replays heartbeats and their replies", () => { + connect() + const heartbeat = JSON.stringify([null, "5", "phoenix", "heartbeat", {}]) + const reply = JSON.stringify([null, "5", "phoenix", "phx_reply", {status: "ok", response: {}}]) + transport.send(heartbeat) + transport.send("a") + expect(last().sent).toEqual([heartbeat, "a"]) + expect(transport.unconfirmed).toEqual(["a"]) + + last().receive(reply) + expect(events).toEqual(["open", reply]) + expect(transport.received).toBe(0) + // there is nothing to acknowledge + jest.advanceTimersByTime(1000) + expect(last().sent).toEqual([heartbeat, "a"]) + + last().drop() + jest.advanceTimersByTime(100) + last().open() + expect(last().sent).toEqual([control("phx_resume", {token: "token", ack: 0})]) + last().receive(control("phx_resumed", {seq: 0, last: 0})) + expect(last().sent.slice(1)).toEqual(["a"]) + }) + it("reports suspension once resuming failed, and resumption once it succeeded", () => { connect() last().drop() diff --git a/lib/phoenix.ex b/lib/phoenix.ex index f1e86e2a44..4e37ae84fd 100644 --- a/lib/phoenix.ex +++ b/lib/phoenix.ex @@ -29,7 +29,12 @@ defmodule Phoenix do children = [ # Code reloading must be serial across all Phoenix apps Phoenix.CodeReloader.Server, - {DynamicSupervisor, name: Phoenix.Transports.Session.Supervisor, strategy: :one_for_one} + {DynamicSupervisor, name: Phoenix.Transports.Session.Supervisor, strategy: :one_for_one}, + # the detached sessions without a client, see Phoenix.Socket.start_detached/3 + {Registry, + name: Phoenix.Transports.Detached.Registry, + keys: :unique, + partitions: System.schedulers_online()} ] Supervisor.start_link(children, strategy: :one_for_one, name: Phoenix.Supervisor) diff --git a/lib/phoenix/endpoint.ex b/lib/phoenix/endpoint.ex index 4c4509c67a..e8d16688ed 100644 --- a/lib/phoenix/endpoint.ex +++ b/lib/phoenix/endpoint.ex @@ -779,6 +779,36 @@ defmodule Phoenix.Endpoint do paths end + @doc false + # The handler of the socket at the given path and the config of its transports, + # as the transports receive it, or nil if the endpoint has no socket at the path. + def __socket_config__(endpoint, path) do + Phoenix.Config.cache(endpoint, {:socket_config, path}, fn _ -> + config = + Enum.find_value(endpoint.__sockets__(), fn + {^path, handler, opts} -> + %{ + handler: handler, + websocket: transport_config(opts, :websocket, true, Phoenix.Transports.WebSocket), + longpoll: transport_config(opts, :longpoll, false, Phoenix.Transports.LongPoll) + } + + _ -> + nil + end) + + {:cache, config} + end) + end + + defp transport_config(opts, key, default, transport) do + if transport_opts = Keyword.get(opts, key, default) do + transport_opts + |> put_socket_opts(opts) + |> Phoenix.Socket.Transport.load_config(transport) + end + end + # Copies top-level socket options into the transport config, # which is what the socket receives on connect. defp put_socket_opts(true, opts), do: put_socket_opts([], opts) diff --git a/lib/phoenix/socket.ex b/lib/phoenix/socket.ex index 2495edfebb..42f6faeaa5 100644 --- a/lib/phoenix/socket.ex +++ b/lib/phoenix/socket.ex @@ -128,6 +128,13 @@ defmodule Phoenix.Socket do and long polling that interacts with Phoenix.Socket and can be used as reference for those interested in implementing custom clients. + ## Detached sessions + + A request can start a session of a socket before its client connects, for + example while it renders a page, and join channels in it. The client attaches + to the session afterwards and receives the replies of these joins, together + with the messages the channels pushed in the meantime. See `start_detached/3`. + ## Custom sockets and transports See the `Phoenix.Socket.Transport` documentation for more information on @@ -366,6 +373,123 @@ defmodule Phoenix.Socket do ## USER API + @doc """ + Starts a detached socket session at `path` using the HTTP connection `conn`. + + The server can join channels with `join_detached/3` before the client connects, + for example while rendering a page. The client then attaches to the session with + the returned token, see `Socket.attachSession` and `Channel.awaitJoin` in + `phoenix.js`. It receives the replies of the joins and the messages the + channels pushed before it attached. + + Builds `connect_info` from `conn` using the socket's `:connect_info` configuration. + If the request has fetched its HTTP session, `connect_info.session` contains + its current contents. The socket receives the params `%{"vsn" => "2.0.0"}`, + and `:detached` as transport. + When the client attaches, the socket checks its connection with `c:resume/3`. + + The socket must be resumable. A WebSocket needs the `:resumable` option of + `Phoenix.Endpoint.socket/3`. When it is a function, it decides after `c:connect/3` + accepted the connection, as for a new WebSocket. Long polling always resumes. + + The session stops when no client attached within its window, like a session + that lost its client. + + ## Options + + * `:window_ms` - the base timeout for client attachment, in milliseconds. + The server uses `1.5 × :window_ms` as the session timeout. Defaults to the + `:window_ms` of the socket transport. + + * `:max_pending` - the maximum number of detached sessions without a client + on this node, across all endpoints. When the node has that many, no session + starts. Requests that start sessions at the same time can exceed it slightly. + Defaults to no limit. See `detached_count/0`. + + ## Return values + + * `{:ok, session, token}` - the session and the token the client attaches with + + * `{:error, :limit}` - the node reached the `:max_pending` limit + + * `{:error, :no_socket}` - the endpoint has no socket at `path` + + * `{:error, :not_resumable}` - the socket is not resumable + + * `{:error, :denied}` - the `:resumable` function of the socket returned `false` + + * `{:error, :not_a_phoenix_socket}` - the socket at `path` does not use `Phoenix.Socket` + + * `{:error, reason}` - `c:connect/3` rejected the connection, with `:rejected` + as reason when it returned `:error` + + > #### Experimental {: .warning} + > + > Detached sessions are experimental and might change in future releases. + """ + @spec start_detached(Plug.Conn.t(), String.t(), keyword) :: + {:ok, pid, String.t()} | {:error, term} + def start_detached(%Plug.Conn{} = conn, path, opts \\ []) when is_binary(path) do + endpoint = + conn.private[:phoenix_endpoint] || + raise ArgumentError, "start_detached/3 requires a conn that went through an endpoint" + + Phoenix.Transports.Detached.start(conn, endpoint, path, opts) + end + + @doc """ + Joins `topic` in a session started with `start_detached/3`, like a client that + sends a join with `payload`. + + The channel cannot tell this join from a join of the client, so `payload` should + contain what the client would send. To pass data that only the server can provide, + use the `:assigns` option. + + ## Options + + * `:assigns` - a map merged into the channel socket assigns, as with `channel/3`. + These values come from the server and cannot be supplied by clients. + + Returns `{:ok, join_ref}` when the channel joins. Pass `join_ref` to + `Channel.awaitJoin` on the client. The session buffers the join reply until + the client attaches. + + Returns `{:error, reply}` if the channel rejects the join, crashes, or has + already joined `topic` in this session. + """ + @spec join_detached(pid, String.t(), term, keyword) :: {:ok, String.t()} | {:error, map} + def join_detached(session, topic, payload, opts \\ []) + when is_pid(session) and is_binary(topic) do + assigns = Keyword.get(opts, :assigns, %{}) + + if not is_map(assigns) do + raise ArgumentError, ":assigns must be a map, got: #{inspect(assigns)}" + end + + Phoenix.Transports.Detached.join(session, topic, payload, assigns) + end + + @doc """ + Returns the number of detached sessions on this node that no client attached to yet. + + See `start_detached/3`. + """ + @spec detached_count() :: non_neg_integer + def detached_count do + Phoenix.Transports.Detached.count() + end + + @doc """ + Stops a session started with `start_detached/3`, together with its channels. + + Use it when the client won't attach, for example because the request redirects + instead of rendering a page. + """ + @spec stop_detached(pid) :: :ok + def stop_detached(session) when is_pid(session) do + Phoenix.Transports.Detached.stop(session) + end + @doc """ Adds a `key`/`value` pair to `socket` assigns. @@ -632,6 +756,41 @@ defmodule Phoenix.Socket do {:ok, {state, socket}} end + # Joins a channel for a client that attaches to the socket later, see join_detached/3. + # The client did not send the join, so we pick its ref. The reply is for the client. + # Clients number their refs, so ours have a prefix to never match the refs of the client. + def __info__({__MODULE__, :join, {pid, ref}, topic, payload, assigns}, {state, socket}) do + count = state.detached_joins + 1 + state = %{state | detached_joins: count} + join_ref = "phx-" <> Integer.to_string(count) + + message = %Message{ + topic: topic, + event: "phx_join", + payload: payload, + ref: join_ref, + join_ref: join_ref + } + + with nil <- Map.get(state.channels, topic), + {%Reply{} = reply, state} <- join(message, state, socket, assigns) do + case reply.status do + :ok -> send(pid, {ref, :ok, join_ref}) + :error -> send(pid, {ref, :error, reply.payload}) + end + + {:push, encode_reply(socket, reply), {state, socket}} + else + :unmatched -> + send(pid, {ref, :error, %{reason: "unmatched topic"}}) + {:ok, {state, socket}} + + {_pid, _monitor_ref, _status} -> + send(pid, {ref, :error, %{reason: "already joined"}}) + {:ok, {state, socket}} + end + end + def __info__(_, state) do {:ok, state} end @@ -711,6 +870,8 @@ defmodule Phoenix.Socket do state = %{ channels: %{}, channels_inverse: %{}, + # the number of joins we sent for clients that attach later, see join_detached/3 + detached_joins: 0, max_channels_per_transport: Keyword.get(options, :max_channels_per_transport, 100) } @@ -771,57 +932,12 @@ defmodule Phoenix.Socket do {:reply, :ok, encode_reply(socket, reply), {state, socket}} end - defp handle_in( - nil, - %{event: "phx_join", topic: topic, ref: ref, join_ref: join_ref} = message, - state, - socket - ) do - case socket.handler.__channel__(topic) do - {channel, opts} -> - if map_size(state.channels) >= state.max_channels_per_transport do - Logger.error( - "Reached max channels per transport limit of #{state.max_channels_per_transport} for socket #{inspect(socket.id)}" - ) + defp handle_in(nil, %{event: "phx_join"} = message, state, socket) do + case join(message, state, socket) do + {%Reply{status: status} = reply, state} -> + {:reply, status, encode_reply(socket, reply), {state, socket}} - reply = %Reply{ - join_ref: join_ref, - ref: ref, - topic: topic, - status: :error, - payload: %{reason: "too many channels joined"} - } - - {:reply, :error, encode_reply(socket, reply), {state, socket}} - else - case Phoenix.Channel.Server.join(socket, channel, message, opts) do - {:ok, reply, pid} -> - reply = %Reply{ - join_ref: join_ref, - ref: ref, - topic: topic, - status: :ok, - payload: reply - } - - state = put_channel(state, pid, topic, join_ref) - {:reply, :ok, encode_reply(socket, reply), {state, socket}} - - {:error, reply} -> - reply = %Reply{ - join_ref: join_ref, - ref: ref, - topic: topic, - status: :error, - payload: reply - } - - {:reply, :error, encode_reply(socket, reply), {state, socket}} - end - end - - _ -> - Logger.warning("Ignoring unmatched topic \"#{topic}\" in #{inspect(socket.handler)}") + :unmatched -> {:reply, :error, encode_ignore(socket, message), {state, socket}} end end @@ -901,6 +1017,63 @@ defmodule Phoenix.Socket do {:reply, :error, encode_ignore(socket, message), {state, socket}} end + # The assigns come from the server, see join_detached/4. + defp join( + %{topic: topic, ref: ref, join_ref: join_ref} = message, + state, + socket, + assigns \\ %{} + ) do + case socket.handler.__channel__(topic) do + {channel, opts} -> + opts = Keyword.update(opts, :assigns, assigns, &Map.merge(&1, assigns)) + + if map_size(state.channels) >= state.max_channels_per_transport do + Logger.error( + "Reached max channels per transport limit of #{state.max_channels_per_transport} for socket #{inspect(socket.id)}" + ) + + reply = %Reply{ + join_ref: join_ref, + ref: ref, + topic: topic, + status: :error, + payload: %{reason: "too many channels joined"} + } + + {reply, state} + else + case Phoenix.Channel.Server.join(socket, channel, message, opts) do + {:ok, reply, pid} -> + reply = %Reply{ + join_ref: join_ref, + ref: ref, + topic: topic, + status: :ok, + payload: reply + } + + {reply, put_channel(state, pid, topic, join_ref)} + + {:error, reply} -> + reply = %Reply{ + join_ref: join_ref, + ref: ref, + topic: topic, + status: :error, + payload: reply + } + + {reply, state} + end + end + + _ -> + Logger.warning("Ignoring unmatched topic \"#{topic}\" in #{inspect(socket.handler)}") + :unmatched + end + end + defp put_channel(state, pid, topic, join_ref) do %{channels: channels, channels_inverse: channels_inverse} = state monitor_ref = Process.monitor(pid) diff --git a/lib/phoenix/transports/detached.ex b/lib/phoenix/transports/detached.ex new file mode 100644 index 0000000000..d624e8f022 --- /dev/null +++ b/lib/phoenix/transports/detached.ex @@ -0,0 +1,126 @@ +defmodule Phoenix.Transports.Detached do + @moduledoc false + # + # Detached sessions, see Phoenix.Socket.start_detached/3. + # + # The client attaches to the session with its token like it resumes a session, over + # a WebSocket or with long polling. The socket must be resumable, as a WebSocket that + # is not resumable treats the attaching client as a new connection. + + alias Phoenix.Socket.Transport + alias Phoenix.Transports.{Session, WebSocket} + + def start(%Plug.Conn{} = conn, endpoint, path, opts \\ []) do + with :ok <- check_limit(Keyword.get(opts, :max_pending)), + {:ok, handler, transport_opts, resumable} <- socket_config(endpoint, path), + config = connect_config(conn, endpoint, transport_opts), + {:ok, handler_state} <- connect(handler, config), + :ok <- check_resumable(resumable, config) do + transport_opts = Keyword.merge(transport_opts, Keyword.take(opts, [:window_ms])) + + case Session.start(endpoint, handler, handler_state, :detached, transport_opts) do + {:ok, session, priv_topic} -> + data = {:v1, endpoint.config(:endpoint_id), session, priv_topic} + {:ok, session, Session.sign_token(endpoint, data, transport_opts)} + + {:error, _} = error -> + error + end + end + end + + def join(session, topic, payload, assigns) do + ref = Process.monitor(session) + send(session, {Phoenix.Socket, :join, {self(), ref}, topic, payload, assigns}) + + receive do + {^ref, :ok, join_ref} -> + Process.demonitor(ref, [:flush]) + {:ok, join_ref} + + {^ref, :error, reply} -> + Process.demonitor(ref, [:flush]) + {:error, reply} + + {:DOWN, ^ref, _, _, _} -> + {:error, %{reason: "join crashed"}} + end + end + + # The sessions register themselves, so the count can exceed the limit when requests + # start sessions concurrently. That is fine, as the limit only sheds load. + def count do + Registry.count(__MODULE__.Registry) + end + + defp check_limit(nil), do: :ok + + defp check_limit(max) when is_integer(max) and max >= 0 do + if count() < max, do: :ok, else: {:error, :limit} + end + + defp check_limit(other) do + raise ArgumentError, ":max_pending must be a non-negative integer, got: #{inspect(other)}" + end + + def stop(session) do + GenServer.stop(session, {:shutdown, :closed}) + catch + :exit, _ -> :ok + end + + # The config of the transport that signs the token. Both transports + # must use the same :crypto options to resume each other's sessions. + defp socket_config(endpoint, path) do + case Phoenix.Endpoint.__socket_config__(endpoint, path) do + %{handler: handler, websocket: websocket, longpoll: longpoll} -> + # long polling always resumes + resumable = if websocket, do: Keyword.get(websocket, :resumable, false), else: true + + cond do + not function_exported?(handler, :__channel__, 1) -> {:error, :not_a_phoenix_socket} + resumable == false -> {:error, :not_resumable} + true -> {:ok, handler, websocket || longpoll, resumable} + end + + nil -> + {:error, :no_socket} + end + end + + defp connect_config(conn, endpoint, transport_opts) do + keys = Keyword.get(transport_opts, :connect_info, []) + # the request is not a cross-site request, unlike a WebSocket connection + connect_info = Transport.connect_info(conn, endpoint, keys, check_csrf: false) + + %{ + endpoint: endpoint, + transport: :detached, + options: transport_opts, + params: %{"vsn" => "2.0.0"}, + connect_info: put_session(connect_info, conn) + } + end + + # The request can change its session before it starts the session, for example + # when it logs the user in, so we prefer it over the session of its cookie. + defp put_session(%{session: _} = connect_info, %{private: %{plug_session_fetch: :done}} = conn) do + %{connect_info | session: Plug.Conn.get_session(conn)} + end + + defp put_session(connect_info, _conn), do: connect_info + + defp connect(handler, config) do + case handler.connect(config) do + {:ok, handler_state} -> {:ok, handler_state} + :error -> {:error, :rejected} + {:error, reason} -> {:error, reason} + end + end + + # A :resumable function decides after the socket accepted the connection, + # like it does for a new WebSocket. + defp check_resumable(resumable, config) do + if WebSocket.resumable?(resumable, config), do: :ok, else: {:error, :denied} + end +end diff --git a/lib/phoenix/transports/session.ex b/lib/phoenix/transports/session.ex index 1c76f1f398..acc2236d60 100644 --- a/lib/phoenix/transports/session.ex +++ b/lib/phoenix/transports/session.ex @@ -101,6 +101,11 @@ defmodule Phoenix.Transports.Session do # It needs the conn to answer a failed connect, and the :resumable option can decide # against a session after connecting. def init({:connected, endpoint, handler, handler_state, transport, options, priv_topic}) do + # detached sessions count against the limit until a client attaches + if transport == :detached do + {:ok, _} = Registry.register(Phoenix.Transports.Detached.Registry, self(), nil) + end + {:ok, handler_state} = handler.init(handler_state) {:ok, init_state(endpoint, handler, handler_state, transport, options, priv_topic)} end @@ -236,6 +241,10 @@ defmodule Phoenix.Transports.Session do case result do {:ok, handler_state} -> + if state.transport == :detached do + :ok = Registry.unregister(Phoenix.Transports.Detached.Registry, self()) + end + send_client(state, client_ref, {:resumed, ref}) # the client gave up on the WebSocket we have, for example to long poll instead state = %{detach(state) | handler: {handler, handler_state}, transport: info.transport} @@ -299,6 +308,28 @@ defmodule Phoenix.Transports.Session do {:noreply, state} end + # The reply to a heartbeat goes to the WebSocket that sent it, without a number, + # as there is no point in sending it again over another connection. + def handle_info({:pipe_heartbeat, pipe, {body, opcode}}, %{client: {:pipe, pipe, _}} = state) do + %{handler: {handler, handler_state}} = state + + case handler.handle_in({body, opcode: opcode}, handler_state) do + {:reply, _status, {opcode, reply}, handler_state} -> + send(pipe, {:push_frames, [{opcode, IO.iodata_to_binary(reply)}]}) + {:noreply, %{state | handler: {handler, handler_state}}} + + {:ok, handler_state} -> + {:noreply, %{state | handler: {handler, handler_state}}} + + {:stop, reason, handler_state} -> + {:stop, reason, %{state | handler: {handler, handler_state}}} + end + end + + def handle_info({:pipe_heartbeat, _pipe, _frame}, state) do + {:noreply, state} + end + def handle_info({:DOWN, ref, _, _, _}, %{client: {:pipe, _, ref}} = state) do {:noreply, %{state | client: nil, last_client_poll: now_ms()}} end diff --git a/lib/phoenix/transports/websocket.ex b/lib/phoenix/transports/websocket.ex index de98e187d3..ec99c04b04 100644 --- a/lib/phoenix/transports/websocket.ex +++ b/lib/phoenix/transports/websocket.ex @@ -116,9 +116,10 @@ defmodule Phoenix.Transports.WebSocket do # Whether a new connection gets a session. Clients that resume don't get here, as # they have the token of a session we started, which the socket checks with resume/3. - defp resumable?(true, _config), do: true + @doc false + def resumable?(true, _config), do: true - defp resumable?({module, function, args}, config) do + def resumable?({module, function, args}, config) do case apply(module, function, [config | args]) do resumable when is_boolean(resumable) -> resumable @@ -130,7 +131,7 @@ defmodule Phoenix.Transports.WebSocket do end end - defp resumable?(invalid, _config) do + def resumable?(invalid, _config) do raise ArgumentError, ":resumable expects a boolean or MFA tuple, got: #{inspect(invalid)}" end diff --git a/lib/phoenix/transports/websocket_pipe.ex b/lib/phoenix/transports/websocket_pipe.ex index 5eb08fc1a9..9e8cb4481f 100644 --- a/lib/phoenix/transports/websocket_pipe.ex +++ b/lib/phoenix/transports/websocket_pipe.ex @@ -20,7 +20,8 @@ defmodule Phoenix.Transports.WebSocket.Pipe do # * "phx_gone" - the session stopped or the client cannot resume it, with the # close code of the session if it stopped # - # Control messages are not numbered. + # Control messages are not numbered. Neither are heartbeats and their replies, + # as they only check the current WebSocket, like ping and pong frames. @behaviour WebSock @@ -82,11 +83,24 @@ defmodule Phoenix.Transports.WebSocket.Pipe do end def handle_in({payload, opcode: opcode}, state) do - seq = state.in_seq + 1 - send(state.session, {:pipe_in, self(), seq, {payload, opcode}}) - {:ok, %{state | in_seq: seq}} + if opcode == :text and heartbeat?(payload) do + send(state.session, {:pipe_heartbeat, self(), {payload, opcode}}) + {:ok, state} + else + seq = state.in_seq + 1 + send(state.session, {:pipe_in, self(), seq, {payload, opcode}}) + {:ok, %{state | in_seq: seq}} + end end + # A heartbeat of phoenix.js, a V2 message on the "phoenix" topic with a ref but + # without join ref. We don't decode the message, as the socket does that. + defp heartbeat?("[null,\"" <> rest) do + match?([_ref, "\"phoenix\",\"heartbeat\"," <> _], :binary.split(rest, "\",")) + end + + defp heartbeat?(_payload), do: false + defp handle_control("phx_resume", %{"token" => token, "ack" => ack}, %{resume: resume} = state) when is_binary(token) and is_integer(ack) and resume != nil do {endpoint, info, timer} = resume diff --git a/test/phoenix/integration/detached_session_test.exs b/test/phoenix/integration/detached_session_test.exs new file mode 100644 index 0000000000..9ea6543eb9 --- /dev/null +++ b/test/phoenix/integration/detached_session_test.exs @@ -0,0 +1,391 @@ +Code.require_file("../../support/websocket_client.exs", __DIR__) +Code.require_file("../../support/http_client.exs", __DIR__) + +defmodule Phoenix.Integration.DetachedSessionTest do + use ExUnit.Case + + import ExUnit.CaptureLog + + alias Phoenix.Integration.{HTTPClient, WebsocketClient} + alias Phoenix.Socket.{Message, V2} + alias __MODULE__.Endpoint + + @moduletag :capture_log + @port 5910 + + @session_config [store: :cookie, key: "_detached_key", signing_salt: "detached"] + + Application.put_env(:phoenix, Endpoint, + https: false, + http: [port: @port], + secret_key_base: String.duplicate("abcdefgh", 8), + server: true, + drainer: false, + pubsub_server: __MODULE__ + ) + + defmodule RoomChannel do + use Phoenix.Channel + + def join(_topic, %{"reject" => true}, _socket), do: {:error, %{reason: "rejected"}} + + # assigns of the server, see join_detached/4 + def join(_topic, _payload, %{assigns: %{notify: {pid, ref}}} = socket) do + send(pid, {ref, socket.assigns}) + {:ok, socket} + end + + def join(topic, payload, socket) do + Process.register(self(), String.to_atom(topic)) + send(self(), :after_join) + {:ok, %{payload: payload, transport: inspect(socket.transport)}, socket} + end + + def handle_info(:after_join, socket) do + push(socket, "joined", %{}) + {:noreply, socket} + end + + def handle_info({:push, event, payload}, socket) do + push(socket, event, payload) + {:noreply, socket} + end + + def handle_in("echo", payload, socket) do + push(socket, "echo", payload) + {:noreply, socket} + end + end + + defmodule UserSocket do + use Phoenix.Socket + + channel "room:*", RoomChannel, assigns: %{from_channel: true} + + def connect(params, socket, connect_info) do + send(Process.whereis(:detached_test), {:connect, params, socket, connect_info}) + + case connect_info.session do + %{"reject" => true} -> :error + session -> {:ok, assign(socket, :user_id, session && session["user_id"])} + end + end + + def resume(%{"reject_resume" => "true"}, _socket, _connect_info), do: :error + + def resume(params, socket, _connect_info) do + send(Process.whereis(:detached_test), {:resume, params, socket.transport}) + {:ok, socket} + end + + def id(socket) do + if id = socket.assigns.user_id, do: "user_sockets:#{id}" + end + + def resumable?(%{connect_info: %{session: session}}) do + is_map(session) and Map.has_key?(session, "user_id") + end + end + + defmodule Endpoint do + use Phoenix.Endpoint, otp_app: :phoenix + + @session_config [store: :cookie, key: "_detached_key", signing_salt: "detached"] + + socket "/ws", UserSocket, + websocket: [resumable: true, window_ms: 200, connect_info: [:uri, session: @session_config]], + longpoll: [window_ms: 200, pubsub_timeout_ms: 200, connect_info: [session: @session_config]] + + socket "/ws/plain", UserSocket, websocket: [connect_info: [session: @session_config]] + + socket "/ws/longpoll", UserSocket, + websocket: false, + longpoll: [window_ms: 200, pubsub_timeout_ms: 200, connect_info: [session: @session_config]] + + socket "/ws/users", UserSocket, + websocket: [ + resumable: {UserSocket, :resumable?, []}, + window_ms: 200, + connect_info: [session: @session_config] + ] + end + + setup do + Process.register(self(), :detached_test) + capture_log(fn -> start_supervised!(Endpoint) end) + start_supervised!({Phoenix.PubSub, name: __MODULE__}) + :ok + end + + # the request of a page, with the session of its cookie + defp page_conn(session \\ %{"user_id" => "1"}) do + Plug.Test.conn(:get, "https://example.com/page") + |> Plug.Conn.put_private(:phoenix_endpoint, Endpoint) + |> Plug.Test.init_test_session(session) + end + + defp attach(token, ack \\ 0, params \\ %{}) do + query = + URI.encode_query( + Map.merge(%{"vsn" => "2.0.0", "phx:resumable" => "1", "phx:resume" => "1"}, params) + ) + + url = "ws://127.0.0.1:#{@port}/ws/websocket?" <> query + {:ok, client} = WebsocketClient.connect(self(), url, V2.JSONSerializer) + + json = + Phoenix.json_library().encode!([ + nil, + nil, + "phoenix", + "phx_resume", + %{token: token, ack: ack} + ]) + + :ok = WebsocketClient.send(client, {:text, json}) + client + end + + defp poll(method, token, params, path, body \\ nil) do + query = URI.encode_query(Map.merge(%{"vsn" => "2.0.0"}, params)) + url = "http://127.0.0.1:#{@port}#{path}/longpoll?" <> query + headers = %{"x-phoenix-longpoll-token" => token, "content-type" => "application/x-ndjson"} + {:ok, resp} = HTTPClient.request(method, url, headers, body || "") + Phoenix.json_library().decode!(resp.body) + end + + defp decode(messages) do + for message <- messages, do: V2.JSONSerializer.decode!(message, opcode: :text) + end + + test "joins channels before a WebSocket attaches to the session" do + assert {:ok, session, token} = Phoenix.Socket.start_detached(page_conn(), "/ws") + + assert_receive {:connect, %{"vsn" => "2.0.0"}, %Phoenix.Socket{transport: :detached}, + connect_info} + + assert connect_info.session == %{"user_id" => "1"} + assert connect_info.uri.host == "example.com" + + assert {:ok, join_ref} = Phoenix.Socket.join_detached(session, "room:a", %{"n" => 1}) + channel = Process.whereis(:"room:a") + send(channel, {:push, "before_attach", %{}}) + + client = attach(token) + assert_receive {:resume, %{"vsn" => "2.0.0"}, :websocket} + assert_receive %Message{event: "phx_resumed", payload: %{"seq" => 0, "last" => 3}} + + assert_receive %Message{ + event: "phx_reply", + topic: "room:a", + ref: ^join_ref, + join_ref: ^join_ref, + payload: %{ + "status" => "ok", + "response" => %{"payload" => %{"n" => 1}, "transport" => ":detached"} + } + } + + assert_receive %Message{event: "joined", topic: "room:a", join_ref: ^join_ref} + assert_receive %Message{event: "before_attach", topic: "room:a", join_ref: ^join_ref} + + # the client sends with the join ref of the server + json = Phoenix.json_library().encode!([join_ref, "1", "room:a", "echo", %{"hi" => true}]) + :ok = WebsocketClient.send(client, {:text, json}) + assert_receive %Message{event: "echo", payload: %{"hi" => true}} + assert Process.whereis(:"room:a") == channel + end + + test "long polling attaches to the session" do + assert {:ok, session, token} = Phoenix.Socket.start_detached(page_conn(), "/ws/longpoll") + assert {:ok, join_ref} = Phoenix.Socket.join_detached(session, "room:a", %{}) + + body = poll(:get, token, %{"phx:ack" => "0", "phx:resume" => "1"}, "/ws/longpoll") + assert body["status"] == 200 + assert_receive {:resume, _params, :longpoll} + + assert [ + %Message{event: "phx_reply", ref: ^join_ref, payload: %{"status" => "ok"}}, + %Message{event: "joined"} + ] = decode(body["messages"]) + end + + test "long polling must attach before it polls" do + assert {:ok, _session, token} = Phoenix.Socket.start_detached(page_conn(), "/ws/longpoll") + assert poll(:get, token, %{"phx:ack" => "0"}, "/ws/longpoll")["status"] == 410 + end + + test "stops the session when it rejects the client" do + assert {:ok, session, token} = Phoenix.Socket.start_detached(page_conn(), "/ws") + ref = Process.monitor(session) + assert {:ok, _join_ref} = Phoenix.Socket.join_detached(session, "room:a", %{}) + channel = Process.monitor(Process.whereis(:"room:a")) + + _client = attach(token, 0, %{"reject_resume" => "true"}) + assert_receive %Message{event: "phx_gone"} + assert_receive {:DOWN, ^ref, _, _, _} + assert_receive {:DOWN, ^channel, _, _, _} + end + + test "stops the session when no client attaches within its window" do + assert {:ok, session, _token} = + Phoenix.Socket.start_detached(page_conn(), "/ws", window_ms: 20) + + ref = Process.monitor(session) + assert {:ok, _join_ref} = Phoenix.Socket.join_detached(session, "room:a", %{}) + channel = Process.monitor(Process.whereis(:"room:a")) + assert_receive {:DOWN, ^ref, _, _, {:shutdown, :inactive}}, 200 + assert_receive {:DOWN, ^channel, _, _, _} + end + + test "stops a session" do + assert {:ok, session, _token} = Phoenix.Socket.start_detached(page_conn(), "/ws") + assert {:ok, _join_ref} = Phoenix.Socket.join_detached(session, "room:a", %{}) + channel = Process.monitor(Process.whereis(:"room:a")) + assert :ok = Phoenix.Socket.stop_detached(session) + refute Process.alive?(session) + assert_receive {:DOWN, ^channel, _, _, _} + assert :ok = Phoenix.Socket.stop_detached(session) + end + + test "numbers the join refs of the session with a prefix" do + assert {:ok, session, _token} = Phoenix.Socket.start_detached(page_conn(), "/ws") + assert {:ok, "phx-1"} = Phoenix.Socket.join_detached(session, "room:a", %{}) + assert {:ok, "phx-2"} = Phoenix.Socket.join_detached(session, "room:b", %{}) + + assert {:ok, other, _token} = Phoenix.Socket.start_detached(page_conn(), "/ws") + assert {:ok, "phx-1"} = Phoenix.Socket.join_detached(other, "room:c", %{}) + end + + test "returns the errors of joins" do + assert {:ok, session, _token} = Phoenix.Socket.start_detached(page_conn(), "/ws") + + assert {:error, %{reason: "rejected"}} = + Phoenix.Socket.join_detached(session, "room:a", %{"reject" => true}) + + assert {:error, %{reason: "unmatched topic"}} = + Phoenix.Socket.join_detached(session, "unknown", %{}) + + assert {:ok, _join_ref} = Phoenix.Socket.join_detached(session, "room:a", %{}) + + assert {:error, %{reason: "already joined"}} = + Phoenix.Socket.join_detached(session, "room:a", %{}) + end + + test "merges the :assigns of the join into the socket of the channel" do + assert {:ok, session, _token} = Phoenix.Socket.start_detached(page_conn(), "/ws") + ref = make_ref() + + assert {:ok, _join_ref} = + Phoenix.Socket.join_detached(session, "room:a", %{}, + assigns: %{notify: {self(), ref}} + ) + + assert_received {^ref, assigns} + assert %{from_channel: true, user_id: "1", notify: {_, ^ref}} = assigns + end + + test "raises when the :assigns of the join are not a map" do + assert {:ok, session, _token} = Phoenix.Socket.start_detached(page_conn(), "/ws") + + assert_raise ArgumentError, ~r/:assigns must be a map/, fn -> + Phoenix.Socket.join_detached(session, "room:a", %{}, assigns: [notify: true]) + end + end + + test "does not start more sessions without a client than :max_pending" do + # the sessions of other tests stop after their window + assert eventually(fn -> Phoenix.Socket.detached_count() == 0 end) + max = 2 + + assert {:ok, first, token} = + Phoenix.Socket.start_detached(page_conn(), "/ws", max_pending: max) + + assert {:ok, second, _token} = + Phoenix.Socket.start_detached(page_conn(), "/ws", max_pending: max) + + assert Phoenix.Socket.detached_count() == max + assert_received {:connect, _, _, _} + assert_received {:connect, _, _, _} + assert {:error, :limit} = Phoenix.Socket.start_detached(page_conn(), "/ws", max_pending: max) + refute_received {:connect, _, _, _} + + # sessions stop counting once a client attached or they stopped + _client = attach(token) + assert_receive %Message{event: "phx_resumed"} + assert Phoenix.Socket.detached_count() == max - 1 + assert Process.alive?(first) + + ref = Process.monitor(second) + Process.exit(second, :kill) + assert_receive {:DOWN, ^ref, _, _, _} + # the registry removes the session once it handled the exit + assert eventually(fn -> Phoenix.Socket.detached_count() == max - 2 end) + end + + defp eventually(fun, tries \\ 50) do + cond do + fun.() -> + true + + tries == 0 -> + false + + true -> + Process.sleep(10) + eventually(fun, tries - 1) + end + end + + test "raises when :max_pending is invalid" do + assert_raise ArgumentError, ~r/:max_pending must be a non-negative integer/, fn -> + Phoenix.Socket.start_detached(page_conn(), "/ws", max_pending: :infinity) + end + end + + test "requires a conn that went through an endpoint" do + assert_raise ArgumentError, ~r/requires a conn that went through an endpoint/, fn -> + Phoenix.Socket.start_detached(Plug.Test.conn(:get, "/"), "/ws") + end + end + + test "does not start a session when the socket rejects the connection" do + assert {:error, :rejected} = + Phoenix.Socket.start_detached(page_conn(%{"reject" => true}), "/ws") + end + + test "does not start a session for a WebSocket that is not resumable" do + assert {:error, :not_resumable} = Phoenix.Socket.start_detached(page_conn(), "/ws/plain") + refute_received {:connect, _, _, _} + end + + test "lets the :resumable function decide" do + assert {:ok, _session, _token} = Phoenix.Socket.start_detached(page_conn(), "/ws/users") + assert {:error, :denied} = Phoenix.Socket.start_detached(page_conn(%{}), "/ws/users") + end + + test "does not start a session without a socket at the path" do + assert {:error, :no_socket} = Phoenix.Socket.start_detached(page_conn(), "/unknown") + end + + test "uses the session of the cookie when the request did not fetch it" do + cookie_conn = + Plug.Test.conn(:get, "/page") + |> Map.put(:secret_key_base, Endpoint.config(:secret_key_base)) + |> Plug.Session.call(Plug.Session.init(@session_config)) + |> Plug.Conn.fetch_session() + |> Plug.Conn.put_session("user_id", "3") + |> Plug.Conn.send_resp(200, "") + + cookie = cookie_conn.resp_cookies["_detached_key"].value + + # a request with a session cookie, which it did not fetch + conn = + Plug.Test.conn(:get, "/page") + |> Plug.Conn.put_private(:phoenix_endpoint, Endpoint) + |> Plug.Test.put_req_cookie("_detached_key", cookie) + + assert {:ok, _session, _token} = Phoenix.Socket.start_detached(conn, "/ws") + assert_receive {:connect, _params, _socket, %{session: %{"user_id" => "3"}}} + end +end diff --git a/test/phoenix/integration/websocket_resume_test.exs b/test/phoenix/integration/websocket_resume_test.exs index 497773e598..aaf2c6d506 100644 --- a/test/phoenix/integration/websocket_resume_test.exs +++ b/test/phoenix/integration/websocket_resume_test.exs @@ -415,6 +415,32 @@ defmodule Phoenix.Integration.WebSocketResumeTest do refute match?({:pipe, _, _}, :sys.get_state(session(token)).client) end + test "answers heartbeats without numbering them" do + {client, token} = new_session() + :ok = WebsocketClient.send(client, {:text, ~s([null,"5","phoenix","heartbeat",{}])}) + + assert_receive %Message{ + topic: "phoenix", + event: "phx_reply", + ref: "5", + payload: %{"status" => "ok"} + } + + # the session did not receive a numbered message, so the pipe does not confirm it + refute_receive %Message{event: "phx_ack"}, 50 + + join(client, "room:a") + assert_receive %Message{event: "phx_ack", payload: %{"seq" => 1}} + drop(client) + + # the client replays neither the heartbeat nor its reply + _client = resume(token, 0) + assert_receive %Message{event: "phx_resumed", payload: %{"seq" => 1, "last" => 2}} + assert_receive %Message{event: "phx_reply", topic: "room:a"} + assert_receive %Message{event: "joined", topic: "room:a"} + refute_receive %Message{topic: "phoenix", event: "phx_reply"}, 50 + end + test "passes control frames to the socket without numbering them" do {client, token} = new_ping_session()