Clear buffered responses on SignalClient closure (#141)

* clear response/request flow in case anything previously unhandled remains

* Add in some verbose logging whenever SignalClient closes

* more comments and disable request buffer clearing

@@ -294,7 +294,7 @@ internal constructor(
         client.sendMuteTrack(sid, muted)
         client.sendMuteTrack(sid, muted)
     }
     }
 
 
-    fun close() {
+    fun close(reason: String = "Normal Closure") {
         if (isClosed) {
         if (isClosed) {
             return
             return
         }
         }
@@ -307,10 +307,10 @@ internal constructor(
         reconnectingJob?.cancel()
         reconnectingJob?.cancel()
         reconnectingJob = null
         reconnectingJob = null
         coroutineScope.close()
         coroutineScope.close()
-        closeResources()
+        closeResources(reason)
     }
     }
 
 
-    private fun closeResources() {
+    private fun closeResources(reason: String) {
         connectionState = ConnectionState.DISCONNECTED
         connectionState = ConnectionState.DISCONNECTED
         _publisher?.close()
         _publisher?.close()
         _publisher = null
         _publisher = null
@@ -325,7 +325,7 @@ internal constructor(
         lossyDataChannelSub?.close()
         lossyDataChannelSub?.close()
         lossyDataChannelSub = null
         lossyDataChannelSub = null
         isSubscriberPrimary = false
         isSubscriberPrimary = false
-        client.close()
+        client.close(reason = reason)
     }
     }
 
 
     /**
     /**
@@ -372,7 +372,7 @@ internal constructor(
                 if (isFullReconnect) {
                 if (isFullReconnect) {
                     LKLog.v { "Attempting full reconnect." }
                     LKLog.v { "Attempting full reconnect." }
                     try {
                     try {
-                        closeResources()
+                        closeResources("Full Reconnecting")
                         listener?.onFullReconnecting()
                         listener?.onFullReconnecting()
                         joinImpl(url, token, connectOptions ?: ConnectOptions(), lastRoomOptions ?: RoomOptions())
                         joinImpl(url, token, connectOptions ?: ConnectOptions(), lastRoomOptions ?: RoomOptions())
                     } catch (e: Exception) {
                     } catch (e: Exception) {
@@ -437,7 +437,7 @@ internal constructor(
                 }
                 }
             }
             }
 
 
-            close()
+            close("Failed reconnecting")
             listener?.onEngineDisconnected("failed reconnecting.")
             listener?.onEngineDisconnected("failed reconnecting.")
         }
         }
 
 

@@ -34,6 +34,7 @@ import javax.inject.Singleton
  * SignalClient to LiveKit WS servers
  * SignalClient to LiveKit WS servers
  * @suppress
  * @suppress
  */
  */
+@OptIn(ExperimentalCoroutinesApi::class)
 @Singleton
 @Singleton
 class SignalClient
 class SignalClient
 @Inject
 @Inject
@@ -99,7 +100,7 @@ constructor(
         roomOptions: RoomOptions
         roomOptions: RoomOptions
     ): Either<LivekitRtc.JoinResponse, Unit> {
     ): Either<LivekitRtc.JoinResponse, Unit> {
         // Clean up any pre-existing connection.
         // Clean up any pre-existing connection.
-        close()
+        close(reason = "Starting new connection")
 
 
         val wsUrlString = "$url/rtc" + createConnectionParams(token, getClientInfo(), options, roomOptions)
         val wsUrlString = "$url/rtc" + createConnectionParams(token, getClientInfo(), options, roomOptions)
         isReconnecting = options.reconnect
         isReconnecting = options.reconnect
@@ -163,7 +164,6 @@ constructor(
      *
      *
      * Should be called after resolving the join message.
      * Should be called after resolving the join message.
      */
      */
-    @OptIn(ExperimentalCoroutinesApi::class)
     fun onReadyForResponses() {
     fun onReadyForResponses() {
         coroutineScope.launch {
         coroutineScope.launch {
             responseFlow.collect {
             responseFlow.collect {
@@ -561,6 +561,7 @@ constructor(
      * Can be reused afterwards.
      * Can be reused afterwards.
      */
      */
     fun close(code: Int = 1000, reason: String = "Normal Closure") {
     fun close(code: Int = 1000, reason: String = "Normal Closure") {
+        LKLog.v(Exception()) { "Closing SignalClient: code = $code, reason = $reason" }
         isConnected = false
         isConnected = false
         isReconnecting = false
         isReconnecting = false
         requestFlowJob = null
         requestFlowJob = null
@@ -571,6 +572,9 @@ constructor(
         currentWs = null
         currentWs = null
         joinContinuation?.cancel()
         joinContinuation?.cancel()
         joinContinuation = null
         joinContinuation = null
+        // TODO: support calling this from connect without wiping any queued requests.
+        //requestFlow.resetReplayCache()
+        responseFlow.resetReplayCache()
         lastUrl = null
         lastUrl = null
         lastOptions = null
         lastOptions = null
         lastRoomOptions = null
         lastRoomOptions = null