rewatch calls after coordinator reconnects (#1029)

Author: Brazol

packages/stream_video/CHANGELOG.md

@@ -8,6 +8,7 @@
 * Fixed an issue where toggling camera enabled quickly could cause AVCaptureMultiCamSession to crash.
 * Fixed an issue where the default camera selection would occasionally be incorrect even when properly configured.
 * Fixed an issue where changing the audio input device while muted from the start of a call would not apply the new device when unmuting. The selected device will now be correctly set upon unmuting.
+* Fixed an issue where coordinator events were not listened to after a fast reconnect in a Call.
 * Fixes an issue with `muteUsers` method when called for all tracks.
 
 ## 0.10.0

packages/stream_video/lib/src/call/call.dart

@@ -1945,6 +1945,7 @@ class Call {
   }) async {
     if (watch) {
       _observeEvents();
+      _streamVideo.state.setWatchedCall(this);
     }
 
     final response = await coordinatorCall();

packages/stream_video/lib/src/coordinator/models/coordinator_events.dart

@@ -52,6 +52,20 @@ class CoordinatorDisconnectedEvent extends CoordinatorEvent {
   List<Object?> get props => [connectionId, userId, closeCode, closeReason];
 }
 
+/// Fired when web socket reconnected.
+class CoordinatorReconnectedEvent extends CoordinatorEvent {
+  const CoordinatorReconnectedEvent({
+    this.connectionId,
+    this.userId,
+  });
+
+  final String? connectionId;
+  final String? userId;
+
+  @override
+  List<Object?> get props => [connectionId, userId];
+}
+
 /// Sent periodically by the server to keep the connection alive.
 class CoordinatorHealthCheckEvent extends CoordinatorEvent {
   const CoordinatorHealthCheckEvent({

packages/stream_video/lib/src/coordinator/open_api/coordinator_ws.dart

@@ -48,7 +48,19 @@ class CoordinatorWebSocket extends StreamWebSocket implements HealthListener {
           _buildUrl(url, apiKey),
           protocols: protocols,
           tag: '$_tag-${++_seq}',
+        ) {
+    onConnectionStateUpdated = (event) {
+      if (event.oldState == ConnectionState.reconnecting &&
+          event.newState == ConnectionState.connected) {
+        _events.emit(
+          CoordinatorReconnectedEvent(
+            userId: userId,
+            connectionId: connectionId,
+          ),
         );
+      }
+    };
+  }
 
   late final _logger = taggedLogger(tag: '$_tag-$_seq');
 

packages/stream_video/lib/src/core/client_state.dart

@@ -22,6 +22,9 @@ abstract class ClientState {
   /// Emits a list of active calls.
   StateEmitter<List<Call>> get activeCalls;
 
+  /// Emits a list of watched calls.
+  StateEmitter<List<Call>> get watchedCalls;
+
   /// Emits an active call.
   /// Will only emit if options.allowMultipleActiveCalls is set to false, use activeCalls otherwise.
   StateEmitter<Call?> get activeCall;
@@ -40,6 +43,9 @@ abstract class ClientState {
   /// Otherwise the call is added to the list of active calls.
   Future<void> setActiveCall(Call? call);
 
+  /// Adds the call to the list of watched calls.
+  void setWatchedCall(Call call);
+
   /// Removes the call from the list of active calls.
   /// It won't `leave` the call, just removes it from the list.
   Future<void> removeActiveCall(Call call);
@@ -50,6 +56,7 @@ class MutableClientState implements ClientState {
       : user = MutableStateEmitterImpl(user),
         activeCalls = MutableStateEmitterImpl([]),
         activeCall = MutableStateEmitterImpl(null),
+        watchedCalls = MutableStateEmitterImpl([]),
         incomingCall = MutableStateEmitterImpl(null),
         outgoingCall = MutableStateEmitterImpl(null),
         connection = MutableStateEmitterImpl(
@@ -64,6 +71,9 @@ class MutableClientState implements ClientState {
   @override
   final MutableStateEmitter<List<Call>> activeCalls;
 
+  @override
+  final MutableStateEmitter<List<Call>> watchedCalls;
+
   @override
   final MutableStateEmitter<Call?> activeCall;
 
@@ -111,6 +121,10 @@ class MutableClientState implements ClientState {
     } else if (call != null) {
       activeCalls.value = [...activeCalls.value, call];
     }
+
+    if (call != null) {
+      setWatchedCall(call);
+    }
   }
 
   @override
@@ -124,6 +138,19 @@ class MutableClientState implements ClientState {
     activeCalls.value = [
       ...activeCalls.value.where((it) => it.callCid != call.callCid),
     ];
+
+    watchedCalls.value = [
+      ...watchedCalls.value.where((it) => it.callCid != call.callCid),
+    ];
+  }
+
+  @override
+  void setWatchedCall(Call call) {
+    if (watchedCalls.value.any((it) => it.callCid == call.callCid)) {
+      return;
+    }
+
+    watchedCalls.value = [...watchedCalls.value, call];
   }
 
   @override

packages/stream_video/lib/src/stream_video.dart

@@ -474,6 +474,30 @@ class StreamVideo extends Disposable {
       _connectionState = ConnectionState.disconnected(
         _state.currentUser.id,
       );
+    } else if (event is CoordinatorReconnectedEvent) {
+      _logger.i(() => '[onCoordinatorEvent] reconnected ${event.userId}');
+      if (state.watchedCalls.value.isNotEmpty) {
+        // Re-watch the previously watched calls.
+        unawaited(
+          queryCalls(
+            watch: true,
+            filterConditions: {
+              'cid': {
+                r'$in': state.watchedCalls.value
+                    .map((call) => call.callCid.value)
+                    .toList(),
+              },
+            },
+          ).onError(
+            (error, stackTrace) {
+              _logger.e(
+                () => '[onCoordinatorEvent] re-watching calls failed: $error',
+              );
+              return Result.failure(VideoErrors.compose(error, stackTrace));
+            },
+          ),
+        );
+      }
     }
   }