From 47287ba8d4436be08c6bbb7f41bc8aa6beec5843 Mon Sep 17 00:00:00 2001 From: Christien Rioux Date: Sat, 3 Aug 2024 21:55:20 -0500 Subject: [PATCH] incremental chat state work --- lib/chat/cubits/chat_component_cubit.dart | 3 +- .../cubits/single_contact_messages_cubit.dart | 39 +++++++++++++++++-- .../active_conversations_bloc_map_cubit.dart | 10 ++--- ...ve_single_contact_chat_bloc_map_cubit.dart | 18 +++++++-- .../cubits/conversation_cubit.dart | 18 ++++----- lib/tools/loggy.dart | 6 +-- lib/tools/state_logger.dart | 6 +-- 7 files changed, 70 insertions(+), 30 deletions(-) diff --git a/lib/chat/cubits/chat_component_cubit.dart b/lib/chat/cubits/chat_component_cubit.dart index 17a481f..326a597 100644 --- a/lib/chat/cubits/chat_component_cubit.dart +++ b/lib/chat/cubits/chat_component_cubit.dart @@ -233,7 +233,8 @@ class ChatComponentCubit extends Cubit { return types.User( id: remoteIdentityPublicKey.toString(), - firstName: activeConversationState.remoteConversation.profile.name, + firstName: activeConversationState.remoteConversation?.profile.name ?? + '', metadata: {metadataKeyIdentityPublicKey: remoteIdentityPublicKey}); } diff --git a/lib/chat/cubits/single_contact_messages_cubit.dart b/lib/chat/cubits/single_contact_messages_cubit.dart index cd2d7e5..ab9c5ab 100644 --- a/lib/chat/cubits/single_contact_messages_cubit.dart +++ b/lib/chat/cubits/single_contact_messages_cubit.dart @@ -55,7 +55,7 @@ class SingleContactMessagesCubit extends Cubit { required TypedKey localConversationRecordKey, required TypedKey localMessagesRecordKey, required TypedKey remoteConversationRecordKey, - required TypedKey remoteMessagesRecordKey, + required TypedKey? remoteMessagesRecordKey, }) : _accountInfo = accountInfo, _remoteIdentityPublicKey = remoteIdentityPublicKey, _localConversationRecordKey = localConversationRecordKey, @@ -147,8 +147,14 @@ class SingleContactMessagesCubit extends Cubit { // Open remote messages key Future _initRcvdMessagesCubit() async { + // Don't bother if we don't have a remote messages record key yet + if (_remoteMessagesRecordKey == null) { + return; + } + + // Open new cubit if one is desired _rcvdMessagesCubit = DHTLogCubit( - open: () async => DHTLog.openRead(_remoteMessagesRecordKey, + open: () async => DHTLog.openRead(_remoteMessagesRecordKey!, debugName: 'SingleContactMessagesCubit::_initRcvdMessagesCubit::' 'RcvdMessages', parent: _remoteConversationRecordKey, @@ -159,6 +165,31 @@ class SingleContactMessagesCubit extends Cubit { _updateRcvdMessagesState(_rcvdMessagesCubit!.state); } + Future updateRemoteMessagesRecordKey( + TypedKey? remoteMessagesRecordKey) async { + await _initWait(); + + _sspRemoteConversationRecordKey.updateState(remoteMessagesRecordKey, + (remoteMessagesRecordKey) async { + // Don't bother if nothing is changing + if (_remoteMessagesRecordKey == remoteMessagesRecordKey) { + return; + } + + // Close existing cubit if we have one + final rcvdMessagesCubit = _rcvdMessagesCubit; + _rcvdMessagesCubit = null; + _remoteMessagesRecordKey = null; + await _rcvdSubscription?.cancel(); + _rcvdSubscription = null; + await rcvdMessagesCubit?.close(); + + // Init the new cubit if we should + _remoteMessagesRecordKey = remoteMessagesRecordKey; + await _initRcvdMessagesCubit(); + }); + } + Future _makeLocalMessagesCrypto() async => VeilidCryptoPrivate.fromTypedKey( _accountInfo.userLogin!.identitySecret, 'tabledb'); @@ -452,7 +483,7 @@ class SingleContactMessagesCubit extends Cubit { final TypedKey _localConversationRecordKey; final TypedKey _localMessagesRecordKey; final TypedKey _remoteConversationRecordKey; - final TypedKey _remoteMessagesRecordKey; + TypedKey? _remoteMessagesRecordKey; late final VeilidCrypto _conversationCrypto; late final MessageIntegrity _senderMessageIntegrity; @@ -471,4 +502,6 @@ class SingleContactMessagesCubit extends Cubit { _reconciledSubscription; final StreamController Function()> _commandController; late final Future _commandRunnerFut; + + final _sspRemoteConversationRecordKey = SingleStateProcessor(); } diff --git a/lib/conversation/cubits/active_conversations_bloc_map_cubit.dart b/lib/conversation/cubits/active_conversations_bloc_map_cubit.dart index b983265..4330db6 100644 --- a/lib/conversation/cubits/active_conversations_bloc_map_cubit.dart +++ b/lib/conversation/cubits/active_conversations_bloc_map_cubit.dart @@ -24,7 +24,7 @@ class ActiveConversationState extends Equatable { final TypedKey localConversationRecordKey; final TypedKey remoteConversationRecordKey; final proto.Conversation localConversation; - final proto.Conversation remoteConversation; + final proto.Conversation? remoteConversation; @override List get props => [ @@ -102,7 +102,8 @@ class ActiveConversationsBlocMapCubit extends BlocMapCubit, ConversationCubit>(conversationCubit, transform: (avstate) => avstate.when( - data: (data) => (data.localConversation == null || - data.remoteConversation == null) + data: (data) => (data.localConversation == null) ? const AsyncValue.loading() : AsyncValue.data(ActiveConversationState( localConversation: data.localConversation!, - remoteConversation: data.remoteConversation!, + remoteConversation: data.remoteConversation, remoteIdentityPublicKey: remoteIdentityPublicKey, localConversationRecordKey: localConversationRecordKey, remoteConversationRecordKey: diff --git a/lib/conversation/cubits/active_single_contact_chat_bloc_map_cubit.dart b/lib/conversation/cubits/active_single_contact_chat_bloc_map_cubit.dart index f3dadcb..90adafa 100644 --- a/lib/conversation/cubits/active_single_contact_chat_bloc_map_cubit.dart +++ b/lib/conversation/cubits/active_single_contact_chat_bloc_map_cubit.dart @@ -25,7 +25,7 @@ class _SingleContactChatState extends Equatable { final TypedKey localConversationRecordKey; final TypedKey remoteConversationRecordKey; final TypedKey localMessagesRecordKey; - final TypedKey remoteMessagesRecordKey; + final TypedKey? remoteMessagesRecordKey; @override List get props => [ @@ -53,8 +53,16 @@ class ActiveSingleContactChatBlocMapCubit extends BlocMapCubit _addConversationMessages(_SingleContactChatState state) async => - add(() => MapEntry( + Future _addConversationMessages(_SingleContactChatState state) async { + // xxx could use atomic update() function + + final cubit = await tryOperateAsync( + state.localConversationRecordKey, closure: (cubit) async { + await cubit.updateRemoteMessagesRecordKey(state.remoteMessagesRecordKey); + return cubit; + }); + if (cubit == null) { + await add(() => MapEntry( state.localConversationRecordKey, SingleContactMessagesCubit( accountInfo: _accountInfo, @@ -64,6 +72,8 @@ class ActiveSingleContactChatBlocMapCubit extends BlocMapCubit avInputState) { @@ -78,7 +88,7 @@ class ActiveSingleContactChatBlocMapCubit extends BlocMapCubit> { localConversation: conv, remoteConversation: _incrementalState.remoteConversation); // return loading still if state isn't complete - if ((_localConversationRecordKey != null && - _incrementalState.localConversation == null) || - (_remoteConversationRecordKey != null && - _incrementalState.remoteConversation == null)) { + if (_localConversationRecordKey != null && + _incrementalState.localConversation == null) { return const AsyncValue.loading(); } - // state is complete, all required keys are open + // local state is complete, all remote state is emitted incrementally return AsyncValue.data(_incrementalState); }, loading: AsyncValue.loading, @@ -247,14 +245,12 @@ class ConversationCubit extends Cubit> { _incrementalState = ConversationState( localConversation: _incrementalState.localConversation, remoteConversation: conv); - // return loading still if state isn't complete - if ((_localConversationRecordKey != null && - _incrementalState.localConversation == null) || - (_remoteConversationRecordKey != null && - _incrementalState.remoteConversation == null)) { + // return loading still if the local state isn't complete + if (_localConversationRecordKey != null && + _incrementalState.localConversation == null) { return const AsyncValue.loading(); } - // state is complete, all required keys are open + // local state is complete, all remote state is emitted incrementally return AsyncValue.data(_incrementalState); }, loading: AsyncValue.loading, diff --git a/lib/tools/loggy.dart b/lib/tools/loggy.dart index 69faeb7..1df4c82 100644 --- a/lib/tools/loggy.dart +++ b/lib/tools/loggy.dart @@ -112,9 +112,9 @@ class CallbackPrinter extends LoggyPrinter { @override void onLog(LogRecord record) { final out = record.pretty(); - if (isDesktop) { - debugPrintSynchronously(out); - } + //if (isDesktop) { + debugPrintSynchronously(out); + //} globalDebugTerminal.write('$out\n'.replaceAll('\n', '\r\n')); callback?.call(record); } diff --git a/lib/tools/state_logger.dart b/lib/tools/state_logger.dart index 08e32b3..6baacae 100644 --- a/lib/tools/state_logger.dart +++ b/lib/tools/state_logger.dart @@ -4,12 +4,12 @@ import 'loggy.dart'; const Map _blocChangeLogLevels = { 'ConnectionStateCubit': LogLevel.off, - 'ActiveSingleContactChatBlocMapCubit': LogLevel.off, - 'ActiveConversationsBlocMapCubit': LogLevel.off, + //'ActiveSingleContactChatBlocMapCubit': LogLevel.off, + //'ActiveConversationsBlocMapCubit': LogLevel.off, 'PersistentQueueCubit': LogLevel.off, 'TableDBArrayProtobufCubit': LogLevel.off, 'DHTLogCubit': LogLevel.off, - 'SingleContactMessagesCubit': LogLevel.off, + //'SingleContactMessagesCubit': LogLevel.off, 'ChatComponentCubit': LogLevel.off, };