Reconnect logic improvements, based on the connection state and/or backend API errors

stream-video-android-core/api/stream-video-android-core.api

@@ -6501,8 +6501,8 @@ public abstract interface class io/getstream/video/android/core/call/audio/Input
 }
 
 public class io/getstream/video/android/core/call/connection/StreamPeerConnection : org/webrtc/PeerConnection$Observer {
-	public fun <init> (Lkotlinx/coroutines/CoroutineScope;Lio/getstream/video/android/core/model/StreamPeerType;Lorg/webrtc/MediaConstraints;Lkotlin/jvm/functions/Function1;Lkotlin/jvm/functions/Function2;Lkotlin/jvm/functions/Function2;IZLio/getstream/video/android/core/trace/Tracer;)V
-	public synthetic fun <init> (Lkotlinx/coroutines/CoroutineScope;Lio/getstream/video/android/core/model/StreamPeerType;Lorg/webrtc/MediaConstraints;Lkotlin/jvm/functions/Function1;Lkotlin/jvm/functions/Function2;Lkotlin/jvm/functions/Function2;IZLio/getstream/video/android/core/trace/Tracer;ILkotlin/jvm/internal/DefaultConstructorMarker;)V
+	public fun <init> (Lkotlinx/coroutines/CoroutineScope;Lio/getstream/video/android/core/model/StreamPeerType;Lorg/webrtc/MediaConstraints;Lkotlin/jvm/functions/Function1;Lkotlin/jvm/functions/Function2;Lkotlin/jvm/functions/Function2;Lkotlin/jvm/functions/Function0;IZLio/getstream/video/android/core/trace/Tracer;)V
+	public synthetic fun <init> (Lkotlinx/coroutines/CoroutineScope;Lio/getstream/video/android/core/model/StreamPeerType;Lorg/webrtc/MediaConstraints;Lkotlin/jvm/functions/Function1;Lkotlin/jvm/functions/Function2;Lkotlin/jvm/functions/Function2;Lkotlin/jvm/functions/Function0;IZLio/getstream/video/android/core/trace/Tracer;ILkotlin/jvm/internal/DefaultConstructorMarker;)V
 	public final fun addAudioTransceiver (Lorg/webrtc/MediaStreamTrack;Ljava/util/List;)V
 	public final fun addIceCandidate (Lio/getstream/video/android/core/model/IceCandidate;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 	public final fun addVideoTransceiver (Lorg/webrtc/MediaStreamTrack;Ljava/util/List;Z)V
@@ -6560,6 +6560,14 @@ public final class io/getstream/video/android/core/call/connection/StreamPeerCon
 	public final fun toggleAudioProcessing ()Z
 }
 
+public class io/getstream/video/android/core/call/connection/job/RestartIceJobDelegate {
+	public fun <init> (Lkotlinx/coroutines/CoroutineScope;J)V
+	public synthetic fun <init> (Lkotlinx/coroutines/CoroutineScope;JILkotlin/jvm/internal/DefaultConstructorMarker;)V
+	public final fun cancelScheduledRestartIce ()V
+	public final fun scheduleRestartIce (JLkotlin/jvm/functions/Function1;)V
+	public static synthetic fun scheduleRestartIce$default (Lio/getstream/video/android/core/call/connection/job/RestartIceJobDelegate;JLkotlin/jvm/functions/Function1;ILjava/lang/Object;)V
+}
+
 public final class io/getstream/video/android/core/call/signal/socket/RTCEventMapper {
 	public static final field INSTANCE Lio/getstream/video/android/core/call/signal/socket/RTCEventMapper;
 	public final fun mapEvent (Lstream/video/sfu/event/SfuEvent;)Lio/getstream/video/android/core/events/SfuDataEvent;
@@ -8362,6 +8370,19 @@ public final class io/getstream/video/android/core/call/stats/model/discriminato
 	public final fun fromAlias (Ljava/lang/String;)Lio/getstream/video/android/core/call/stats/model/discriminator/RtcReportType;
 }
 
+public final class io/getstream/video/android/core/call/utils/SignalLostSignalingServiceDecorator : io/getstream/video/android/core/api/SignalServerService {
+	public fun <init> (Lio/getstream/video/android/core/api/SignalServerService;Lkotlin/jvm/functions/Function1;)V
+	public fun iceRestart (Lstream/video/sfu/signal/ICERestartRequest;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
+	public fun iceTrickle (Lstream/video/sfu/models/ICETrickle;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
+	public fun sendAnswer (Lstream/video/sfu/signal/SendAnswerRequest;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
+	public fun sendStats (Lstream/video/sfu/signal/SendStatsRequest;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
+	public fun setPublisher (Lstream/video/sfu/signal/SetPublisherRequest;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
+	public fun startNoiseCancellation (Lstream/video/sfu/signal/StartNoiseCancellationRequest;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
+	public fun stopNoiseCancellation (Lstream/video/sfu/signal/StopNoiseCancellationRequest;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
+	public fun updateMuteStates (Lstream/video/sfu/signal/UpdateMuteStatesRequest;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
+	public fun updateSubscriptions (Lstream/video/sfu/signal/UpdateSubscriptionsRequest;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
+}
+
 public final class io/getstream/video/android/core/call/utils/TrackOverridesHandlerKt {
 	public static final field ALL_PARTICIPANTS Ljava/lang/String;
 }
@@ -13704,6 +13725,7 @@ public final class stream/video/sfu/models/ErrorCode : java/lang/Enum, com/squar
 	public static final field ADAPTER Lcom/squareup/wire/ProtoAdapter;
 	public static final field Companion Lstream/video/sfu/models/ErrorCode$Companion;
 	public static final field ERROR_CODE_CALL_NOT_FOUND Lstream/video/sfu/models/ErrorCode;
+	public static final field ERROR_CODE_CALL_PARTICIPANT_LIMIT_REACHED Lstream/video/sfu/models/ErrorCode;
 	public static final field ERROR_CODE_INTERNAL_SERVER_ERROR Lstream/video/sfu/models/ErrorCode;
 	public static final field ERROR_CODE_LIVE_ENDED Lstream/video/sfu/models/ErrorCode;
 	public static final field ERROR_CODE_PARTICIPANT_MEDIA_TRANSPORT_FAILURE Lstream/video/sfu/models/ErrorCode;
@@ -13712,6 +13734,7 @@ public final class stream/video/sfu/models/ErrorCode : java/lang/Enum, com/squar
 	public static final field ERROR_CODE_PARTICIPANT_MIGRATION_FAILED Lstream/video/sfu/models/ErrorCode;
 	public static final field ERROR_CODE_PARTICIPANT_NOT_FOUND Lstream/video/sfu/models/ErrorCode;
 	public static final field ERROR_CODE_PARTICIPANT_RECONNECT_FAILED Lstream/video/sfu/models/ErrorCode;
+	public static final field ERROR_CODE_PARTICIPANT_SIGNAL_LOST Lstream/video/sfu/models/ErrorCode;
 	public static final field ERROR_CODE_PERMISSION_DENIED Lstream/video/sfu/models/ErrorCode;
 	public static final field ERROR_CODE_PUBLISH_TRACKS_MISMATCH Lstream/video/sfu/models/ErrorCode;
 	public static final field ERROR_CODE_PUBLISH_TRACK_NOT_FOUND Lstream/video/sfu/models/ErrorCode;
@@ -13929,10 +13952,10 @@ public final class stream/video/sfu/models/PublishOption : com/squareup/wire/Mes
 	public static final field ADAPTER Lcom/squareup/wire/ProtoAdapter;
 	public static final field Companion Lstream/video/sfu/models/PublishOption$Companion;
 	public fun <init> ()V
-	public fun <init> (Lstream/video/sfu/models/TrackType;Lstream/video/sfu/models/Codec;IIIILstream/video/sfu/models/VideoDimension;ILokio/ByteString;)V
-	public synthetic fun <init> (Lstream/video/sfu/models/TrackType;Lstream/video/sfu/models/Codec;IIIILstream/video/sfu/models/VideoDimension;ILokio/ByteString;ILkotlin/jvm/internal/DefaultConstructorMarker;)V
-	public final fun copy (Lstream/video/sfu/models/TrackType;Lstream/video/sfu/models/Codec;IIIILstream/video/sfu/models/VideoDimension;ILokio/ByteString;)Lstream/video/sfu/models/PublishOption;
-	public static synthetic fun copy$default (Lstream/video/sfu/models/PublishOption;Lstream/video/sfu/models/TrackType;Lstream/video/sfu/models/Codec;IIIILstream/video/sfu/models/VideoDimension;ILokio/ByteString;ILjava/lang/Object;)Lstream/video/sfu/models/PublishOption;
+	public fun <init> (Lstream/video/sfu/models/TrackType;Lstream/video/sfu/models/Codec;IIIILstream/video/sfu/models/VideoDimension;IZLokio/ByteString;)V
+	public synthetic fun <init> (Lstream/video/sfu/models/TrackType;Lstream/video/sfu/models/Codec;IIIILstream/video/sfu/models/VideoDimension;IZLokio/ByteString;ILkotlin/jvm/internal/DefaultConstructorMarker;)V
+	public final fun copy (Lstream/video/sfu/models/TrackType;Lstream/video/sfu/models/Codec;IIIILstream/video/sfu/models/VideoDimension;IZLokio/ByteString;)Lstream/video/sfu/models/PublishOption;
+	public static synthetic fun copy$default (Lstream/video/sfu/models/PublishOption;Lstream/video/sfu/models/TrackType;Lstream/video/sfu/models/Codec;IIIILstream/video/sfu/models/VideoDimension;IZLokio/ByteString;ILjava/lang/Object;)Lstream/video/sfu/models/PublishOption;
 	public fun equals (Ljava/lang/Object;)Z
 	public final fun getBitrate ()I
 	public final fun getCodec ()Lstream/video/sfu/models/Codec;
@@ -13941,6 +13964,7 @@ public final class stream/video/sfu/models/PublishOption : com/squareup/wire/Mes
 	public final fun getMax_spatial_layers ()I
 	public final fun getMax_temporal_layers ()I
 	public final fun getTrack_type ()Lstream/video/sfu/models/TrackType;
+	public final fun getUse_single_layer ()Z
 	public final fun getVideo_dimension ()Lstream/video/sfu/models/VideoDimension;
 	public fun hashCode ()I
 	public synthetic fun newBuilder ()Lcom/squareup/wire/Message$Builder;

stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/call/RtcSession.kt

@@ -30,6 +30,7 @@ import androidx.lifecycle.Lifecycle
 import io.getstream.android.video.generated.models.OwnCapability
 import io.getstream.android.video.generated.models.VideoEvent
 import io.getstream.log.taggedLogger
+import io.getstream.result.Error
 import io.getstream.result.Result
 import io.getstream.result.Result.Failure
 import io.getstream.result.Result.Success
@@ -228,6 +229,9 @@ public class RtcSession internal constructor(
             connectionTimeoutInMs = 2000L,
             userToken = sfuToken,
             lifecycle = lifecycle,
+            onSignalingLost = { error ->
+                call.debug.fastReconnect()
+            },
             tracer = sfuTracer,
         )
     },
@@ -418,6 +422,16 @@ public class RtcSession internal constructor(
                         call.state._connection.value =
                             RealtimeConnection.InProgress
 
+                    is SfuSocketState.Disconnected.WebSocketEventLost -> {
+                        _peerConnectionStates.value.let {
+                            if (publisher?.isHealthy() == true && subscriber?.isHealthy() == true) {
+                                call.fastReconnect()
+                            } else {
+                                call.rejoin()
+                            }
+                        }
+                    }
+
                     else -> {
                         // Ignore it
                     }
@@ -837,6 +851,11 @@ public class RtcSession internal constructor(
             sessionId = sessionId,
             enableStereo = clientImpl.enableStereoForSubscriber,
             tracer = subscriberTracer,
+            rejoin = {
+                coroutineScope.launch {
+                    call.rejoin()
+                }
+            },
             onIceCandidateRequest = ::sendIceCandidate,
         )
         return peerConnection
@@ -1332,6 +1351,7 @@ public class RtcSession internal constructor(
     }
 
     private suspend fun parseError(e: Throwable): Failure {
+        Error.NetworkError
         return Failure(
             io.getstream.result.Error.ThrowableError(
                 "CallClientImpl error needs to be handled",

stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/call/connection/Publisher.kt

@@ -21,6 +21,7 @@ import io.getstream.result.onErrorSuspend
 import io.getstream.video.android.core.MediaManagerImpl
 import io.getstream.video.android.core.ParticipantState
 import io.getstream.video.android.core.api.SignalServerService
+import io.getstream.video.android.core.call.connection.job.RestartIceJobDelegate
 import io.getstream.video.android.core.call.connection.stats.ComputedStats
 import io.getstream.video.android.core.call.connection.transceivers.TransceiverCache
 import io.getstream.video.android.core.call.connection.utils.OptimalVideoLayer
@@ -79,13 +80,16 @@ internal class Publisher(
     private val rejoin: () -> Unit,
     private val transceiverCache: TransceiverCache = TransceiverCache(),
     private val tracer: Tracer,
+    private val restartIceJobDelegate: RestartIceJobDelegate =
+        RestartIceJobDelegate(coroutineScope),
 ) : StreamPeerConnection(
     coroutineScope,
     type,
     mediaConstraints,
     onStreamAdded,
     onNegotiationNeeded,
     onIceCandidate,
+    rejoin,
     maxBitRate,
     true,
     tracer,
@@ -131,6 +135,29 @@ internal class Publisher(
         }
     }
 
+    override fun onIceConnectionChange(newState: PeerConnection.IceConnectionState?) {
+        super.onIceConnectionChange(newState)
+        when (newState) {
+            PeerConnection.IceConnectionState.CONNECTED -> {
+                restartIceJobDelegate.cancelScheduledRestartIce()
+            }
+
+            PeerConnection.IceConnectionState.FAILED -> {
+                restartIceJobDelegate.scheduleRestartIce {
+                    negotiate(true)
+                }
+            }
+            PeerConnection.IceConnectionState.DISCONNECTED -> {
+                restartIceJobDelegate.scheduleRestartIce(3000) {
+                    negotiate(true)
+                }
+            }
+            else -> {
+                // no-op
+            }
+        }
+    }
+
     @VisibleForTesting
     public suspend fun negotiate(iceRestart: Boolean = false) {
         if (isIceRestarting) {

stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/call/connection/StreamPeerConnection.kt

@@ -77,6 +77,7 @@ open class StreamPeerConnection(
     private val onStreamAdded: ((MediaStream) -> Unit)?,
     private val onNegotiationNeeded: ((StreamPeerConnection, StreamPeerType) -> Unit)?,
     private val onIceCandidate: ((IceCandidate, StreamPeerType) -> Unit)?,
+    private val onRejoinNeeded: () -> Unit,
     private val maxBitRate: Int,
     private val traceCreateAnswer: Boolean = true,
     private val tracer: Tracer,
@@ -525,22 +526,16 @@ open class StreamPeerConnection(
         logger.i { "[onConnectionChange] #sfu; #$typeTag; newState: $newState" }
         state.value = newState
         tracer.trace(PeerConnectionTraceKey.ON_CONNECTION_STATE_CHANGE.value, newState.name)
+        if (newState == PeerConnection.PeerConnectionState.FAILED) {
+            onRejoinNeeded()
+        }
     }
 
     // better to monitor onConnectionChange for the state
     override fun onIceConnectionChange(newState: PeerConnection.IceConnectionState?) {
         logger.i { "[onIceConnectionChange] #ice; #sfu; #$typeTag; newState: $newState" }
         iceState.value = newState
         tracer.trace(PeerConnectionTraceKey.ON_ICE_CONNECTION_STATE_CHANGE.value, newState?.name)
-        when (newState) {
-            PeerConnection.IceConnectionState.CLOSED, PeerConnection.IceConnectionState.FAILED, PeerConnection.IceConnectionState.DISCONNECTED -> {
-            }
-
-            PeerConnection.IceConnectionState.CONNECTED -> {
-            }
-
-            else -> Unit
-        }
     }
 
     fun close() {

stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/call/connection/StreamPeerConnectionFactory.kt

@@ -297,6 +297,7 @@ public class StreamPeerConnectionFactory(
             onNegotiationNeeded = onNegotiationNeeded,
             onIceCandidate = onIceCandidateRequest,
             maxBitRate = maxPublishingBitrate,
+            onRejoinNeeded = { },
             tracer = Tracer(type.toPeerType().name),
         )
         val connection = makePeerConnectionInternal(
@@ -317,12 +318,15 @@ public class StreamPeerConnectionFactory(
         enableStereo: Boolean = true,
         tracer: Tracer,
         onIceCandidateRequest: (IceCandidate, StreamPeerType) -> Unit,
+        rejoin: () -> Unit,
     ): Subscriber {
         val peerConnection = Subscriber(
             sessionId = sessionId,
             sfuClient = sfuClient,
             coroutineScope = coroutineScope,
             tracer = tracer,
+            enableStereo = enableStereo,
+            rejoin = rejoin,
             onIceCandidateRequest = onIceCandidateRequest,
         )
         val connection = makePeerConnectionInternal(

stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/call/connection/Subscriber.kt

@@ -23,6 +23,7 @@ import io.getstream.result.onErrorSuspend
 import io.getstream.video.android.core.ParticipantState
 import io.getstream.video.android.core.api.SignalServerService
 import io.getstream.video.android.core.call.TrackDimensions
+import io.getstream.video.android.core.call.connection.job.RestartIceJobDelegate
 import io.getstream.video.android.core.call.connection.stats.ComputedStats
 import io.getstream.video.android.core.call.connection.utils.wrapAPICall
 import io.getstream.video.android.core.call.utils.TrackOverridesHandler
@@ -51,6 +52,7 @@ import kotlinx.coroutines.launch
 import org.webrtc.MediaConstraints
 import org.webrtc.MediaStream
 import org.webrtc.MediaStreamTrack
+import org.webrtc.PeerConnection
 import org.webrtc.RtpTransceiver
 import org.webrtc.SessionDescription
 import stream.video.sfu.models.Participant
@@ -70,6 +72,9 @@ internal class Subscriber(
     private val coroutineScope: CoroutineScope,
     private val enableStereo: Boolean = true,
     private val tracer: Tracer,
+    private val rejoin: () -> Unit,
+    private val restartIceJobDelegate: RestartIceJobDelegate =
+        RestartIceJobDelegate(coroutineScope),
     onIceCandidateRequest: ((IceCandidate, StreamPeerType) -> Unit)?,
 ) : StreamPeerConnection(
     coroutineScope = coroutineScope,
@@ -83,6 +88,7 @@ internal class Subscriber(
             StreamPeerType.SUBSCRIBER,
         )
     },
+    onRejoinNeeded = rejoin,
     traceCreateAnswer = false,
     tracer = tracer,
     maxBitRate = 0, // Set as needed
@@ -135,6 +141,31 @@ internal class Subscriber(
         return statsTracer?.get(trackIdToTrackType)
     }
 
+    override fun onIceConnectionChange(newState: PeerConnection.IceConnectionState?) {
+        super.onIceConnectionChange(newState)
+        when (newState) {
+            PeerConnection.IceConnectionState.CONNECTED -> {
+                restartIceJobDelegate.cancelScheduledRestartIce()
+            }
+
+            PeerConnection.IceConnectionState.FAILED -> {
+                restartIceJobDelegate.scheduleRestartIce {
+                    restartIce()
+                }
+            }
+
+            PeerConnection.IceConnectionState.DISCONNECTED -> {
+                restartIceJobDelegate.scheduleRestartIce(3000) {
+                    restartIce()
+                }
+            }
+
+            else -> {
+                // no-op
+            }
+        }
+    }
+
     /**
      * Returns the track dimensions for this subscriber.
      *

stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/call/connection/job/RestartIceJobDelegate.kt

@@ -0,0 +1,42 @@
+/*
+ * Copyright (c) 2014-2024 Stream.io Inc. All rights reserved.
+ *
+ * Licensed under the Stream License;
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    https://github.com/GetStream/stream-video-android/blob/main/LICENSE
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.getstream.video.android.core.call.connection.job
+
+import io.getstream.video.android.core.utils.safeCall
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Job
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.launch
+
+open class RestartIceJobDelegate(private val scope: CoroutineScope, private val scheduledJobTimeoutMs: Long = 3000) {
+
+    private var scheduledJob: Job? = null
+
+    fun scheduleRestartIce(timeoutMs: Long = scheduledJobTimeoutMs, restartIce: suspend () -> Unit) {
+        scheduledJob?.cancel()
+        scheduledJob = scope.launch {
+            safeCall {
+                delay(timeoutMs)
+                restartIce()
+            }
+        }
+    }
+
+    fun cancelScheduledRestartIce() {
+        scheduledJob?.cancel()
+    }
+}