implement connectivity plus

packages/stream_chat/lib/src/api/websocket.dart

@@ -100,7 +100,7 @@ class WebSocket {
   /// connection unhealthy
   final int reconnectionMonitorTimeout;
 
-  final _connectionStatusController =
+  final BehaviorSubject<ConnectionStatus> _connectionStatusController =
       BehaviorSubject.seeded(ConnectionStatus.disconnected);
 
   set _connectionStatus(ConnectionStatus status) =>

packages/stream_chat_flutter/example/lib/main.dart

@@ -4,7 +4,6 @@ import 'package:stream_chat_persistence/stream_chat_persistence.dart';
 
 final chatPersistentClient = StreamChatPersistenceClient(
   logLevel: Level.INFO,
-  connectionMode: ConnectionMode.background,
 );
 
 void main() async {

packages/stream_chat_flutter_core/lib/src/stream_chat_core.dart

@@ -1,5 +1,6 @@
 import 'dart:async';
 
+import 'package:connectivity_plus/connectivity_plus.dart';
 import 'package:flutter/foundation.dart';
 import 'package:flutter/material.dart';
 import 'package:stream_chat/stream_chat.dart';
@@ -43,6 +44,7 @@ class StreamChatCore extends StatefulWidget {
     required this.child,
     this.onBackgroundEventReceived,
     this.backgroundKeepAlive = const Duration(minutes: 1),
+    this.connectivityStream,
   }) : super(key: key);
 
   /// Instance of Stream Chat Client containing information about the current
@@ -61,6 +63,11 @@ class StreamChatCore extends StatefulWidget {
   /// upon the [Event.type]
   final EventHandler? onBackgroundEventReceived;
 
+  /// Stream of connectivity result
+  /// Visible for testing
+  @visibleForTesting
+  final Stream<ConnectivityResult>? connectivityStream;
+
   @override
   StreamChatCoreState createState() => StreamChatCoreState();
 
@@ -96,50 +103,84 @@ class StreamChatCoreState extends State<StreamChatCore>
   /// The current user as a stream
   Stream<User?> get userStream => client.state.userStream;
 
+  late final StreamSubscription<ConnectivityResult> _connectivitySubscription;
+
+  var _isInForeground = true;
+  var _isConnectionAvailable = true;
+
   @override
   void initState() {
     super.initState();
     WidgetsBinding.instance?.addObserver(this);
+    _connectivitySubscription =
+        (widget.connectivityStream ?? Connectivity().onConnectivityChanged)
+            .listen((ConnectivityResult result) async {
+      _isConnectionAvailable = result != ConnectivityResult.none;
+      if (!_isInForeground) {
+        return;
+      }
+      if (_isConnectionAvailable) {
+        if (client.wsConnectionStatus == ConnectionStatus.disconnected) {
+          await client.connect();
+        }
+      } else {
+        if (client.wsConnectionStatus == ConnectionStatus.connected) {
+          await client.disconnect();
+        }
+      }
+    });
   }
 
   StreamSubscription? _eventSubscription;
 
   @override
   void didChangeAppLifecycleState(AppLifecycleState state) {
+    _isInForeground = state == AppLifecycleState.resumed;
     if (user != null) {
-      if (state == AppLifecycleState.paused) {
+      if (!_isInForeground) {
-        if (widget.onBackgroundEventReceived == null) {
+        _onBackground();
-          client.disconnect();
+      } else {
-          return;
+        _onForeground();
-        }
-        _eventSubscription = client.on().listen(
-              widget.onBackgroundEventReceived,
-            );
-
-        void onTimerComplete() {
-          _eventSubscription?.cancel();
-          client.disconnect();
-        }
-
-        _disconnectTimer = Timer(widget.backgroundKeepAlive, onTimerComplete);
-      } else if (state == AppLifecycleState.resumed) {
-        if (_disconnectTimer?.isActive == true) {
-          _eventSubscription?.cancel();
-          _disconnectTimer?.cancel();
-        } else {
-          if (client.wsConnectionStatus == ConnectionStatus.disconnected) {
-            client.connect();
-          }
-        }
       }
     }
   }
 
+  void _onForeground() {
+    if (_disconnectTimer?.isActive == true) {
+      _eventSubscription?.cancel();
+      _disconnectTimer?.cancel();
+    } else if (client.wsConnectionStatus == ConnectionStatus.disconnected &&
+        _isConnectionAvailable) {
+      client.connect();
+    }
+  }
+
+  void _onBackground() {
+    if (widget.onBackgroundEventReceived == null) {
+      if (client.wsConnectionStatus != ConnectionStatus.disconnected) {
+        client.disconnect();
+      }
+      return;
+    }
+    _eventSubscription = client.on().listen(
+          widget.onBackgroundEventReceived,
+        );
+
+    void onTimerComplete() {
+      _eventSubscription?.cancel();
+      client.disconnect();
+    }
+
+    _disconnectTimer = Timer(widget.backgroundKeepAlive, onTimerComplete);
+    return;
+  }
+
   @override
   void dispose() {
     WidgetsBinding.instance?.removeObserver(this);
     _eventSubscription?.cancel();
     _disconnectTimer?.cancel();
+    _connectivitySubscription.cancel();
     super.dispose();
   }
 }

packages/stream_chat_flutter_core/pubspec.yaml

@@ -16,6 +16,7 @@ dependencies:
   meta: ^1.3.0
   rxdart: ^0.27.0
   stream_chat: ^2.0.0-nullsafety.2
+  connectivity_plus: ^1.0.1
 
 dev_dependencies:
   fake_async: ^1.2.0

packages/stream_chat_flutter_core/test/stream_chat_core_test.dart

@@ -1,8 +1,10 @@
 import 'dart:async';
 
+import 'package:connectivity_plus/connectivity_plus.dart';
 import 'package:flutter/material.dart';
 import 'package:flutter_test/flutter_test.dart';
 import 'package:mocktail/mocktail.dart';
+import 'package:rxdart/rxdart.dart';
 import 'package:stream_chat_flutter_core/stream_chat_flutter_core.dart';
 
 import 'mocks.dart';
@@ -100,6 +102,7 @@ void main() {
           child: Offstage(key: childKey),
           onBackgroundEventReceived: mockOnBackgroundEventReceived,
           backgroundKeepAlive: backgroundKeepAlive,
+          connectivityStream: Stream.value(ConnectivityResult.mobile),
         );
 
         await tester.pumpWidget(streamChatCore);
@@ -191,6 +194,7 @@ void main() {
           key: streamChatCoreKey,
           client: mockClient,
           child: Offstage(key: childKey),
+          connectivityStream: Stream.value(ConnectivityResult.mobile),
         );
 
         await tester.pumpWidget(streamChatCore);
@@ -222,6 +226,51 @@ void main() {
     },
   );
 
+  testWidgets(
+    'didChangeAppLifecycleState should not call client.connect() '
+    'if connection is not available in case the '
+    'widget lifestyle changes to AppLifecycleState.resume',
+    (tester) async {
+      await tester.runAsync(() async {
+        final mockClient = MockClient();
+        const streamChatCoreKey = Key('streamChatCore');
+        const childKey = Key('child');
+        final streamChatCore = StreamChatCore(
+          key: streamChatCoreKey,
+          client: mockClient,
+          child: Offstage(key: childKey),
+          connectivityStream: Stream.value(ConnectivityResult.none),
+        );
+
+        await tester.pumpWidget(streamChatCore);
+
+        expect(find.byKey(streamChatCoreKey), findsOneWidget);
+        expect(find.byKey(childKey), findsOneWidget);
+
+        final event = Event();
+        when(() => mockClient.on()).thenAnswer((_) => Stream.value(event));
+        when(() => mockClient.connect()).thenAnswer((_) async => event);
+        when(mockClient.disconnect).thenAnswer((_) async => null);
+        when(() => mockClient.wsConnectionStatus)
+            .thenReturn(ConnectionStatus.disconnected);
+
+        final streamChatCoreState = tester.state<StreamChatCoreState>(
+          find.byKey(streamChatCoreKey),
+        );
+
+        streamChatCoreState
+            .didChangeAppLifecycleState(AppLifecycleState.paused);
+
+        await Future.delayed(const Duration(seconds: 1));
+
+        streamChatCoreState
+            .didChangeAppLifecycleState(AppLifecycleState.resumed);
+
+        verifyNever(() => mockClient.connect());
+      });
+    },
+  );
+
   testWidgets(
     'streamChatCoreState.userStream should emit all the user events '
     'provided by client',
@@ -264,4 +313,124 @@ void main() {
       });
     },
   );
