Fixes deadlock on publish track (#604)

* Fixes deadlock on publish track

During a publish track, if the room is leaved before the the track is published, we can't publish a track anymore. Let's abord the pending publish track on leave event and when the websocket is disconnected

* Add changeset

* Lint fixes

+---
+"client-sdk-android": patch
+---
+
+Fixes deadlock on publish track

@@ -86,6 +86,7 @@ import javax.inject.Named
 import javax.inject.Singleton
 import javax.inject.Singleton
 import kotlin.coroutines.Continuation
 import kotlin.coroutines.Continuation
 import kotlin.coroutines.resume
 import kotlin.coroutines.resume
+import kotlin.coroutines.resumeWithException
 import kotlin.coroutines.suspendCoroutine
 import kotlin.coroutines.suspendCoroutine
 
 
 /**
 /**
@@ -324,12 +325,17 @@ internal constructor(
         stream: String?,
         stream: String?,
         builder: LivekitRtc.AddTrackRequest.Builder = LivekitRtc.AddTrackRequest.newBuilder(),
         builder: LivekitRtc.AddTrackRequest.Builder = LivekitRtc.AddTrackRequest.newBuilder(),
     ): LivekitModels.TrackInfo {
     ): LivekitModels.TrackInfo {
-        if (pendingTrackResolvers[cid] != null) {
-            throw TrackException.DuplicateTrackException("Track with same ID $cid has already been published!")
+        synchronized(pendingTrackResolvers) {
+            if (pendingTrackResolvers[cid] != null) {
+                throw TrackException.DuplicateTrackException("Track with same ID $cid has already been published!")
+            }
         }
         }
+
         // Suspend until signal client receives message confirming track publication.
         // Suspend until signal client receives message confirming track publication.
         return suspendCoroutine { cont ->
         return suspendCoroutine { cont ->
-            pendingTrackResolvers[cid] = cont
+            synchronized(pendingTrackResolvers) {
+                pendingTrackResolvers[cid] = cont
+            }
             client.sendAddTrack(
             client.sendAddTrack(
                 cid = cid,
                 cid = cid,
                 name = name,
                 name = name,
@@ -380,6 +386,7 @@ internal constructor(
         lastRoomOptions = null
         lastRoomOptions = null
         participantSid = null
         participantSid = null
         regionUrlProvider = null
         regionUrlProvider = null
+        abortPendingPublishTracks()
         closeResources(reason)
         closeResources(reason)
         connectionState = ConnectionState.DISCONNECTED
         connectionState = ConnectionState.DISCONNECTED
     }
     }
@@ -416,6 +423,15 @@ internal constructor(
         client.close(reason = reason)
         client.close(reason = reason)
     }
     }
 
 
+    private fun abortPendingPublishTracks() {
+        synchronized(pendingTrackResolvers) {
+            pendingTrackResolvers.values.forEach {
+                it.resumeWithException(TrackException.PublishException("pending track aborted"))
+            }
+            pendingTrackResolvers.clear()
+        }
+    }
+
     /**
     /**
      * reconnect Signal and PeerConnections
      * reconnect Signal and PeerConnections
      */
      */
@@ -907,7 +923,9 @@ internal constructor(
         }
         }
 
 
         LKLog.v { "local track published $cid" }
         LKLog.v { "local track published $cid" }
-        val cont = pendingTrackResolvers.remove(cid)
+        val cont = synchronized(pendingTrackResolvers) {
+            pendingTrackResolvers.remove(cid)
+        }
         if (cont == null) {
         if (cont == null) {
             LKLog.d { "missing track resolver for: $cid" }
             LKLog.d { "missing track resolver for: $cid" }
             return
             return
@@ -929,6 +947,7 @@ internal constructor(
 
 
     override fun onClose(reason: String, code: Int) {
     override fun onClose(reason: String, code: Int) {
         LKLog.i { "received close event: $reason, code: $code" }
         LKLog.i { "received close event: $reason, code: $code" }
+        abortPendingPublishTracks()
         reconnect()
         reconnect()
     }
     }
 
 
@@ -946,6 +965,9 @@ internal constructor(
 
 
     override fun onLeave(leave: LeaveRequest) {
     override fun onLeave(leave: LeaveRequest) {
         LKLog.d { "leave request received: reason = ${leave.reason.name}" }
         LKLog.d { "leave request received: reason = ${leave.reason.name}" }
+
+        abortPendingPublishTracks()
+
         if (leave.hasRegions()) {
         if (leave.hasRegions()) {
             regionUrlProvider?.let {
             regionUrlProvider?.let {
                 it.setServerReportedRegions(RegionSettings.fromProto(leave.regions))
                 it.setServerReportedRegions(RegionSettings.fromProto(leave.regions))

@@ -482,13 +482,19 @@ internal constructor(
         val builder = AddTrackRequest.newBuilder().apply {
         val builder = AddTrackRequest.newBuilder().apply {
             this.requestConfig()
             this.requestConfig()
         }
         }
-        val trackInfo = engine.addTrack(
-            cid = cid,
-            name = options.name ?: track.name,
-            kind = track.kind.toProto(),
-            stream = options.stream,
-            builder = builder,
-        )
+
+        val trackInfo = try {
+            engine.addTrack(
+                cid = cid,
+                name = options.name ?: track.name,
+                kind = track.kind.toProto(),
+                stream = options.stream,
+                builder = builder,
+            )
+        } catch (e: Exception) {
+            publishListener?.onPublishFailure(TrackException.PublishException("Failed to publish track", e))
+            return null
+        }
 
 
         if (options is VideoTrackPublishOptions) {
         if (options is VideoTrackPublishOptions) {
             // server might not support the codec the client has requested, in that case, fallback
             // server might not support the codec the client has requested, in that case, fallback