fix: 1. Reset connectionId for coordinator socket when connection is disconnected (#1582)

2. Coordinator socket will execute its task serially

Author: rahul-lohra

stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/socket/common/scope/UserScope.kt

@@ -23,6 +23,8 @@ import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Job
 import kotlinx.coroutines.job
 import kotlinx.coroutines.plus
+import kotlin.coroutines.CoroutineContext
+import kotlin.coroutines.EmptyCoroutineContext
 
 /**
  * A user aware implementation of [CoroutineScope].
@@ -38,22 +40,29 @@ internal interface UserScope : CoroutineScope {
      * Cancels all children of the [UserJob] in this scope's context connected to the specified [userId].
      */
     fun cancelChildren(userId: UserId? = null)
+
+    operator fun plus(context: CoroutineContext): UserScope
 }
 
 /**
  * Creates a user aware [CoroutineScope].
  */
-internal fun UserScope(clientScope: ClientScope): UserScope = UserScopeImpl(clientScope)
+internal fun UserScope(
+    clientScope: ClientScope = ClientScope(),
+    context: CoroutineContext = EmptyCoroutineContext,
+): UserScope = UserScopeImpl(clientScope, context)
 
 /**
  * Inherits [ClientScope] and adds elements such as [UserIdentifier] and [UserJob].
  */
 private class UserScopeImpl(
-    clientScope: ClientScope,
-    userIdentifier: UserIdentifier = UserIdentifier(),
+    val clientScope: ClientScope,
+    additionalContext: CoroutineContext,
+    val userIdentifier: UserIdentifier = UserIdentifier(),
 ) : UserScope,
     CoroutineScope by (
-        clientScope + userIdentifier + UserJob(clientScope.coroutineContext.job) { userIdentifier.value }
+        clientScope + userIdentifier + UserJob(clientScope.coroutineContext.job) { userIdentifier.value } +
+            additionalContext
         ) {
 
     /**
@@ -68,4 +77,7 @@ private class UserScopeImpl(
     override fun cancelChildren(userId: UserId?) {
         (coroutineContext[Job] as UserJob).cancelChildren(userId)
     }
+
+    override fun plus(context: CoroutineContext): UserScope =
+        UserScopeImpl(clientScope, this.coroutineContext + context, userIdentifier)
 }

stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/socket/coordinator/CoordinatorSocketConnection.kt

@@ -46,6 +46,7 @@ import io.getstream.video.android.model.User
 import io.getstream.video.android.model.User.Companion.isAnonymous
 import io.getstream.video.android.model.UserToken
 import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.channels.BufferOverflow
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.MutableSharedFlow
@@ -109,7 +110,7 @@ public open class CoordinatorSocketConnection(
             parser,
             httpClient,
         ),
-        scope as? UserScope ?: UserScope(ClientScope()),
+        scope as? UserScope ?: UserScope(ClientScope(), Dispatchers.IO.limitedParallelism(1)),
         StreamLifecycleObserver(scope, lifecycle),
         networkStateProvider,
     ).also {
@@ -155,6 +156,7 @@ public open class CoordinatorSocketConnection(
 
     override fun onConnecting() {
         super.onConnecting()
+        connectionId.value = null
         logger.d { "[onConnecting] Socket is connecting" }
     }
 
@@ -186,6 +188,7 @@ public open class CoordinatorSocketConnection(
 
     override fun onDisconnected(cause: DisconnectCause) {
         super.onDisconnected(cause)
+        connectionId.value = null
         logger.d { "[onDisconnected] Socket disconnected. Cause: $cause" }
     }