+
+  testWidgets(
+    'should call connect if in foreground and connection is back',
+    (tester) async {
+      await tester.runAsync(() async {
+        final mockClient = MockClient();
+        const streamChatCoreKey = Key('streamChatCore');
+        const childKey = Key('child');
+        final _connectivityController =
+            BehaviorSubject.seeded(ConnectivityResult.none);
+        final streamChatCore = StreamChatCore(
+          key: streamChatCoreKey,
+          client: mockClient,
+          child: Offstage(key: childKey),
+          connectivityStream: _connectivityController.stream,
+        );
+
+        await tester.pumpWidget(streamChatCore);
+
+        expect(find.byKey(streamChatCoreKey), findsOneWidget);
+        expect(find.byKey(childKey), findsOneWidget);
+
+        final event = Event();
+        when(() => mockClient.on()).thenAnswer((_) => Stream.value(event));
+        when(() => mockClient.connect()).thenAnswer((_) async => event);
+        when(mockClient.disconnect).thenAnswer((_) async => null);
+        when(() => mockClient.wsConnectionStatus)
+            .thenReturn(ConnectionStatus.disconnected);
+
+        _connectivityController.add(ConnectivityResult.mobile);
+
+        await Future.delayed(const Duration(seconds: 1));
+
+        verify(() => mockClient.connect()).called(1);
+      });
+    },
+  );
+
+  testWidgets(
+    'should call disconnect if in foreground and connection goes away',
+    (tester) async {
+      await tester.runAsync(() async {
+        final mockClient = MockClient();
+        const streamChatCoreKey = Key('streamChatCore');
+        const childKey = Key('child');
+        final _connectivityController =
+            BehaviorSubject.seeded(ConnectivityResult.mobile);
+        final streamChatCore = StreamChatCore(
+          key: streamChatCoreKey,
+          client: mockClient,
+          child: Offstage(key: childKey),
+          connectivityStream: _connectivityController.stream,
+        );
+
+        await tester.pumpWidget(streamChatCore);
+
+        expect(find.byKey(streamChatCoreKey), findsOneWidget);
+        expect(find.byKey(childKey), findsOneWidget);
+
+        final event = Event();
+        when(() => mockClient.on()).thenAnswer((_) => Stream.value(event));
+        when(() => mockClient.connect()).thenAnswer((_) async => event);
+        when(mockClient.disconnect).thenAnswer((_) async => null);
+        when(() => mockClient.wsConnectionStatus)
+            .thenReturn(ConnectionStatus.connected);
+
+        _connectivityController.add(ConnectivityResult.none);
+
+        await Future.delayed(const Duration(seconds: 1));
+
+        verify(() => mockClient.disconnect()).called(1);
+      });
+    },
+  );
+
+  testWidgets(
+    'should ignore connectivity in background',
+    (tester) async {
+      await tester.runAsync(() async {
+        final mockClient = MockClient();
+        const streamChatCoreKey = Key('streamChatCore');
+        const childKey = Key('child');
+        final _connectivityController =
+            BehaviorSubject.seeded(ConnectivityResult.none);
+        final streamChatCore = StreamChatCore(
+          key: streamChatCoreKey,
+          client: mockClient,
+          child: Offstage(key: childKey),
+          connectivityStream: _connectivityController.stream,
+        );
+
+        await tester.pumpWidget(streamChatCore);
+
+        expect(find.byKey(streamChatCoreKey), findsOneWidget);
+        expect(find.byKey(childKey), findsOneWidget);
+
+        final event = Event();
+        when(() => mockClient.on()).thenAnswer((_) => Stream.value(event));
+        when(() => mockClient.connect()).thenAnswer((_) async => event);
+        when(mockClient.disconnect).thenAnswer((_) async => null);
+        when(() => mockClient.wsConnectionStatus)
+            .thenReturn(ConnectionStatus.disconnected);
+
+        final streamChatCoreState = tester.state<StreamChatCoreState>(
+          find.byKey(streamChatCoreKey),
+        );
+
+        streamChatCoreState
+            .didChangeAppLifecycleState(AppLifecycleState.paused);
+
+        await Future.delayed(const Duration(seconds: 1));
+
+        _connectivityController.add(ConnectivityResult.mobile);
+
+        await Future.delayed(const Duration(seconds: 1));
+
+        verifyNever(() => mockClient.disconnect());
+      });
+    },
+  );
 }