talksik/stream-chat-flutter
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

refactor: remove mutex as drift already uses one internally.

Browse Source 
Signed-off-by: xsahil03x <[email protected]>
This commit is contained in:
Sahil Kumar
2023-05-05 01:26:40 +05:30
committed by xsahil03x
parent abab2e0a06
commit 91a378f720
2 changed files with 128 additions and 160 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/stream_chat_persistence/lib/src/stream_chat_persistence_client.dart
+128 -159
View File
@@ -1,7 +1,5 @@
import 'package:flutter/foundation.dart'; import 'package:flutter/foundation.dart';
import 'package:mutex/mutex.dart';
import 'package:stream_chat/stream_chat.dart'; import 'package:stream_chat/stream_chat.dart';
import 'package:stream_chat_persistence/src/db/drift_chat_database.dart'; import 'package:stream_chat_persistence/src/db/drift_chat_database.dart';
/// Various connection modes on which [StreamChatPersistenceClient] can work /// Various connection modes on which [StreamChatPersistenceClient] can work
@@ -48,7 +46,6 @@ class StreamChatPersistenceClient extends ChatPersistenceClient {
final Logger _logger; final Logger _logger;
final ConnectionMode _connectionMode; final ConnectionMode _connectionMode;
final bool _webUseIndexedDbIfSupported; final bool _webUseIndexedDbIfSupported;
final _mutex = ReadWriteMutex();
void _defaultLogHandler(LogRecord record) { void _defaultLogHandler(LogRecord record) {
print( print(
@@ -59,9 +56,6 @@ class StreamChatPersistenceClient extends ChatPersistenceClient {
if (record.stackTrace != null) print(record.stackTrace); if (record.stackTrace != null) print(record.stackTrace);
} }
Future<T> _readProtected<T>(AsyncValueGetter<T> func) =>
_mutex.protectRead(func);
bool get _debugIsConnected { bool get _debugIsConnected {
assert(() { assert(() {
if (db == null) { if (db == null) {
@@ -104,90 +98,84 @@ class StreamChatPersistenceClient extends ChatPersistenceClient {
Future<Event?> getConnectionInfo() { Future<Event?> getConnectionInfo() {
assert(_debugIsConnected, ''); assert(_debugIsConnected, '');
_logger.info('getConnectionInfo'); _logger.info('getConnectionInfo');
return _readProtected(() => db!.connectionEventDao.connectionEvent); return db!.connectionEventDao.connectionEvent;
} }
@override @override
Future<void> updateConnectionInfo(Event event) { Future<void> updateConnectionInfo(Event event) {
assert(_debugIsConnected, ''); assert(_debugIsConnected, '');
_logger.info('updateConnectionInfo'); _logger.info('updateConnectionInfo');
return _readProtected( return db!.connectionEventDao.updateConnectionEvent(event);
() => db!.connectionEventDao.updateConnectionEvent(event),
);
} }
@override @override
Future<void> updateLastSyncAt(DateTime lastSyncAt) { Future<void> updateLastSyncAt(DateTime lastSyncAt) {
assert(_debugIsConnected, ''); assert(_debugIsConnected, '');
_logger.info('updateLastSyncAt'); _logger.info('updateLastSyncAt');
return _readProtected( return db!.connectionEventDao.updateLastSyncAt(lastSyncAt);
() => db!.connectionEventDao.updateLastSyncAt(lastSyncAt),
);
} }
@override @override
Future<DateTime?> getLastSyncAt() { Future<DateTime?> getLastSyncAt() {
assert(_debugIsConnected, ''); assert(_debugIsConnected, '');
_logger.info('getLastSyncAt'); _logger.info('getLastSyncAt');
return _readProtected(() => db!.connectionEventDao.lastSyncAt); return db!.connectionEventDao.lastSyncAt;
} }
@override @override
Future<void> deleteChannels(List<String> cids) { Future<void> deleteChannels(List<String> cids) {
assert(_debugIsConnected, ''); assert(_debugIsConnected, '');
_logger.info('deleteChannels'); _logger.info('deleteChannels');
return _readProtected(() => db!.channelDao.deleteChannelByCids(cids)); return db!.channelDao.deleteChannelByCids(cids);
} }
@override @override
Future<List<String>> getChannelCids() { Future<List<String>> getChannelCids() {
assert(_debugIsConnected, ''); assert(_debugIsConnected, '');
_logger.info('getChannelCids'); _logger.info('getChannelCids');
return _readProtected(() => db!.channelDao.cids); return db!.channelDao.cids;
} }
@override @override
Future<void> deleteMessageByIds(List<String> messageIds) { Future<void> deleteMessageByIds(List<String> messageIds) {
assert(_debugIsConnected, ''); assert(_debugIsConnected, '');
_logger.info('deleteMessageByIds'); _logger.info('deleteMessageByIds');
return _readProtected(() => db!.messageDao.deleteMessageByIds(messageIds)); return db!.messageDao.deleteMessageByIds(messageIds);
} }
@override @override
Future<void> deletePinnedMessageByIds(List<String> messageIds) { Future<void> deletePinnedMessageByIds(List<String> messageIds) {
assert(_debugIsConnected, ''); assert(_debugIsConnected, '');
_logger.info('deletePinnedMessageByIds'); _logger.info('deletePinnedMessageByIds');
return _readProtected( return db!.pinnedMessageDao.deleteMessageByIds(messageIds);
() => db!.pinnedMessageDao.deleteMessageByIds(messageIds),
);
} }
@override @override
Future<void> deleteMessageByCids(List<String> cids) { Future<void> deleteMessageByCids(List<String> cids) {
assert(_debugIsConnected, ''); assert(_debugIsConnected, '');
_logger.info('deleteMessageByCids'); _logger.info('deleteMessageByCids');
return _readProtected(() => db!.messageDao.deleteMessageByCids(cids)); return db!.messageDao.deleteMessageByCids(cids);
} }
@override @override
Future<void> deletePinnedMessageByCids(List<String> cids) { Future<void> deletePinnedMessageByCids(List<String> cids) {
assert(_debugIsConnected, ''); assert(_debugIsConnected, '');
_logger.info('deletePinnedMessageByCids'); _logger.info('deletePinnedMessageByCids');
return _readProtected(() => db!.pinnedMessageDao.deleteMessageByCids(cids)); return db!.pinnedMessageDao.deleteMessageByCids(cids);
} }
@override @override
Future<List<Member>> getMembersByCid(String cid) { Future<List<Member>> getMembersByCid(String cid) {
assert(_debugIsConnected, ''); assert(_debugIsConnected, '');
_logger.info('getMembersByCid'); _logger.info('getMembersByCid');
return _readProtected(() => db!.memberDao.getMembersByCid(cid)); return db!.memberDao.getMembersByCid(cid);
} }
@override @override
Future<ChannelModel?> getChannelByCid(String cid) { Future<ChannelModel?> getChannelByCid(String cid) {
assert(_debugIsConnected, ''); assert(_debugIsConnected, '');
_logger.info('getChannelByCid'); _logger.info('getChannelByCid');
return _readProtected(() => db!.channelDao.getChannelByCid(cid)); return db!.channelDao.getChannelByCid(cid);
} }
@override @override
@@ -197,11 +185,9 @@ class StreamChatPersistenceClient extends ChatPersistenceClient {
}) { }) {
assert(_debugIsConnected, ''); assert(_debugIsConnected, '');
_logger.info('getMessagesByCid'); _logger.info('getMessagesByCid');
return _readProtected( return db!.messageDao.getMessagesByCid(
() => db!.messageDao.getMessagesByCid( cid,
cid, messagePagination: messagePagination,
messagePagination: messagePagination,
),
); );
} }
@@ -212,37 +198,34 @@ class StreamChatPersistenceClient extends ChatPersistenceClient {
}) { }) {
assert(_debugIsConnected, ''); assert(_debugIsConnected, '');
_logger.info('getPinnedMessagesByCid'); _logger.info('getPinnedMessagesByCid');
return _readProtected( return db!.pinnedMessageDao.getMessagesByCid(
() => db!.pinnedMessageDao.getMessagesByCid( cid,
cid, messagePagination: messagePagination,
messagePagination: messagePagination,
),
); );
} }
@override @override
Future<List<Read>> getReadsByCid(String cid) { Future<List<Read>> getReadsByCid(String cid) async {
assert(_debugIsConnected, ''); assert(_debugIsConnected, '');
_logger.info('getReadsByCid'); _logger.info('getReadsByCid');
return _readProtected(() => db!.readDao.getReadsByCid(cid)); return db!.readDao.getReadsByCid(cid);
} }
@override @override
Future<Map<String, List<Message>>> getChannelThreads(String cid) { Future<Map<String, List<Message>>> getChannelThreads(String cid) async {
assert(_debugIsConnected, ''); assert(_debugIsConnected, '');
_logger.info('getChannelThreads'); _logger.info('getChannelThreads');
return _readProtected(() async { final messages = await db!.messageDao.getThreadMessages(cid);
final messages = await db!.messageDao.getThreadMessages(cid); final messageByParentIdDictionary = <String, List<Message>>{};
final messageByParentIdDictionary = <String, List<Message>>{}; for (final message in messages) {
for (final message in messages) { final parentId = message.parentId!;
final parentId = message.parentId!; messageByParentIdDictionary[parentId] = [
messageByParentIdDictionary[parentId] = [ ...messageByParentIdDictionary[parentId] ?? [],
...messageByParentIdDictionary[parentId] ?? [], message,
message, ];
]; }
}
return messageByParentIdDictionary; return messageByParentIdDictionary;
});
} }
@override @override
@@ -252,11 +235,9 @@ class StreamChatPersistenceClient extends ChatPersistenceClient {
}) { }) {
assert(_debugIsConnected, ''); assert(_debugIsConnected, '');
_logger.info('getReplies'); _logger.info('getReplies');
return _readProtected( return db!.messageDao.getThreadMessagesByParentId(
() => db!.messageDao.getThreadMessagesByParentId( parentId,
parentId, options: options,
options: options,
),
); );
} }
@@ -268,73 +249,69 @@ class StreamChatPersistenceClient extends ChatPersistenceClient {
Please use channelStateSort instead.''') List<SortOption<ChannelModel>>? sort, Please use channelStateSort instead.''') List<SortOption<ChannelModel>>? sort,
List<SortOption<ChannelState>>? channelStateSort, List<SortOption<ChannelState>>? channelStateSort,
PaginationParams? paginationParams, PaginationParams? paginationParams,
}) { }) async {
assert(_debugIsConnected, ''); assert(_debugIsConnected, '');
assert( assert(
sort == null || channelStateSort == null, sort == null || channelStateSort == null,
'sort and channelStateSort cannot be used together', 'sort and channelStateSort cannot be used together',
); );
_logger.info('getChannelStates'); _logger.info('getChannelStates');
return _readProtected(
() async {
final channels = await db!.channelQueryDao.getChannels(
filter: filter,
sort: sort,
);
final channelStates = await Future.wait( final channels = await db!.channelQueryDao.getChannels(
channels.map((e) => getChannelStateByCid(e.cid)), filter: filter,
); sort: sort,
// 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 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;
} }
@override @override
@@ -345,12 +322,10 @@ class StreamChatPersistenceClient extends ChatPersistenceClient {
}) { }) {
assert(_debugIsConnected, ''); assert(_debugIsConnected, '');
_logger.info('updateChannelQueries'); _logger.info('updateChannelQueries');
return _readProtected( return db!.channelQueryDao.updateChannelQueries(
() => db!.channelQueryDao.updateChannelQueries( filter,
filter, cids,
cids, clearQueryCache: clearQueryCache,
clearQueryCache: clearQueryCache,
),
); );
} }
@@ -358,60 +333,56 @@ class StreamChatPersistenceClient extends ChatPersistenceClient {
Future<void> updateChannels(List<ChannelModel> channels) { Future<void> updateChannels(List<ChannelModel> channels) {
assert(_debugIsConnected, ''); assert(_debugIsConnected, '');
_logger.info('updateChannels'); _logger.info('updateChannels');
return _readProtected(() => db!.channelDao.updateChannels(channels)); return db!.channelDao.updateChannels(channels);
} }
@override @override
Future<void> bulkUpdateMembers(Map<String, List<Member>?> members) { Future<void> bulkUpdateMembers(Map<String, List<Member>?> members) {
assert(_debugIsConnected, ''); assert(_debugIsConnected, '');
_logger.info('bulkUpdateMembers'); _logger.info('bulkUpdateMembers');
return _readProtected(() => db!.memberDao.bulkUpdateMembers(members)); return db!.memberDao.bulkUpdateMembers(members);
} }
@override @override
Future<void> bulkUpdateMessages(Map<String, List<Message>?> messages) { Future<void> bulkUpdateMessages(Map<String, List<Message>?> messages) {
assert(_debugIsConnected, ''); assert(_debugIsConnected, '');
_logger.info('bulkUpdateMessages'); _logger.info('bulkUpdateMessages');
return _readProtected(() => db!.messageDao.bulkUpdateMessages(messages)); return db!.messageDao.bulkUpdateMessages(messages);
} }
@override @override
Future<void> bulkUpdatePinnedMessages(Map<String, List<Message>?> messages) { Future<void> bulkUpdatePinnedMessages(Map<String, List<Message>?> messages) {
assert(_debugIsConnected, ''); assert(_debugIsConnected, '');
_logger.info('bulkUpdatePinnedMessages'); _logger.info('bulkUpdatePinnedMessages');
return _readProtected( return db!.pinnedMessageDao.bulkUpdateMessages(messages);
() => db!.pinnedMessageDao.bulkUpdateMessages(messages),
);
} }
@override @override
Future<void> updatePinnedMessageReactions(List<Reaction> reactions) { Future<void> updatePinnedMessageReactions(List<Reaction> reactions) {
assert(_debugIsConnected, ''); assert(_debugIsConnected, '');
_logger.info('updatePinnedMessageReactions'); _logger.info('updatePinnedMessageReactions');
return _readProtected( return db!.pinnedMessageReactionDao.updateReactions(reactions);
() => db!.pinnedMessageReactionDao.updateReactions(reactions),
);
} }
@override @override
Future<void> updateReactions(List<Reaction> reactions) { Future<void> updateReactions(List<Reaction> reactions) {
assert(_debugIsConnected, ''); assert(_debugIsConnected, '');
_logger.info('updateReactions'); _logger.info('updateReactions');
return _readProtected(() => db!.reactionDao.updateReactions(reactions)); return db!.reactionDao.updateReactions(reactions);
} }
@override @override
Future<void> bulkUpdateReads(Map<String, List<Read>?> reads) { Future<void> bulkUpdateReads(Map<String, List<Read>?> reads) {
assert(_debugIsConnected, ''); assert(_debugIsConnected, '');
_logger.info('bulkUpdateReads'); _logger.info('bulkUpdateReads');
return _readProtected(() => db!.readDao.bulkUpdateReads(reads)); return db!.readDao.bulkUpdateReads(reads);
} }
@override @override
Future<void> updateUsers(List<User> users) { Future<void> updateUsers(List<User> users) {
assert(_debugIsConnected, ''); assert(_debugIsConnected, '');
_logger.info('updateUsers'); _logger.info('updateUsers');
return _readProtected(() => db!.userDao.updateUsers(users)); return db!.userDao.updateUsers(users);
} }
@override @override
@@ -420,53 +391,51 @@ class StreamChatPersistenceClient extends ChatPersistenceClient {
) { ) {
assert(_debugIsConnected, ''); assert(_debugIsConnected, '');
_logger.info('deletePinnedMessageReactionsByMessageId'); _logger.info('deletePinnedMessageReactionsByMessageId');
return _readProtected( return db!.pinnedMessageReactionDao.deleteReactionsByMessageIds(messageIds);
() =>
db!.pinnedMessageReactionDao.deleteReactionsByMessageIds(messageIds),
);
} }
@override @override
Future<void> deleteReactionsByMessageId(List<String> messageIds) { Future<void> deleteReactionsByMessageId(List<String> messageIds) {
assert(_debugIsConnected, ''); assert(_debugIsConnected, '');
_logger.info('deleteReactionsByMessageId'); _logger.info('deleteReactionsByMessageId');
return _readProtected( return db!.reactionDao.deleteReactionsByMessageIds(messageIds);
() => db!.reactionDao.deleteReactionsByMessageIds(messageIds),
);
} }
@override @override
Future<void> deleteMembersByCids(List<String> cids) { Future<void> deleteMembersByCids(List<String> cids) {
assert(_debugIsConnected, ''); assert(_debugIsConnected, '');
_logger.info('deleteMembersByCids'); _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 @override
Future<void> updateChannelStates(List<ChannelState> channelStates) { Future<void> updateChannelStates(List<ChannelState> channelStates) {
assert(_debugIsConnected, ''); assert(_debugIsConnected, '');
_logger.info('updateChannelStates'); _logger.info('updateChannelStates');
return _readProtected( return db!.transaction(() => super.updateChannelStates(channelStates));
() async => db!.transaction(
() async {
await super.updateChannelStates(channelStates);
},
),
);
} }
@override @override
Future<void> disconnect({bool flush = false}) async => Future<void> disconnect({bool flush = false}) async {
_mutex.protectWrite(() async { _logger.info('disconnect');
_logger.info('disconnect'); if (db != null) {
if (db != null) { _logger.info('Disconnecting');
_logger.info('Disconnecting'); if (flush) {
if (flush) { _logger.info('Flushing');
_logger.info('Flushing'); await db!.flush();
await db!.flush(); }
} await db!.disconnect();
await db!.disconnect(); db = null;
db = null; }
} }
});
} }
packages/stream_chat_persistence/pubspec.yaml
-1
View File
@@ -15,7 +15,6 @@ dependencies:
sdk: flutter sdk: flutter
logging: ^1.0.1 logging: ^1.0.1
meta: ^1.8.0 meta: ^1.8.0
mutex: ^3.0.0
path: ^1.8.2 path: ^1.8.2
path_provider: ^2.0.1 path_provider: ^2.0.1
sqlite3_flutter_libs: ^0.5.0 sqlite3_flutter_libs: ^0.5.0