Merge pull request #1521 from GetStream/feat/synchronize-sync

.github/workflows/stream_flutter_workflow.yml

@@ -8,6 +8,11 @@ on:
   pull_request:
     paths:
       - 'packages/**'
+    types:
+      - opened
+      - reopened
+      - synchronize
+      - ready_for_review
   push:
     branches:
       - master

packages/stream_chat/CHANGELOG.md

@@ -8,6 +8,8 @@
 ✅ Added
 
 - Expose `ChannelMute` class. [#1473](https://github.com/GetStream/stream-chat-flutter/issues/1473)
+- Added synchronization to the `StreamChatClient.sync`
+  api. [#1392](https://github.com/GetStream/stream-chat-flutter/issues/1392)
 
 ## 6.0.0
 

packages/stream_chat/lib/src/client/client.dart

@@ -32,6 +32,7 @@ import 'package:stream_chat/src/event_type.dart';
 import 'package:stream_chat/src/ws/connection_status.dart';
 import 'package:stream_chat/src/ws/websocket.dart';
 import 'package:stream_chat/version.dart';
+import 'package:synchronized/extension.dart';
 
 /// Handler function used for logging records. Function requires a single
 /// [LogRecord] as the only parameter.
@@ -488,37 +489,40 @@ class StreamChatClient {
 
   /// Get the events missed while offline to sync the offline storage
   /// Will automatically fetch [cids] and [lastSyncedAt] if [persistenceEnabled]
-  Future<void> sync({List<String>? cids, DateTime? lastSyncAt}) async {
-    cids ??= await _chatPersistenceClient?.getChannelCids();
-    if (cids == null || cids.isEmpty) {
-      return;
-    }
-
-    lastSyncAt ??= await _chatPersistenceClient?.getLastSyncAt();
-    if (lastSyncAt == null) {
-      return;
-    }
-
-    try {
-      final res = await _chatApi.general.sync(cids, lastSyncAt);
-      final events = res.events
-        ..sort((a, b) => a.createdAt.compareTo(b.createdAt));
-
-      for (final event in events) {
-        logger.fine('event.type: ${event.type}');
-        final messageText = event.message?.text;
-        if (messageText != null) {
-          logger.fine('event.message.text: $messageText');
-        }
-        handleEvent(event);
+  Future<void> sync({List<String>? cids, DateTime? lastSyncAt}) {
+    return synchronized(() async {
+      final channels = cids ?? await _chatPersistenceClient?.getChannelCids();
+      if (channels == null || channels.isEmpty) {
+        return;
       }
 
-      final now = DateTime.now();
-      _lastSyncedAt = now;
-      _chatPersistenceClient?.updateLastSyncAt(now);
-    } catch (e, stk) {
-      logger.severe('Error during sync', e, stk);
-    }
+      final syncAt =
+          lastSyncAt ?? await _chatPersistenceClient?.getLastSyncAt();
+      if (syncAt == null) {
+        return;
+      }
+
+      try {
+        final res = await _chatApi.general.sync(channels, syncAt);
+        final events = res.events
+          ..sort((a, b) => a.createdAt.compareTo(b.createdAt));
+
+        for (final event in events) {
+          logger.fine('event.type: ${event.type}');
+          final messageText = event.message?.text;
+          if (messageText != null) {
+            logger.fine('event.message.text: $messageText');
+          }
+          handleEvent(event);
+        }
+
+        final now = DateTime.now();
+        _lastSyncedAt = now;
+        _chatPersistenceClient?.updateLastSyncAt(now);
+      } catch (e, stk) {
+        logger.severe('Error during sync', e, stk);
+      }
+    });
   }
 
   final _queryChannelsStreams = <String, Future<List<Channel>>>{};

packages/stream_chat/pubspec.yaml

@@ -22,6 +22,7 @@ dependencies:
   mime: ^1.0.4
   rate_limiter: ^1.0.0
   rxdart: ^0.27.7
+  synchronized: ^3.0.0
   uuid: ^3.0.7
   web_socket_channel: ^2.3.0
 

packages/stream_chat_persistence/README.md

@@ -37,7 +37,7 @@ The usage is pretty simple.
 ```dart
 final chatPersistentClient = StreamChatPersistenceClient(
   logLevel: Level.INFO,
-  connectionMode: ConnectionMode.background,
+  connectionMode: ConnectionMode.regular,
 );
 ```
 2. Pass the instance to the official Stream chat client.

packages/stream_chat_persistence/lib/src/stream_chat_persistence_client.dart

@@ -1,7 +1,5 @@
 import 'package:flutter/foundation.dart';
-import 'package:mutex/mutex.dart';
 import 'package:stream_chat/stream_chat.dart';
-
 import 'package:stream_chat_persistence/src/db/drift_chat_database.dart';
 
 /// Various connection modes on which [StreamChatPersistenceClient] can work
@@ -48,7 +46,6 @@ class StreamChatPersistenceClient extends ChatPersistenceClient {
   final Logger _logger;
   final ConnectionMode _connectionMode;
   final bool _webUseIndexedDbIfSupported;
-  final _mutex = ReadWriteMutex();
 
   void _defaultLogHandler(LogRecord record) {
     print(
@@ -59,9 +56,6 @@ class StreamChatPersistenceClient extends ChatPersistenceClient {
     if (record.stackTrace != null) print(record.stackTrace);
   }
 
-  Future<T> _readProtected<T>(AsyncValueGetter<T> func) =>
-      _mutex.protectRead(func);
-
   bool get _debugIsConnected {
     assert(() {
       if (db == null) {
@@ -96,6 +90,7 @@ class StreamChatPersistenceClient extends ChatPersistenceClient {
         'disconnect the previous instance before connecting again.',
       );
     }
+    _logger.info('connect');
     db = databaseProvider?.call(userId, _connectionMode) ??
         await _defaultDatabaseProvider(userId, _connectionMode);
   }
@@ -104,90 +99,84 @@ class StreamChatPersistenceClient extends ChatPersistenceClient {
   Future<Event?> getConnectionInfo() {
     assert(_debugIsConnected, '');
     _logger.info('getConnectionInfo');
-    return _readProtected(() => db!.connectionEventDao.connectionEvent);
+    return db!.connectionEventDao.connectionEvent;
   }
 
   @override
   Future<void> updateConnectionInfo(Event event) {
     assert(_debugIsConnected, '');
     _logger.info('updateConnectionInfo');
-    return _readProtected(
-      () => db!.connectionEventDao.updateConnectionEvent(event),
-    );
+    return db!.connectionEventDao.updateConnectionEvent(event);
   }
 
   @override
   Future<void> updateLastSyncAt(DateTime lastSyncAt) {
     assert(_debugIsConnected, '');
     _logger.info('updateLastSyncAt');
-    return _readProtected(
-      () => db!.connectionEventDao.updateLastSyncAt(lastSyncAt),
-    );
+    return db!.connectionEventDao.updateLastSyncAt(lastSyncAt);
   }
 
   @override
   Future<DateTime?> getLastSyncAt() {
     assert(_debugIsConnected, '');
     _logger.info('getLastSyncAt');
-    return _readProtected(() => db!.connectionEventDao.lastSyncAt);
+    return db!.connectionEventDao.lastSyncAt;
   }
 
   @override
   Future<void> deleteChannels(List<String> cids) {
     assert(_debugIsConnected, '');
     _logger.info('deleteChannels');
-    return _readProtected(() => db!.channelDao.deleteChannelByCids(cids));
+    return db!.channelDao.deleteChannelByCids(cids);
   }
 
   @override
   Future<List<String>> getChannelCids() {
     assert(_debugIsConnected, '');
     _logger.info('getChannelCids');
-    return _readProtected(() => db!.channelDao.cids);
+    return db!.channelDao.cids;
   }
 
   @override
   Future<void> deleteMessageByIds(List<String> messageIds) {
     assert(_debugIsConnected, '');
     _logger.info('deleteMessageByIds');
-    return _readProtected(() => db!.messageDao.deleteMessageByIds(messageIds));
+    return db!.messageDao.deleteMessageByIds(messageIds);
   }
 
   @override
   Future<void> deletePinnedMessageByIds(List<String> messageIds) {
     assert(_debugIsConnected, '');
     _logger.info('deletePinnedMessageByIds');
-    return _readProtected(
-      () => db!.pinnedMessageDao.deleteMessageByIds(messageIds),
-    );
+    return db!.pinnedMessageDao.deleteMessageByIds(messageIds);
   }
 
   @override
   Future<void> deleteMessageByCids(List<String> cids) {
     assert(_debugIsConnected, '');
     _logger.info('deleteMessageByCids');
-    return _readProtected(() => db!.messageDao.deleteMessageByCids(cids));
+    return db!.messageDao.deleteMessageByCids(cids);
   }
 
   @override
   Future<void> deletePinnedMessageByCids(List<String> cids) {
     assert(_debugIsConnected, '');
     _logger.info('deletePinnedMessageByCids');
-    return _readProtected(() => db!.pinnedMessageDao.deleteMessageByCids(cids));
+    return db!.pinnedMessageDao.deleteMessageByCids(cids);
   }
 
   @override
   Future<List<Member>> getMembersByCid(String cid) {
     assert(_debugIsConnected, '');
     _logger.info('getMembersByCid');
-    return _readProtected(() => db!.memberDao.getMembersByCid(cid));
+    return db!.memberDao.getMembersByCid(cid);
   }
 
   @override
   Future<ChannelModel?> getChannelByCid(String cid) {
     assert(_debugIsConnected, '');
     _logger.info('getChannelByCid');
-    return _readProtected(() => db!.channelDao.getChannelByCid(cid));
+    return db!.channelDao.getChannelByCid(cid);
   }
 
   @override
@@ -197,11 +186,9 @@ class StreamChatPersistenceClient extends ChatPersistenceClient {
   }) {
     assert(_debugIsConnected, '');
     _logger.info('getMessagesByCid');
-    return _readProtected(
-      () => db!.messageDao.getMessagesByCid(
-        cid,
-        messagePagination: messagePagination,
-      ),
+    return db!.messageDao.getMessagesByCid(
+      cid,
+      messagePagination: messagePagination,
     );
   }
 
@@ -212,37 +199,34 @@ class StreamChatPersistenceClient extends ChatPersistenceClient {
   }) {
     assert(_debugIsConnected, '');
     _logger.info('getPinnedMessagesByCid');
-    return _readProtected(
-      () => db!.pinnedMessageDao.getMessagesByCid(
-        cid,
-        messagePagination: messagePagination,
-      ),
+    return db!.pinnedMessageDao.getMessagesByCid(
+      cid,
+      messagePagination: messagePagination,
     );
   }
 
   @override
-  Future<List<Read>> getReadsByCid(String cid) {
+  Future<List<Read>> getReadsByCid(String cid) async {
     assert(_debugIsConnected, '');
     _logger.info('getReadsByCid');
-    return _readProtected(() => db!.readDao.getReadsByCid(cid));
+    return db!.readDao.getReadsByCid(cid);
   }
 
   @override
-  Future<Map<String, List<Message>>> getChannelThreads(String cid) {
+  Future<Map<String, List<Message>>> getChannelThreads(String cid) async {
     assert(_debugIsConnected, '');
     _logger.info('getChannelThreads');
-    return _readProtected(() async {
-      final messages = await db!.messageDao.getThreadMessages(cid);
-      final messageByParentIdDictionary = <String, List<Message>>{};
-      for (final message in messages) {
-        final parentId = message.parentId!;
-        messageByParentIdDictionary[parentId] = [
-          ...messageByParentIdDictionary[parentId] ?? [],
-          message,
-        ];
-      }
-      return messageByParentIdDictionary;
-    });
+    final messages = await db!.messageDao.getThreadMessages(cid);
+    final messageByParentIdDictionary = <String, List<Message>>{};
+    for (final message in messages) {
+      final parentId = message.parentId!;
+      messageByParentIdDictionary[parentId] = [
+        ...messageByParentIdDictionary[parentId] ?? [],
+        message,
+      ];
+    }
+
+    return messageByParentIdDictionary;
   }
 
   @override
@@ -252,11 +236,9 @@ class StreamChatPersistenceClient extends ChatPersistenceClient {
   }) {
     assert(_debugIsConnected, '');
     _logger.info('getReplies');
-    return _readProtected(
-      () => db!.messageDao.getThreadMessagesByParentId(
-        parentId,
-        options: options,
-      ),
+    return db!.messageDao.getThreadMessagesByParentId(
+      parentId,
+      options: options,
     );
   }
 
@@ -268,73 +250,45 @@ class StreamChatPersistenceClient extends ChatPersistenceClient {
     Please use channelStateSort instead.''') List<SortOption<ChannelModel>>? sort,
     List<SortOption<ChannelState>>? channelStateSort,
     PaginationParams? paginationParams,
-  }) {
+  }) async {
     assert(_debugIsConnected, '');
     assert(
       sort == null || channelStateSort == null,
       'sort and channelStateSort cannot be used together',
     );
     _logger.info('getChannelStates');
-    return _readProtected(
-      () async {
-        final channels = await db!.channelQueryDao.getChannels(
-          filter: filter,
-          sort: sort,
-        );
 
-        final channelStates = await Future.wait(
-          channels.map((e) => getChannelStateByCid(e.cid)),
-        );
-
-        // Only sort the channel states if the channels are not already sorted.
-        if (sort == null) {
-          var chainedComparator = (ChannelState a, ChannelState b) {
-            final dateA = a.channel?.lastMessageAt ?? a.channel?.createdAt;
-            final dateB = b.channel?.lastMessageAt ?? b.channel?.createdAt;
-
-            if (dateA == null && dateB == null) {
-              return 0;
-            } else if (dateA == null) {
-              return 1;
-            } else if (dateB == null) {
-              return -1;
-            } else {
-              return dateB.compareTo(dateA);
-            }
-          };
-
-          if (channelStateSort != null && channelStateSort.isNotEmpty) {
-            chainedComparator = (a, b) {
-              int result;
-              for (final comparator in channelStateSort
-                  .map((it) => it.comparator)
-                  .withNullifyer) {
-                try {
-                  result = comparator(a, b);
-                } catch (e) {
-                  result = 0;
-                }
-                if (result != 0) return result;
-              }
-              return 0;
-            };
-          }
-
-          channelStates.sort(chainedComparator);
-        }
-
-        final offset = paginationParams?.offset;
-        if (offset != null && offset > 0 && channelStates.isNotEmpty) {
-          channelStates.removeRange(0, offset);
-        }
-
-        if (paginationParams?.limit != null) {
-          return channelStates.take(paginationParams!.limit).toList();
-        }
-
-        return channelStates;
-      },
+    final channels = await db!.channelQueryDao.getChannels(
+      filter: filter,
+      sort: sort,
     );
+
+    final channelStates = await Future.wait(
+      channels.map((e) => getChannelStateByCid(e.cid)),
+    );
+
+    // Only sort the channel states if the channels are not already sorted.
+    if (sort == null) {
+      var comparator = _defaultChannelStateComparator;
+      if (channelStateSort != null && channelStateSort.isNotEmpty) {
+        comparator = _combineComparators(
+          channelStateSort.map((it) => it.comparator).withNullifyer,
+        );
+      }
+
+      channelStates.sort(comparator);
+    }
+
+    final offset = paginationParams?.offset;
+    if (offset != null && offset > 0 && channelStates.isNotEmpty) {
+      channelStates.removeRange(0, offset);
+    }
+
+    if (paginationParams?.limit != null) {
+      return channelStates.take(paginationParams!.limit).toList();
+    }
+
+    return channelStates;
   }
 
   @override
@@ -345,12 +299,10 @@ class StreamChatPersistenceClient extends ChatPersistenceClient {
   }) {
     assert(_debugIsConnected, '');
     _logger.info('updateChannelQueries');
-    return _readProtected(
-      () => db!.channelQueryDao.updateChannelQueries(
-        filter,
-        cids,
-        clearQueryCache: clearQueryCache,
-      ),
+    return db!.channelQueryDao.updateChannelQueries(
+      filter,
+      cids,
+      clearQueryCache: clearQueryCache,
     );
   }
 
@@ -358,60 +310,56 @@ class StreamChatPersistenceClient extends ChatPersistenceClient {
   Future<void> updateChannels(List<ChannelModel> channels) {
     assert(_debugIsConnected, '');
     _logger.info('updateChannels');
-    return _readProtected(() => db!.channelDao.updateChannels(channels));
+    return db!.channelDao.updateChannels(channels);
   }
 
   @override
   Future<void> bulkUpdateMembers(Map<String, List<Member>?> members) {
     assert(_debugIsConnected, '');
     _logger.info('bulkUpdateMembers');
-    return _readProtected(() => db!.memberDao.bulkUpdateMembers(members));
+    return db!.memberDao.bulkUpdateMembers(members);
   }
 
   @override
   Future<void> bulkUpdateMessages(Map<String, List<Message>?> messages) {
     assert(_debugIsConnected, '');
     _logger.info('bulkUpdateMessages');
-    return _readProtected(() => db!.messageDao.bulkUpdateMessages(messages));
+    return db!.messageDao.bulkUpdateMessages(messages);
   }
 
   @override
   Future<void> bulkUpdatePinnedMessages(Map<String, List<Message>?> messages) {
     assert(_debugIsConnected, '');
     _logger.info('bulkUpdatePinnedMessages');
-    return _readProtected(
-      () => db!.pinnedMessageDao.bulkUpdateMessages(messages),
-    );
+    return db!.pinnedMessageDao.bulkUpdateMessages(messages);
   }
 
   @override
   Future<void> updatePinnedMessageReactions(List<Reaction> reactions) {
     assert(_debugIsConnected, '');
     _logger.info('updatePinnedMessageReactions');
-    return _readProtected(
-      () => db!.pinnedMessageReactionDao.updateReactions(reactions),
-    );
+    return db!.pinnedMessageReactionDao.updateReactions(reactions);
   }
 
   @override
   Future<void> updateReactions(List<Reaction> reactions) {
     assert(_debugIsConnected, '');
     _logger.info('updateReactions');
-    return _readProtected(() => db!.reactionDao.updateReactions(reactions));
+    return db!.reactionDao.updateReactions(reactions);
   }
 
   @override
   Future<void> bulkUpdateReads(Map<String, List<Read>?> reads) {
     assert(_debugIsConnected, '');
     _logger.info('bulkUpdateReads');
-    return _readProtected(() => db!.readDao.bulkUpdateReads(reads));
+    return db!.readDao.bulkUpdateReads(reads);
   }
 
   @override
   Future<void> updateUsers(List<User> users) {
     assert(_debugIsConnected, '');
     _logger.info('updateUsers');
-    return _readProtected(() => db!.userDao.updateUsers(users));
+    return db!.userDao.updateUsers(users);
   }
 
   @override
@@ -420,53 +368,83 @@ class StreamChatPersistenceClient extends ChatPersistenceClient {
   ) {
     assert(_debugIsConnected, '');
     _logger.info('deletePinnedMessageReactionsByMessageId');
-    return _readProtected(
-      () =>
-          db!.pinnedMessageReactionDao.deleteReactionsByMessageIds(messageIds),
-    );
+    return db!.pinnedMessageReactionDao.deleteReactionsByMessageIds(messageIds);
   }
 
   @override
   Future<void> deleteReactionsByMessageId(List<String> messageIds) {
     assert(_debugIsConnected, '');
     _logger.info('deleteReactionsByMessageId');
-    return _readProtected(
-      () => db!.reactionDao.deleteReactionsByMessageIds(messageIds),
-    );
+    return db!.reactionDao.deleteReactionsByMessageIds(messageIds);
   }
 
   @override
   Future<void> deleteMembersByCids(List<String> cids) {
     assert(_debugIsConnected, '');
     _logger.info('deleteMembersByCids');
-    return _readProtected(() => db!.memberDao.deleteMemberByCids(cids));
+    return db!.memberDao.deleteMemberByCids(cids);
+  }
+
+  @override
+  Future<void> updateChannelThreads(
+    String cid,
+    Map<String, List<Message>> threads,
+  ) {
+    assert(_debugIsConnected, '');
+    _logger.info('updateChannelThreads');
+    return db!.transaction(() => super.updateChannelThreads(cid, threads));
   }
 
   @override
   Future<void> updateChannelStates(List<ChannelState> channelStates) {
     assert(_debugIsConnected, '');
     _logger.info('updateChannelStates');
-    return _readProtected(
-      () async => db!.transaction(
-        () async {
-          await super.updateChannelStates(channelStates);
-        },
-      ),
-    );
+    return db!.transaction(() => super.updateChannelStates(channelStates));
   }
 
   @override
-  Future<void> disconnect({bool flush = false}) async =>
-      _mutex.protectWrite(() async {
-        _logger.info('disconnect');
-        if (db != null) {
-          _logger.info('Disconnecting');
-          if (flush) {
-            _logger.info('Flushing');
-            await db!.flush();
-          }
-          await db!.disconnect();
-          db = null;
-        }
-      });
+  Future<void> disconnect({bool flush = false}) async {
+    _logger.info('disconnect');
+    if (db != null) {
+      _logger.info('Disconnecting');
+      if (flush) {
+        _logger.info('Flushing');
+        await db!.flush();
+      }
+      await db!.disconnect();
+      db = null;
+    }
+  }
+}
+
+// Creates a new combined [Comparator] which sorts items
+// by the given [comparators].
+Comparator<T> _combineComparators<T>(Iterable<Comparator<T>> comparators) {
+  return (T a, T b) {
+    for (final comparator in comparators) {
+      try {
+        final result = comparator(a, b);
+        if (result != 0) return result;
+      } catch (e) {
+        // If the comparator throws an exception, we ignore it and
+        // continue with the next comparator.
+        continue;
+      }
+    }
+    return 0;
+  };
+}
+
+// The default [Comparator] used to sort [ChannelState]s.
+int _defaultChannelStateComparator(ChannelState a, ChannelState b) {
+  final dateA = a.channel?.lastMessageAt ?? a.channel?.createdAt;
+  final dateB = b.channel?.lastMessageAt ?? b.channel?.createdAt;
+
+  if (dateA == null && dateB == null) return 0;
+  if (dateA == null) return 1;
+  if (dateB == null) {
+    return -1;
+  } else {
+    return dateB.compareTo(dateA);
+  }
 }

packages/stream_chat_persistence/pubspec.yaml

@@ -15,7 +15,6 @@ dependencies:
     sdk: flutter
   logging: ^1.0.1
   meta: ^1.8.0
-  mutex: ^3.0.0
   path: ^1.8.2
   path_provider: ^2.0.1
   sqlite3_flutter_libs: ^0.5.0

packages/stream_chat_persistence/test/stream_chat_persistence_client_test.dart

@@ -55,6 +55,15 @@ void main() {
     expect(client.db, isNull);
   });
 
+  test('client function throws stateError if db is not yet connected', () {
+    final client = StreamChatPersistenceClient(logLevel: Level.ALL);
+    expect(
+      // Running a function that requires db connection.
+      () => client.getReplies('testParentId'),
+      throwsA(isA<StateError>()),
+    );
+  });
+
   group('client functions', () {
     const userId = 'testUserId';
     final mockDatabase = MockChatDatabase();
@@ -66,6 +75,10 @@ void main() {
       await client.connect(userId, databaseProvider: _mockDatabaseProvider);
     });
 
+    tearDown(() async {
+      await client.disconnect();
+    });
+
     test('getReplies', () async {
       const parentId = 'testParentId';
       final replies = List.generate(3, (index) => Message(id: 'testId$index'));