fix: disconnect when in background and await connecting to be ready

Author: Vinzent03

packages/realtime_client/lib/realtime_client.dart

@@ -1,4 +1,5 @@
-export 'src/constants.dart' show RealtimeConstants, RealtimeLogLevel;
+export 'src/constants.dart'
+    show RealtimeConstants, RealtimeLogLevel, SocketStates;
 export 'src/realtime_channel.dart';
 export 'src/realtime_client.dart';
 export 'src/realtime_presence.dart';

packages/realtime_client/lib/src/constants.dart

@@ -18,8 +18,8 @@ enum SocketStates {
   /// Connection is live and connected
   open,
 
-  /// Socket is closing.
-  closing,
+  /// Socket is closing by the user
+  disconnecting,
 
   /// Socket being close not by the user. Realtime should attempt to reconnect.
   closed,

packages/realtime_client/lib/src/realtime_channel.dart

@@ -646,6 +646,19 @@ class RealtimeChannel {
     joinPush.resend(timeout ?? _timeout);
   }
 
+  /// Usually a rejoin only happens when the channel timeouts or errors out.
+  /// When manually disconnecting, the channel is still marked as
+  /// [ChannelStates.joined]. Calling [RealtimeClient.leaveOpenTopic] will
+  /// unsubscribe itself, which causes issues when trying to rejoin. This method
+  /// therefore doesn't call [RealtimeClient.leaveOpenTopic].
+  void forceRejoin([Duration? timeout]) {
+    if (isLeaving) {
+      return;
+    }
+    _state = ChannelStates.joining;
+    joinPush.resend(timeout ?? _timeout);
+  }
+
   void trigger(String type, [dynamic payload, String? ref]) {
     final typeLower = type.toLowerCase();
 

packages/realtime_client/lib/src/realtime_client.dart

@@ -139,17 +139,17 @@ class RealtimeClient {
         (String payload, Function(dynamic result) callback) =>
             callback(json.decode(payload));
     reconnectTimer = RetryTimer(
-      () {
-        disconnect();
-        connect();
+      () async {
+        await disconnect();
+        await connect();
       },
       this.reconnectAfterMs,
     );
   }
 
   /// Connects the socket.
   @internal
-  void connect() async {
+  Future<void> connect() async {
     if (conn != null) {
       return;
     }
@@ -161,8 +161,15 @@ class RealtimeClient {
       try {
         await conn!.ready;
       } catch (error) {
-        _onConnError(error);
-        reconnectTimer.scheduleTimeout();
+        // Don't schedule a reconnect and emit error if connection has been
+        // closed by the user or [disconnect] waits for the connection to be
+        // ready before closing it.
+        if (connState != SocketStates.disconnected &&
+            connState != SocketStates.disconnecting) {
+          connState = SocketStates.closed;
+          _onConnError(error);
+          reconnectTimer.scheduleTimeout();
+        }
         return;
       }
 
@@ -176,7 +183,8 @@ class RealtimeClient {
         onError: _onConnError,
         onDone: () {
           // communication has been closed
-          if (connState != SocketStates.disconnected) {
+          if (connState != SocketStates.disconnected &&
+              connState != SocketStates.disconnecting) {
             connState = SocketStates.closed;
           }
           _onConnClose();
@@ -189,17 +197,26 @@ class RealtimeClient {
   }
 
   /// Disconnects the socket with status [code] and [reason] for the disconnect
-  void disconnect({int? code, String? reason}) {
+  Future<void> disconnect({int? code, String? reason}) async {
     final conn = this.conn;
     if (conn != null) {
-      final connectionWasOpen = connState == SocketStates.open;
-      connState = SocketStates.disconnected;
-      if (connectionWasOpen) {
+      final oldState = connState;
+      connState = SocketStates.disconnecting;
+
+      // Connection cannot be closed while it's still connecting. Wait for connection to
+      // be ready and then close it.
+      if (oldState == SocketStates.connecting) {
+        await conn.ready.catchError((_) {});
+      }
+
+      if (oldState == SocketStates.open ||
+          oldState == SocketStates.connecting) {
         if (code != null) {
-          conn.sink.close(code, reason ?? '');
+          await conn.sink.close(code, reason ?? '');
         } else {
-          conn.sink.close();
+          await conn.sink.close();
         }
+        connState = SocketStates.disconnected;
         reconnectTimer.reset();
       }
       this.conn = null;
@@ -264,8 +281,8 @@ class RealtimeClient {
         return 'connecting';
       case SocketStates.open:
         return 'open';
-      case SocketStates.closing:
-        return 'closing';
+      case SocketStates.disconnecting:
+        return 'disconnecting';
       case SocketStates.disconnected:
         return 'disconnected';
       case SocketStates.closed:
@@ -275,7 +292,7 @@ class RealtimeClient {
   }
 
   /// Retuns `true` is the connection is open.
-  bool get isConnected => connectionState == 'open';
+  bool get isConnected => connState == SocketStates.open;
 
   /// Removes a subscription from the socket.
   @internal
@@ -374,7 +391,7 @@ class RealtimeClient {
     }
   }
 
-  /// Unsubscribe from channels with the specified topic.
+  /// Unsubscribe from joined or joining channels with the specified topic.
   @internal
   void leaveOpenTopic(String topic) {
     final dupChannel = channels.firstWhereOrNull(

packages/realtime_client/test/socket_test.dart

@@ -215,8 +215,8 @@ void main() {
     });
 
     test('removes existing connection', () async {
-      socket.connect();
-      socket.disconnect();
+      await socket.connect();
+      await socket.disconnect();
 
       expect(socket.conn, null);
     });

packages/supabase_flutter/lib/src/supabase_auth.dart

@@ -4,6 +4,7 @@ import 'dart:io' show Platform;
 import 'dart:math';
 
 import 'package:app_links/app_links.dart';
+import 'package:async/async.dart';
 import 'package:flutter/foundation.dart' show kIsWeb;
 import 'package:flutter/material.dart';
 import 'package:flutter/services.dart';
@@ -29,6 +30,8 @@ class SupabaseAuth with WidgetsBindingObserver {
 
   StreamSubscription<Uri?>? _deeplinkSubscription;
 
+  CancelableOperation<void>? _realtimeReconnectOperation;
+
   final _appLinks = AppLinks();
 
   /// - Obtains session from local storage and sets it as the current session
@@ -113,22 +116,67 @@ class SupabaseAuth with WidgetsBindingObserver {
   void didChangeAppLifecycleState(AppLifecycleState state) {
     switch (state) {
       case AppLifecycleState.resumed:
-        if (_autoRefreshToken) {
-          Supabase.instance.client.auth.startAutoRefresh();
-        }
+        onResumed();
       case AppLifecycleState.detached:
-      case AppLifecycleState.inactive:
       case AppLifecycleState.paused:
-        // Realtime channels are kept alive in the background for some amount
-        // of time after the app is paused. If we stop refreshing the token
-        // here, the channels will be closed.
-        if (Supabase.instance.client.realtime.getChannels().isEmpty) {
+        if (kIsWeb || Platform.isAndroid || Platform.isIOS) {
           Supabase.instance.client.auth.stopAutoRefresh();
+          _realtimeReconnectOperation?.cancel();
+          Supabase.instance.client.realtime.disconnect();
         }
       default:
     }
   }
 
+  Future<void> onResumed() async {
+    if (_autoRefreshToken) {
+      Supabase.instance.client.auth.startAutoRefresh();
+    }
+    final realtime = Supabase.instance.client.realtime;
+    if (realtime.channels.isNotEmpty) {
+      if (realtime.connState == SocketStates.disconnecting) {
+        // If the socket is still disconnecting from e.g.
+        // [AppLifecycleState.paused] we should wait for it to finish before
+        // reconnecting.
+
+        bool cancel = false;
+        final connectFuture = realtime.conn!.sink.done.then(
+          (_) {
+            // Make this connect cancelable so that it does not connect if the
+            // disconnect took so long that the app is already in background
+            // again.
+
+            // ignore: invalid_use_of_internal_member
+            if (!cancel) return realtime.connect();
+          },
+          onError: (error) {},
+        );
+        _realtimeReconnectOperation = CancelableOperation.fromFuture(
+          connectFuture,
+          onCancel: () => cancel = true,
+        );
+      } else if (!realtime.isConnected) {
+        // Reconnect if the socket is currently not connected.
+        // When coming from [AppLifecycleState.paused] this should be the case,
+        // but when coming from [AppLifecycleState.inactive] no disconnect
+        // happened and therefore connection should still be intanct and we
+        // should not reconnect.
+
+        // ignore: invalid_use_of_internal_member
+        await realtime.connect();
+        for (final channel in realtime.channels) {
+          // Only rejoin channels that think they are still joined and not
+          // which were manually unsubscribed by the user while in background
+
+          // ignore: invalid_use_of_internal_member
+          if (channel.isJoined) {
+            channel.forceRejoin();
+          }
+        }
+      }
+    }
+  }
+
   void _onAuthStateChange(AuthChangeEvent event, Session? session) {
     Supabase.instance.log('**** onAuthStateChange: $event');
     if (session != null) {