Add Phoenix.Socket.start_detached/3

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

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)

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 => {

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()
   }
 

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) }