Refactored channel rejoins

Author: Daniel Rees

src/main/kotlin/org/phoenixframework/Channel.kt

@@ -127,8 +127,17 @@ class Channel(
     this.pushBuffer = mutableListOf()
     this.rejoinTimer = TimeoutTimer(
         dispatchQueue = socket.dispatchQueue,
-        callback = { rejoinUntilConnected() },
-        timerCalculation = socket.reconnectAfterMs)
+        timerCalculation = socket.rejoinAfterMs,
+        callback = { if (socket.isConnected) rejoin() }
+    )
+
+    // Respond to socket events
+    this.socket.onError { _, _-> this.rejoinTimer.reset() }
+    this.socket.onOpen {
+      this.rejoinTimer.reset()
+      if (this.isErrored) { this.rejoin() }
+    }
+
 
     // Setup Push to be sent when joining
     this.joinPush = Push(
@@ -150,12 +159,15 @@ class Channel(
       this.pushBuffer.clear()
     }
 
+    // Perform if Channel errors while attempting to join
+    this.joinPush.receive("error") {
+      this.state = State.ERRORED
+      if (this.socket.isConnected) { this.rejoinTimer.scheduleTimeout() }
+    }
+
     // Perform if Channel timed out while attempting to join
     this.joinPush.receive("timeout") {
-
-      // Only handle a timeout if the Channel is in the 'joining' state
-      if (!this.isJoining) return@receive
-
+      // Log the timeout
       this.socket.logItems("Channel: timeouts $topic, $joinRef after $timeout ms")
 
       // Send a Push to the server to leave the Channel
@@ -165,10 +177,11 @@ class Channel(
           timeout = this.timeout)
       leavePush.send()
 
-      // Mark the Channel as in an error and attempt to rejoin
+      // Mark the Channel as in an error and attempt to rejoin if socket is connected
       this.state = State.ERRORED
       this.joinPush.reset()
-      this.rejoinTimer.scheduleTimeout()
+
+      if (this.socket.isConnected) { this.rejoinTimer.scheduleTimeout() }
     }
 
     // Clean up when the channel closes
@@ -177,7 +190,7 @@ class Channel(
       this.rejoinTimer.reset()
 
       // Log that the channel was left
-      this.socket.logItems("Channel: close $topic")
+      this.socket.logItems("Channel: close $topic $joinRef")
 
       // Mark the channel as closed and remove it from the socket
       this.state = State.CLOSED
@@ -186,16 +199,15 @@ class Channel(
 
     // Handles an error, attempts to rejoin
     this.onError {
-      // Do not emit error if the channel is in the process of leaving
-      // or if it has already closed
-      if (this.isLeaving || this.isClosed) return@onError
-
       // Log that the channel received an error
-      this.socket.logItems("Channel: error $topic")
+      this.socket.logItems("Channel: error $topic ${it.payload}")
+
+      // If error was received while joining, then reset the Push
+      if (isJoining) { this.joinPush.reset() }
 
-      // Mark the channel as errored and attempt to rejoin
+      // Mark the channel as errored and attempt to rejoin if socket is currently connected
       this.state = State.ERRORED
-      this.rejoinTimer.scheduleTimeout()
+      if (socket.isConnected) { this.rejoinTimer.scheduleTimeout() }
     }
 
     // Perform when the join reply is received
@@ -245,8 +257,9 @@ class Channel(
     }
 
     // Join the channel
+    this.timeout = timeout
     this.joinedOnce = true
-    this.rejoin(timeout)
+    this.rejoin()
     return joinPush
   }
 
@@ -304,6 +317,9 @@ class Channel(
     // will return false, so instead store it _before_ starting the leave
     val canPush = this.canPush
 
+    // If attempting a rejoin during a leave, then reset, cancelling the rejoin
+    this.rejoinTimer.reset()
+
     // Now set the state to leaving
     this.state = State.LEAVING
 
@@ -384,12 +400,6 @@ class Channel(
   //------------------------------------------------------------------------------
   // Private
   //------------------------------------------------------------------------------
-  /** Will continually attempt to rejoin the Channel on a timer. */
-  private fun rejoinUntilConnected() {
-    this.rejoinTimer.scheduleTimeout()
-    if (this.socket.isConnected) this.rejoin()
-  }
-
   /** Sends the Channel's joinPush to the Server */
   private fun sendJoin(timeout: Long) {
     this.state = State.JOINING
@@ -398,6 +408,10 @@ class Channel(
 
   /** Rejoins the Channel e.g. after a disconnect */
   private fun rejoin(timeout: Long = this.timeout) {
+    // Do not attempt to rejoin if the channel is in the process of leaving
+    if (isLeaving) return
+
+    // Send the joinPush
     this.sendJoin(timeout)
   }
 }

src/main/kotlin/org/phoenixframework/Defaults.kt

@@ -34,11 +34,6 @@ object Defaults {
   /** Default heartbeat interval of 30s */
   const val HEARTBEAT: Long = 30_000
 
-  /** Default reconnect algorithm. Reconnects after 1s, 2s, 5s and then 10s thereafter */
-  val steppedBackOff: (Int) -> Long = { tries ->
-    if (tries > 3) 10_000 else listOf(1_000L, 2_000L, 5_000L)[tries - 1]
-  }
-
   /** Default reconnect algorithm for the socket */
   val reconnectSteppedBackOff: (Int) -> Long = { tries ->
     if (tries > 9) 5_000 else listOf(10L, 50L, 100L, 150L, 200L, 250L, 500L, 1_000L, 2_000L)[tries - 1]

src/test/kotlin/org/phoenixframework/ChannelTest.kt

@@ -22,14 +22,19 @@ import org.phoenixframework.utilities.getBindings
 
 class ChannelTest {
 
+  companion object {
+    const val DEFAULT_REF = "1"
+    const val DEFAULT_TIMEOUT = 10_000L
+  }
+
+
   @Mock lateinit var socket: Socket
   @Mock lateinit var mockCallback: ((Message) -> Unit)
 
   private val kDefaultRef = "1"
   private val kDefaultTimeout = 10_000L
   private val kDefaultPayload: Payload = mapOf("one" to "two")
   private val kEmptyPayload: Payload = mapOf()
-  private val reconnectAfterMs: (Int) -> Long = Defaults.steppedBackOff
 
   lateinit var fakeClock: ManualDispatchQueue
   lateinit var channel: Channel
@@ -50,7 +55,8 @@ class ChannelTest {
     whenever(socket.dispatchQueue).thenReturn(fakeClock)
     whenever(socket.makeRef()).thenReturn(kDefaultRef)
     whenever(socket.timeout).thenReturn(kDefaultTimeout)
-    whenever(socket.reconnectAfterMs).thenReturn(reconnectAfterMs)
+    whenever(socket.reconnectAfterMs).thenReturn(Defaults.reconnectSteppedBackOff)
+    whenever(socket.rejoinAfterMs).thenReturn(Defaults.rejoinSteppedBackOff)
 
     channel = Channel("topic", kDefaultPayload, socket)
   }

src/test/kotlin/org/phoenixframework/PresenceTest.kt

@@ -38,6 +38,7 @@ class PresenceTest {
     whenever(socket.timeout).thenReturn(Defaults.TIMEOUT)
     whenever(socket.makeRef()).thenReturn("1")
     whenever(socket.reconnectAfterMs).thenReturn { 1_000 }
+    whenever(socket.rejoinAfterMs).thenReturn(Defaults.rejoinSteppedBackOff)
     whenever(socket.dispatchQueue).thenReturn(mock())
 
     channel = Channel("topic", mapOf(), socket)