diff --git a/packages/stream_chat/lib/src/api/websocket.dart b/packages/stream_chat/lib/src/api/websocket.dart index b5bbbb69..657aa125 100644 --- a/packages/stream_chat/lib/src/api/websocket.dart +++ b/packages/stream_chat/lib/src/api/websocket.dart @@ -100,7 +100,7 @@ class WebSocket { /// connection unhealthy final int reconnectionMonitorTimeout; - final _connectionStatusController = + final BehaviorSubject _connectionStatusController = BehaviorSubject.seeded(ConnectionStatus.disconnected); set _connectionStatus(ConnectionStatus status) => diff --git a/packages/stream_chat_flutter/example/lib/main.dart b/packages/stream_chat_flutter/example/lib/main.dart index 1f3b51d2..d3522886 100644 --- a/packages/stream_chat_flutter/example/lib/main.dart +++ b/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 { diff --git a/packages/stream_chat_flutter/lib/src/message_list_view.dart b/packages/stream_chat_flutter/lib/src/message_list_view.dart index 03c64606..f323b763 100644 --- a/packages/stream_chat_flutter/lib/src/message_list_view.dart +++ b/packages/stream_chat_flutter/lib/src/message_list_view.dart @@ -574,9 +574,9 @@ class _MessageListViewState extends State { return const SizedBox(); } - var index = _getTopElement(values).index; + var index = _getTopElement(values)?.index; - if (index > messages.length) { + if (index == null || index > messages.length) { return const SizedBox(); } @@ -602,10 +602,17 @@ class _MessageListViewState extends State { StreamChannelState? channel, QueryDirection direction) => _messageListController.paginateData!(direction: direction); - ItemPosition _getTopElement(Iterable values) => values - .where((ItemPosition position) => position.itemLeadingEdge < 0.9) - .reduce((ItemPosition max, ItemPosition position) => - position.itemLeadingEdge > max.itemLeadingEdge ? position : max); + ItemPosition? _getTopElement(Iterable values) { + final inView = + values.where((ItemPosition position) => position.itemLeadingEdge < 0.9); + + if (inView.isEmpty) { + return null; + } + + return inView.reduce((ItemPosition max, ItemPosition position) => + position.itemLeadingEdge > max.itemLeadingEdge ? position : max); + } Widget _buildScrollToBottom() => StreamBuilder>( stream: Rx.combineLatest2( diff --git a/packages/stream_chat_flutter/lib/src/stream_chat.dart b/packages/stream_chat_flutter/lib/src/stream_chat.dart index 424d4bcf..64acae98 100644 --- a/packages/stream_chat_flutter/lib/src/stream_chat.dart +++ b/packages/stream_chat_flutter/lib/src/stream_chat.dart @@ -39,6 +39,7 @@ class StreamChat extends StatefulWidget { this.streamChatThemeData, this.onBackgroundEventReceived, this.backgroundKeepAlive = const Duration(minutes: 1), + this.connectivityStream, }) : super(key: key); /// Client to do chat ops with @@ -59,6 +60,11 @@ class StreamChat extends StatefulWidget { /// upon the [Event.type] final EventHandler? onBackgroundEventReceived; + /// Stream of connectivity result + /// Visible for testing + @visibleForTesting + final Stream? connectivityStream; + @override StreamChatState createState() => StreamChatState(); @@ -102,6 +108,7 @@ class StreamChatState extends State { client: client, onBackgroundEventReceived: widget.onBackgroundEventReceived, backgroundKeepAlive: widget.backgroundKeepAlive, + connectivityStream: widget.connectivityStream, child: widget.child ?? const Offstage(), ), ); diff --git a/packages/stream_chat_flutter/test/src/deleted_message_test.dart b/packages/stream_chat_flutter/test/src/deleted_message_test.dart index bcf9cf62..206d92e3 100644 --- a/packages/stream_chat_flutter/test/src/deleted_message_test.dart +++ b/packages/stream_chat_flutter/test/src/deleted_message_test.dart @@ -79,6 +79,7 @@ void main() { ), ), ), + connectivityStream: Stream.value(ConnectivityResult.mobile), ), ), surfaceSize: const Size.square(200), @@ -131,6 +132,7 @@ void main() { ), ), ), + connectivityStream: Stream.value(ConnectivityResult.mobile), ), ), surfaceSize: const Size.square(200), @@ -187,6 +189,7 @@ void main() { ), ), ), + connectivityStream: Stream.value(ConnectivityResult.mobile), ), ), surfaceSize: const Size.square(200), diff --git a/packages/stream_chat_flutter/test/src/reaction_bubble_test.dart b/packages/stream_chat_flutter/test/src/reaction_bubble_test.dart index 770a040b..c5183f35 100644 --- a/packages/stream_chat_flutter/test/src/reaction_bubble_test.dart +++ b/packages/stream_chat_flutter/test/src/reaction_bubble_test.dart @@ -60,6 +60,7 @@ void main() { maskColor: theme.ownMessageTheme.reactionsMaskColor!, ), ), + connectivityStream: Stream.value(ConnectivityResult.mobile), ), surfaceSize: const Size(100, 100), ); @@ -96,6 +97,7 @@ void main() { maskColor: theme.ownMessageTheme.reactionsMaskColor!, ), ), + connectivityStream: Stream.value(ConnectivityResult.mobile), ), surfaceSize: const Size(100, 100), ); @@ -118,6 +120,7 @@ void main() { StreamChat( client: client, streamChatThemeData: StreamChatThemeData.fromTheme(themeData), + connectivityStream: Stream.value(ConnectivityResult.mobile), child: Container( color: Colors.black, child: ReactionBubble( @@ -162,6 +165,7 @@ void main() { StreamChat( client: client, streamChatThemeData: StreamChatThemeData.fromTheme(themeData), + connectivityStream: Stream.value(ConnectivityResult.mobile), child: Container( color: Colors.black, child: ReactionBubble( @@ -204,6 +208,7 @@ void main() { await tester.pumpWidgetBuilder( StreamChat( client: client, + connectivityStream: Stream.value(ConnectivityResult.mobile), streamChatThemeData: StreamChatThemeData.fromTheme(themeData), child: SizedBox( child: ReactionBubble( diff --git a/packages/stream_chat_flutter/test/src/system_message_test.dart b/packages/stream_chat_flutter/test/src/system_message_test.dart index 8fa3ad44..6bb1da43 100644 --- a/packages/stream_chat_flutter/test/src/system_message_test.dart +++ b/packages/stream_chat_flutter/test/src/system_message_test.dart @@ -100,6 +100,7 @@ void main() { ), ), ), + connectivityStream: Stream.value(ConnectivityResult.mobile), ), ), surfaceSize: const Size.square(200), @@ -151,6 +152,7 @@ void main() { ), ), ), + connectivityStream: Stream.value(ConnectivityResult.mobile), ), ), surfaceSize: const Size.square(200), diff --git a/packages/stream_chat_flutter_core/lib/src/stream_chat_core.dart b/packages/stream_chat_flutter_core/lib/src/stream_chat_core.dart index 89d4853a..d3ed4a0b 100644 --- a/packages/stream_chat_flutter_core/lib/src/stream_chat_core.dart +++ b/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? connectivityStream; + @override StreamChatCoreState createState() => StreamChatCoreState(); @@ -96,48 +103,104 @@ class StreamChatCoreState extends State /// The current user as a stream Stream get userStream => client.state.userStream; + StreamSubscription? _connectivitySubscription; + + var _isInForeground = true; + var _isConnectionAvailable = true; + @override void initState() { super.initState(); WidgetsBinding.instance?.addObserver(this); + _subscribeToConnectivityChange(widget.connectivityStream); + } + + void _subscribeToConnectivityChange([ + Stream? connectivityStream, + ]) { + if (_connectivitySubscription == null) { + connectivityStream ??= Connectivity().onConnectivityChanged; + _connectivitySubscription = + connectivityStream.distinct().listen((result) { + _isConnectionAvailable = result != ConnectivityResult.none; + if (!_isInForeground) return; + if (_isConnectionAvailable) { + if (client.wsConnectionStatus == ConnectionStatus.disconnected) { + client.connect(); + } + } else { + if (client.wsConnectionStatus == ConnectionStatus.connected) { + client.disconnect(); + } + } + }); + } + } + + void _unsubscribeFromConnectivityChange() { + if (_connectivitySubscription != null) { + _connectivitySubscription?.cancel(); + _connectivitySubscription = null; + } + } + + @override + void didUpdateWidget(StreamChatCore oldWidget) { + super.didUpdateWidget(oldWidget); + final connectivityStream = widget.connectivityStream; + if (connectivityStream != oldWidget.connectivityStream) { + _unsubscribeFromConnectivityChange(); + _subscribeToConnectivityChange(connectivityStream); + } } StreamSubscription? _eventSubscription; @override void didChangeAppLifecycleState(AppLifecycleState state) { + _isInForeground = state == AppLifecycleState.resumed; if (user != null) { - if (state == AppLifecycleState.paused) { - if (widget.onBackgroundEventReceived == null) { - client.disconnect(); - return; - } - _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(); - } - } + if (_isInForeground) { + _onForeground(); + } else { + _onBackground(); } } } + 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); + _unsubscribeFromConnectivityChange(); _eventSubscription?.cancel(); _disconnectTimer?.cancel(); super.dispose(); diff --git a/packages/stream_chat_flutter_core/lib/stream_chat_flutter_core.dart b/packages/stream_chat_flutter_core/lib/stream_chat_flutter_core.dart index 01463de0..ef947a0f 100644 --- a/packages/stream_chat_flutter_core/lib/stream_chat_flutter_core.dart +++ b/packages/stream_chat_flutter_core/lib/stream_chat_flutter_core.dart @@ -1,5 +1,6 @@ library stream_chat_flutter_core; +export 'package:connectivity_plus/connectivity_plus.dart'; export 'package:stream_chat/stream_chat.dart'; export 'src/channel_list_core.dart' hide ChannelListCoreState; diff --git a/packages/stream_chat_flutter_core/pubspec.yaml b/packages/stream_chat_flutter_core/pubspec.yaml index e8d1014d..4292772b 100644 --- a/packages/stream_chat_flutter_core/pubspec.yaml +++ b/packages/stream_chat_flutter_core/pubspec.yaml @@ -11,6 +11,7 @@ environment: dependencies: collection: ^1.15.0 + connectivity_plus: ^1.0.1 flutter: sdk: flutter meta: ^1.3.0 @@ -21,5 +22,5 @@ dev_dependencies: fake_async: ^1.2.0 flutter_test: sdk: flutter - mocktail: ^0.1.1 + mocktail: ^0.1.3 diff --git a/packages/stream_chat_flutter_core/test/stream_chat_core_test.dart b/packages/stream_chat_flutter_core/test/stream_chat_core_test.dart index bc5324b8..936e03f0 100644 --- a/packages/stream_chat_flutter_core/test/stream_chat_core_test.dart +++ b/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( + 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( + 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()); + }); + }, + ); }