relisten to websocket events after reconnect

Author: renefloor

packages/stream_feeds/lib/src/client/feeds_client_impl.dart

@@ -1,5 +1,6 @@
 import 'dart:async';
 
+import 'package:rxdart/rxdart.dart';
 import 'package:stream_core/stream_core.dart';
 
 import '../cdn/cdn_api.dart';
@@ -270,6 +271,7 @@ class StreamFeedsClientImpl implements StreamFeedsClient {
       feedsRepository: _feedsRepository,
       pollsRepository: _pollsRepository,
       eventsEmitter: events,
+      onReconnectEmitter: onReconnectEmitter,
     );
   }
 
@@ -474,4 +476,25 @@ class StreamFeedsClientImpl implements StreamFeedsClient {
   Future<Result<void>> deleteImage({required String url}) {
     return _cdnClient.deleteImage(url);
   }
+
+  Stream<void> get onReconnectEmitter {
+    return connectionState.stream.scan(
+      (state, connectionStatus, i) => switch (connectionStatus) {
+        Initialized() || Connecting() => (
+            wasDisconnected: state.wasDisconnected,
+            reconnected: false,
+          ),
+        Disconnecting() || Disconnected() => (
+            wasDisconnected: true,
+            reconnected: false,
+          ),
+        Connected() => (
+            wasDisconnected: false,
+            reconnected: state.wasDisconnected,
+          ),
+        _ => state,
+      },
+      (wasDisconnected: false, reconnected: false),
+    ).mapNotNull((state) => state.reconnected ? () : null);
+  }
 }

packages/stream_feeds/lib/src/state/feed.dart

@@ -1,9 +1,11 @@
 import 'dart:async';
 
 import 'package:freezed_annotation/freezed_annotation.dart';
+import 'package:rxdart/rxdart.dart';
 import 'package:state_notifier/state_notifier.dart';
 import 'package:stream_core/stream_core.dart';
 
+import '../../stream_feeds.dart';
 import '../generated/api/api.dart' as api;
 import '../models/activity_data.dart';
 import '../models/bookmark_data.dart';
@@ -48,6 +50,7 @@ class Feed with Disposable {
     required this.feedsRepository,
     required this.pollsRepository,
     required this.eventsEmitter,
+    required Stream<void> onReconnectEmitter,
   }) {
     final fid = query.fid;
 
@@ -65,7 +68,12 @@ class Feed with Disposable {
 
     // Attach event handlers for the feed events
     final handler = FeedEventHandler(fid: fid, state: _stateNotifier);
-    _eventsSubscription = eventsEmitter.listen(handler.handleEvent);
+    _feedSubscriptions.add(eventsEmitter.listen(handler.handleEvent));
+
+    // Automatically refetch data on reconnection
+    if (query.watch) {
+      _subscribeToReconnectionUpdates(onReconnectEmitter: onReconnectEmitter);
+    }
   }
 
   FeedId get fid => query.fid;
@@ -86,11 +94,11 @@ class Feed with Disposable {
   late final FeedStateNotifier _stateNotifier;
 
   final SharedEmitter<WsEvent> eventsEmitter;
-  StreamSubscription<WsEvent>? _eventsSubscription;
+  final CompositeSubscription _feedSubscriptions = CompositeSubscription();
 
   @override
   void dispose() {
-    _eventsSubscription?.cancel();
+    _feedSubscriptions.cancel();
     _stateNotifier.dispose();
     _memberList.dispose();
     super.dispose();
@@ -628,4 +636,14 @@ class Feed with Disposable {
   }
 
   // endregion
+
+  void _subscribeToReconnectionUpdates({
+    required Stream<void> onReconnectEmitter,
+  }) {
+    _feedSubscriptions.add(
+      onReconnectEmitter.listen((_) {
+        getOrCreate();
+      }),
+    );
+  }
 }