Git-Mirrors/veilidchat
1
0
Fork 0
mirror of https://gitlab.com/veilid/veilidchat.git synced 2025-01-11 07:39:32 -05:00
Code Issues Packages Projects Releases Wiki Activity

busy handling

Browse Source
This commit is contained in:
Christien Rioux 2024-02-27 12:45:58 -05:00
parent 43b01c7555
commit c6f017b0d1
23 changed files with 307 additions and 179 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
lib/chat/cubits/messages_cubit.dart
View File

@ -1,6 +1,7 @@
import 'dart:async';
import 'package:async_tools/async_tools.dart';
import 'package:bloc_tools/bloc_tools.dart';
import 'package:fast_immutable_collections/fast_immutable_collections.dart';
import 'package:flutter_bloc/flutter_bloc.dart';
import 'package:veilid_support/veilid_support.dart';
@ -61,9 +62,10 @@ class MessagesCubit extends Cubit<AsyncValue<IList<proto.Message>>> {
await super.close();
}
void updateLocalMessagesState(AsyncValue<IList<proto.Message>> avmessages) {
void updateLocalMessagesState(
BlocBusyState<AsyncValue<IList<proto.Message>>> avmessages) {
// Updated local messages from online just update the state immediately
emit(avmessages);
emit(avmessages.state);
}
Future<void> _updateRemoteMessagesStateAsync(_MessageQueueEntry entry) async {
@ -97,16 +99,17 @@ class MessagesCubit extends Cubit<AsyncValue<IList<proto.Message>>> {
// Insert at this position
if (!skip) {
// Insert into dht backing array
await _localMessagesCubit!.shortArray
.tryInsertItem(pos, newMessage.writeToBuffer());
await _localMessagesCubit!.operate((shortArray) =>
shortArray.tryInsertItem(pos, newMessage.writeToBuffer()));
// Insert into local copy as well for this operation
localMessages = localMessages.insert(pos, newMessage);
}
}
}
void updateRemoteMessagesState(AsyncValue<IList<proto.Message>> avmessages) {
final remoteMessages = avmessages.data?.value;
void updateRemoteMessagesState(
BlocBusyState<AsyncValue<IList<proto.Message>>> avmessages) {
final remoteMessages = avmessages.state.data?.value;
if (remoteMessages == null) {
return;
}
@ -171,7 +174,8 @@ class MessagesCubit extends Cubit<AsyncValue<IList<proto.Message>>> {
}
Future<void> addMessage({required proto.Message message}) async {
await _localMessagesCubit!.shortArray.tryAddItem(message.writeToBuffer());
await _localMessagesCubit!.operate(
(shortArray) => shortArray.tryAddItem(message.writeToBuffer()));
}
Future<DHTRecordCrypto> getMessagesCrypto() async {

3
lib/chat/views/chat_component.dart
View File

@ -48,7 +48,8 @@ class ChatComponent extends StatelessWidget {
if (accountRecordInfo == null) {
return debugPage('should always have an account record here');
}
final contactList = context.watch<ContactListCubit>().state.data?.value;
final contactList =
context.watch<ContactListCubit>().state.state.data?.value;
if (contactList == null) {
return debugPage('should always have a contact list here');
}

11
lib/chat_list/cubits/active_conversations_bloc_map_cubit.dart
View File

@ -36,7 +36,9 @@ typedef ActiveConversationsBlocMapState
// Automatically follows the state of a ChatListCubit.
class ActiveConversationsBlocMapCubit extends BlocMapCubit<TypedKey,
AsyncValue<ActiveConversationState>, ActiveConversationCubit>
with StateFollower<AsyncValue<IList<proto.Chat>>, TypedKey, proto.Chat> {
with
StateFollower<BlocBusyState<AsyncValue<IList<proto.Chat>>>, TypedKey,
proto.Chat> {
ActiveConversationsBlocMapCubit(
{required ActiveAccountInfo activeAccountInfo,
required ContactListCubit contactListCubit})
@ -73,8 +75,9 @@ class ActiveConversationsBlocMapCubit extends BlocMapCubit<TypedKey,
/// StateFollower /////////////////////////
@override
IMap<TypedKey, proto.Chat> getStateMap(AsyncValue<IList<proto.Chat>> state) {
final stateValue = state.data?.value;
IMap<TypedKey, proto.Chat> getStateMap(
BlocBusyState<AsyncValue<IList<proto.Chat>>> state) {
final stateValue = state.state.data?.value;
if (stateValue == null) {
return IMap();
}
@ -88,7 +91,7 @@ class ActiveConversationsBlocMapCubit extends BlocMapCubit<TypedKey,
@override
Future<void> updateState(TypedKey key, proto.Chat value) async {
final contactList = _contactListCubit.state.data?.value;
final contactList = _contactListCubit.state.state.data?.value;
if (contactList == null) {
await addState(key, const AsyncValue.loading());
return;

9
lib/chat_list/cubits/chat_list_cubit.dart
View File

@ -1,7 +1,5 @@
import 'dart:async';
import 'package:bloc/bloc.dart';
import 'package:bloc_tools/bloc_tools.dart';
import 'package:veilid_support/veilid_support.dart';
import '../../account_manager/account_manager.dart';
@ -44,7 +42,9 @@ class ChatListCubit extends DHTShortArrayCubit<proto.Chat> {
// Add Chat to account's list
// if this fails, don't keep retrying, user can try again later
if (await shortArray.tryAddItem(chat.writeToBuffer()) == false) {
final added = await operate(
(shortArray) => shortArray.tryAddItem(chat.writeToBuffer()));
if (!added) {
throw Exception('Failed to add chat');
}
}
@ -57,7 +57,7 @@ class ChatListCubit extends DHTShortArrayCubit<proto.Chat> {
// Remove Chat from account's list
// if this fails, don't keep retrying, user can try again later
await operate((shortArray) async {
for (var i = 0; i < shortArray.length; i++) {
final cbuf = await shortArray.getItem(i);
if (cbuf == null) {
@ -69,5 +69,6 @@ class ChatListCubit extends DHTShortArrayCubit<proto.Chat> {
return;
}
}
});
}
}

20
lib/chat_list/views/chat_single_contact_item_widget.dart
View File

@ -10,10 +10,15 @@ import '../../theme/theme.dart';
import '../chat_list.dart';
class ChatSingleContactItemWidget extends StatelessWidget {
const ChatSingleContactItemWidget({required proto.Contact contact, super.key})
: _contact = contact;
const ChatSingleContactItemWidget({
required proto.Contact contact,
required bool disabled,
super.key,
}) : _contact = contact,
_disabled = disabled;
final proto.Contact _contact;
final bool _disabled;
@override
// ignore: prefer_expression_function_bodies
@ -43,7 +48,9 @@ class ChatSingleContactItemWidget extends StatelessWidget {
motion: const DrawerMotion(),
children: [
SlidableAction(
onPressed: (context) async {
onPressed: _disabled
? null
: (context) async {
final chatListCubit = context.read<ChatListCubit>();
await chatListCubit.deleteChat(
remoteConversationRecordKey:
@ -67,9 +74,12 @@ class ChatSingleContactItemWidget extends StatelessWidget {
// The child of the Slidable is what the user sees when the
// component is not dragged.
child: ListTile(
onTap: () {
onTap: _disabled
? null
: () {
singleFuture(activeChatCubit, () async {
activeChatCubit.setActiveChat(remoteConversationRecordKey);
activeChatCubit
.setActiveChat(remoteConversationRecordKey);
});
},
title: Text(_contact.editedProfile.name),

3
lib/chat_list/views/chat_single_contact_list_widget.dart
View File

@ -45,7 +45,8 @@ class ChatSingleContactListWidget extends StatelessWidget {
return const Text('...');
}
return ChatSingleContactItemWidget(
contact: contact);
contact: contact,
disabled: contactListV.busy);
},
filter: (value) {
final lowerValue = value.toLowerCase();

9
lib/contact_invitation/cubits/contact_invitation_list_cubit.dart
View File

@ -138,11 +138,13 @@ class ContactInvitationListCubit
// Add ContactInvitationRecord to account's list
// if this fails, don't keep retrying, user can try again later
await operate((shortArray) async {
if (await shortArray.tryAddItem(cinvrec.writeToBuffer()) == false) {
throw Exception('Failed to add contact invitation record');
}
});
});
});
return signedContactInvitationBytes;
}
@ -155,6 +157,7 @@ class ContactInvitationListCubit
_activeAccountInfo.userLogin.accountRecordInfo.accountRecord.recordKey;
// Remove ContactInvitationRecord from account's list
await operate((shortArray) async {
for (var i = 0; i < shortArray.length; i++) {
final item = await shortArray.getItemProtobuf(
proto.ContactInvitationRecord.fromBuffer, i);
@ -173,13 +176,15 @@ class ContactInvitationListCubit
await contactRequestInbox.delete();
});
if (!accepted) {
await (await pool.openRead(item.localConversationRecordKey.toVeilid(),
await (await pool.openRead(
item.localConversationRecordKey.toVeilid(),
parent: accountRecordKey))
.delete();
}
return;
}
}
});
}
Future<ValidContactInvitation?> validateInvitation(
@ -205,7 +210,7 @@ class ContactInvitationListCubit
// inbox with our list of extant invitations
// If we're chatting to ourselves,
// we are validating an invitation we have created
final isSelf = state.data!.value.indexWhere((cir) =>
final isSelf = state.state.data!.value.indexWhere((cir) =>
cir.contactRequestInbox.recordKey.toVeilid() ==
contactRequestInboxKey) !=
-1;

10
lib/contact_invitation/cubits/waiting_invitations_bloc_map_cubit.dart
View File

@ -16,8 +16,10 @@ typedef WaitingInvitationsBlocMapState
class WaitingInvitationsBlocMapCubit extends BlocMapCubit<TypedKey,
AsyncValue<InvitationStatus>, WaitingInvitationCubit>
with
StateFollower<AsyncValue<IList<proto.ContactInvitationRecord>>,
TypedKey, proto.ContactInvitationRecord> {
StateFollower<
BlocBusyState<AsyncValue<IList<proto.ContactInvitationRecord>>>,
TypedKey,
proto.ContactInvitationRecord> {
WaitingInvitationsBlocMapCubit(
{required this.activeAccountInfo, required this.account});
@ -37,8 +39,8 @@ class WaitingInvitationsBlocMapCubit extends BlocMapCubit<TypedKey,
/// StateFollower /////////////////////////
@override
IMap<TypedKey, proto.ContactInvitationRecord> getStateMap(
AsyncValue<IList<proto.ContactInvitationRecord>> state) {
final stateValue = state.data?.value;
BlocBusyState<AsyncValue<IList<proto.ContactInvitationRecord>>> state) {
final stateValue = state.state.data?.value;
if (stateValue == null) {
return IMap();
}

11
lib/contact_invitation/views/contact_invitation_item_widget.dart
View File

@ -9,15 +9,20 @@ import '../contact_invitation.dart';
class ContactInvitationItemWidget extends StatelessWidget {
const ContactInvitationItemWidget(
{required this.contactInvitationRecord, super.key});
{required this.contactInvitationRecord,
required this.disabled,
super.key});
final proto.ContactInvitationRecord contactInvitationRecord;
final bool disabled;
@override
void debugFillProperties(DiagnosticPropertiesBuilder properties) {
super.debugFillProperties(properties);
properties.add(DiagnosticsProperty<proto.ContactInvitationRecord>(
'contactInvitationRecord', contactInvitationRecord));
properties
..add(DiagnosticsProperty<proto.ContactInvitationRecord>(
'contactInvitationRecord', contactInvitationRecord))
..add(DiagnosticsProperty<bool>('disabled', disabled));
}
@override

9
lib/contact_invitation/views/contact_invitation_list_widget.dart
View File

@ -10,10 +10,12 @@ import 'contact_invitation_item_widget.dart';
class ContactInvitationListWidget extends StatefulWidget {
const ContactInvitationListWidget({
required this.contactInvitationRecordList,
required this.disabled,
super.key,
});
final IList<proto.ContactInvitationRecord> contactInvitationRecordList;
final bool disabled;
@override
ContactInvitationListWidgetState createState() =>
@ -21,8 +23,10 @@ class ContactInvitationListWidget extends StatefulWidget {
@override
void debugFillProperties(DiagnosticPropertiesBuilder properties) {
super.debugFillProperties(properties);
properties.add(IterableProperty<proto.ContactInvitationRecord>(
'contactInvitationRecordList', contactInvitationRecordList));
properties
..add(IterableProperty<proto.ContactInvitationRecord>(
'contactInvitationRecordList', contactInvitationRecordList))
..add(DiagnosticsProperty<bool>('disabled', disabled));
}
}
@ -63,6 +67,7 @@ class ContactInvitationListWidgetState
return ContactInvitationItemWidget(
contactInvitationRecord:
widget.contactInvitationRecordList[index],
disabled: widget.disabled,
key: ObjectKey(widget.contactInvitationRecordList[index]))
.paddingLTRB(4, 2, 4, 2);
},

4
lib/contacts/cubits/contact_list_cubit.dart
View File

@ -53,9 +53,11 @@ class ContactListCubit extends DHTShortArrayCubit<proto.Contact> {
// Add Contact to account's list
// if this fails, don't keep retrying, user can try again later
await operate((shortArray) async {
if (await shortArray.tryAddItem(contact.writeToBuffer()) == false) {
throw Exception('Failed to add contact record');
}
});
}
Future<void> deleteContact({required proto.Contact contact}) async {
@ -67,6 +69,7 @@ class ContactListCubit extends DHTShortArrayCubit<proto.Contact> {
contact.remoteConversationRecordKey.toVeilid();
// Remove Contact from account's list
await operate((shortArray) async {
for (var i = 0; i < shortArray.length; i++) {
final item =
await shortArray.getItemProtobuf(proto.Contact.fromBuffer, i);
@ -95,6 +98,7 @@ class ContactListCubit extends DHTShortArrayCubit<proto.Contact> {
} on Exception catch (e) {
log.debug('error removing remote conversation record key: $e', e);
}
});
}
//

25
lib/contacts/views/contact_item_widget.dart
View File

@ -11,9 +11,11 @@ import '../../theme/theme.dart';
import '../contacts.dart';
class ContactItemWidget extends StatelessWidget {
const ContactItemWidget({required this.contact, super.key});
const ContactItemWidget(
{required this.contact, required this.disabled, super.key});
final proto.Contact contact;
final bool disabled;
@override
// ignore: prefer_expression_function_bodies
@ -41,16 +43,21 @@ class ContactItemWidget extends StatelessWidget {
motion: const DrawerMotion(),
children: [
SlidableAction(
onPressed: (context) async {
final contactListCubit = context.read<ContactListCubit>();
onPressed: disabled || context.read<ChatListCubit>().isBusy
? null
: (context) async {
final contactListCubit =
context.read<ContactListCubit>();
final chatListCubit = context.read<ChatListCubit>();
// Remove any chats for this contact
await chatListCubit.deleteChat(
remoteConversationRecordKey: remoteConversationKey);
remoteConversationRecordKey:
remoteConversationKey);
// Delete the contact itself
await contactListCubit.deleteContact(contact: contact);
await contactListCubit.deleteContact(
contact: contact);
},
backgroundColor: scale.tertiaryScale.background,
foregroundColor: scale.tertiaryScale.text,
@ -70,14 +77,18 @@ class ContactItemWidget extends StatelessWidget {
// The child of the Slidable is what the user sees when the
// component is not dragged.
child: ListTile(
onTap: () async {
onTap: disabled || context.read<ChatListCubit>().isBusy
? null
: () async {
// Start a chat
final chatListCubit = context.read<ChatListCubit>();
await chatListCubit.getOrCreateChatSingleContact(
remoteConversationRecordKey: remoteConversationKey);
// Click over to chats
if (context.mounted) {
await MainPager.of(context)?.pageController.animateToPage(1,
await MainPager.of(context)
?.pageController
.animateToPage(1,
duration: 250.ms, curve: Curves.easeInOut);
}
},

11
lib/contacts/views/contact_list_widget.dart
View File

@ -12,13 +12,17 @@ import 'contact_item_widget.dart';
import 'empty_contact_list_widget.dart';
class ContactListWidget extends StatelessWidget {
const ContactListWidget({required this.contactList, super.key});
const ContactListWidget(
{required this.contactList, required this.disabled, super.key});
final IList<proto.Contact> contactList;
final bool disabled;
@override
void debugFillProperties(DiagnosticPropertiesBuilder properties) {
super.debugFillProperties(properties);
properties.add(IterableProperty<proto.Contact>('contactList', contactList));
properties
..add(IterableProperty<proto.Contact>('contactList', contactList))
..add(DiagnosticsProperty<bool>('disabled', disabled));
}
@override
@ -36,7 +40,8 @@ class ContactListWidget extends StatelessWidget {
? const EmptyContactListWidget()
: SearchableList<proto.Contact>(
initialList: contactList.toList(),
builder: (l, i, c) => ContactItemWidget(contact: c),
builder: (l, i, c) =>
ContactItemWidget(contact: c, disabled: disabled),
filter: (value) {
final lowerValue = value.toLowerCase();
return contactList

10
lib/layout/home/home_account_ready/home_account_ready_shell.dart
View File

@ -1,4 +1,5 @@
import 'package:async_tools/async_tools.dart';
import 'package:bloc_tools/bloc_tools.dart';
import 'package:fast_immutable_collections/fast_immutable_collections.dart';
import 'package:flutter/foundation.dart';
import 'package:flutter/material.dart';
@ -75,8 +76,7 @@ class HomeAccountReadyShellState extends State<HomeAccountReadyShell> {
// Process all accepted or rejected invitations
void _invitationStatusListener(
BuildContext context, WaitingInvitationsBlocMapState state) {
_singleInvitationStatusProcessor.updateState(state,
closure: (newState) async {
_singleInvitationStatusProcessor.updateState(state, (newState) async {
final contactListCubit = context.read<ContactListCubit>();
final contactInvitationListCubit =
context.read<ContactInvitationListCubit>();
@ -146,7 +146,8 @@ class HomeAccountReadyShellState extends State<HomeAccountReadyShell> {
activeAccountInfo: widget.activeAccountInfo,
contactListCubit: context.read<ContactListCubit>())
..follow(
initialInputState: const AsyncValue.loading(),
initialInputState:
const BlocBusyState(AsyncValue.loading()),
stream: context.read<ChatListCubit>().stream)),
BlocProvider(
create: (context) =>
@ -167,7 +168,8 @@ class HomeAccountReadyShellState extends State<HomeAccountReadyShell> {
activeAccountInfo: widget.activeAccountInfo,
account: account)
..follow(
initialInputState: const AsyncValue.loading(),
initialInputState:
const BlocBusyState(AsyncValue.loading()),
stream: context
.read<ContactInvitationListCubit>()
.stream))

16
lib/layout/home/home_account_ready/main_pager/account_page.dart
View File

@ -38,11 +38,14 @@ class AccountPageState extends State<AccountPage> {
final textTheme = theme.textTheme;
final scale = theme.extension<ScaleScheme>()!;
final cilState = context.watch<ContactInvitationListCubit>().state;
final cilBusy = cilState.busy;
final contactInvitationRecordList =
context.watch<ContactInvitationListCubit>().state.data?.value ??
const IListConst([]);
final contactList = context.watch<ContactListCubit>().state.data?.value ??
const IListConst([]);
cilState.state.data?.value ?? const IListConst([]);
final ciState = context.watch<ContactListCubit>().state;
final ciBusy = ciState.busy;
final contactList = ciState.state.data?.value ?? const IListConst([]);
return SizedBox(
child: Column(children: <Widget>[
@ -66,10 +69,11 @@ class AccountPageState extends State<AccountPage> {
initiallyExpanded: true,
children: [
ContactInvitationListWidget(
contactInvitationRecordList: contactInvitationRecordList)
contactInvitationRecordList: contactInvitationRecordList,
disabled: cilBusy)
],
).paddingLTRB(8, 0, 8, 8),
ContactListWidget(contactList: contactList).expanded(),
ContactListWidget(contactList: contactList, disabled: ciBusy).expanded(),
]));
}
}

12
lib/tools/widget_helpers.dart
View File

@ -1,5 +1,6 @@
import 'package:async_tools/async_tools.dart';
import 'package:awesome_extensions/awesome_extensions.dart';
import 'package:bloc_tools/bloc_tools.dart';
import 'package:blurry_modal_progress_hud/blurry_modal_progress_hud.dart';
import 'package:flutter/material.dart';
import 'package:flutter_bloc/flutter_bloc.dart';
@ -77,6 +78,17 @@ extension AsyncValueBuilderExt<T> on AsyncValue<T> {
data: (d) => debugPage('AsyncValue should not be data here'));
}
extension BusyAsyncValueBuilderExt<T> on BlocBusyState<AsyncValue<T>> {
Widget builder(Widget Function(BuildContext, T) builder) =>
AbsorbPointer(absorbing: busy, child: state.builder(builder));
Widget buildNotData(
{Widget Function()? loading,
Widget Function(Object, StackTrace?)? error}) =>
AbsorbPointer(
absorbing: busy,
child: state.buildNotData(loading: loading, error: error));
}
class AsyncBlocBuilder<B extends StateStreamable<AsyncValue<S>>, S>
extends BlocBuilder<B, AsyncValue<S>> {
AsyncBlocBuilder({

1
packages/async_tools/lib/async_tools.dart
View File

@ -6,3 +6,4 @@ export 'src/async_value.dart';
export 'src/serial_future.dart';
export 'src/single_future.dart';
export 'src/single_state_processor.dart';
export 'src/single_stateless_processor.dart';

3
packages/async_tools/lib/src/single_state_processor.dart
View File

@ -14,8 +14,7 @@ import '../async_tools.dart';
class SingleStateProcessor<State> {
SingleStateProcessor();
void updateState(State newInputState,
{required Future<void> Function(State) closure}) {
void updateState(State newInputState, Future<void> Function(State) closure) {
// Use a singlefuture here to ensure we get dont lose any updates
// If the input stream gives us an update while we are
// still processing the last update, the most recent input state will

47
packages/async_tools/lib/src/single_stateless_processor.dart Normal file
View File

@ -0,0 +1,47 @@
import 'dart:async';
import '../async_tools.dart';
// Process a single stateless update at a time ensuring each request
// gets processed asynchronously, and continuously while update is requested.
//
// This is useful for processing updates asynchronously without waiting
// from a synchronous execution context
class SingleStatelessProcessor {
SingleStatelessProcessor();
void update(Future<void> Function() closure) {
singleFuture(this, () async {
do {
_more = false;
await closure();
// See if another update was requested
} while (_more);
}, onBusy: () {
// Keep this state until we process again
_more = true;
});
}
// Like update, but with a busy wrapper that clears once the updating is finished
void busyUpdate<T, S>(
Future<void> Function(Future<void> Function(void Function(S))) busy,
Future<void> Function(void Function(S)) closure) {
singleFuture(
this,
() async => busy((emit) async {
do {
_more = false;
await closure(emit);
// See if another update was requested
} while (_more);
}), onBusy: () {
// Keep this state until we process again
_more = true;
});
}
bool _more = false;
}

2
packages/bloc_tools/lib/src/async_transformer_cubit.dart
View File

@ -10,7 +10,7 @@ class AsyncTransformerCubit<T, S> extends Cubit<AsyncValue<T>> {
_subscription = input.stream.listen(_asyncTransform);
}
void _asyncTransform(AsyncValue<S> newInputState) {
_singleStateProcessor.updateState(newInputState, closure: (newState) async {
_singleStateProcessor.updateState(newInputState, (newState) async {
// Emit the transformed state
try {
if (newState is AsyncLoading<S>) {

25
packages/bloc_tools/lib/src/bloc_busy_wrapper.dart
View File

@ -17,7 +17,7 @@ class BlocBusyState<S> extends Equatable {
}
mixin BlocBusyWrapper<S> on BlocBase<BlocBusyState<S>> {
Future<T> busy<T>(Future<T> Function(void Function(S) emit) closure) async =>
Future<T> busyValue<T>(Future<T> Function(void Function(S) emit) closure) =>
_mutex.protect(() async {
void busyemit(S state) {
changedState = state;
@ -41,6 +41,27 @@ mixin BlocBusyWrapper<S> on BlocBase<BlocBusyState<S>> {
return out;
});
Future<void> busy(Future<void> Function(void Function(S) emit) closure) =>
_mutex.protect(() async {
void busyemit(S state) {
changedState = state;
}
// Turn on busy state
emit(BlocBusyState._busy(state.state));
// Run the closure
await closure(busyemit);
// If the closure did one or more 'busy emits' then
// take the most recent one and emit it for real
final finalState = changedState;
if (finalState != null && finalState != state.state) {
emit(BlocBusyState._busy(finalState));
} else {
emit(BlocBusyState._busy(state.state));
}
});
void changeState(S state) {
if (_mutex.isLocked) {
changedState = state;
@ -49,6 +70,8 @@ mixin BlocBusyWrapper<S> on BlocBase<BlocBusyState<S>> {
}
}
bool get isBusy => _mutex.isLocked;
final Mutex _mutex = Mutex();
S? changedState;
}

2
packages/bloc_tools/lib/src/state_follower.dart
View File

@ -31,7 +31,7 @@ abstract mixin class StateFollower<S extends Object, K, V> {
void _updateFollow(S newInputState) {
_singleStateProcessor.updateState(getStateMap(newInputState),
closure: (newStateMap) async {
(newStateMap) async {
for (final k in _lastInputStateMap.keys) {
if (!newStateMap.containsKey(k)) {
// deleted

47
packages/veilid_support/lib/dht_support/src/dht_short_array_cubit.dart
View File

@ -4,18 +4,19 @@ import 'package:async_tools/async_tools.dart';
import 'package:bloc/bloc.dart';
import 'package:bloc_tools/bloc_tools.dart';
import 'package:fast_immutable_collections/fast_immutable_collections.dart';
import 'package:mutex/mutex.dart';
import '../../veilid_support.dart';
class DHTShortArrayCubit<T> extends Cubit<BlocBusyState<AsyncValue<IList<T>>>>
with BlocBusyWrapper<AsyncValue<IList<T>>> {
typedef DHTShortArrayState<T> = AsyncValue<IList<T>>;
typedef DHTShortArrayBusyState<T> = BlocBusyState<DHTShortArrayState<T>>;
class DHTShortArrayCubit<T> extends Cubit<DHTShortArrayBusyState<T>>
with BlocBusyWrapper<DHTShortArrayState<T>> {
DHTShortArrayCubit({
required Future<DHTShortArray> Function() open,
required T Function(List<int> data) decodeElement,
}) : _decodeElement = decodeElement,
_wantsUpdate = false,
_isUpdating = false,
_wantsCloseRecord = false,
super(const BlocBusyState(AsyncValue.loading())) {
Future.delayed(Duration.zero, () async {
// Open DHT record
@ -33,9 +34,6 @@ class DHTShortArrayCubit<T> extends Cubit<BlocBusyState<AsyncValue<IList<T>>>>
required T Function(List<int> data) decodeElement,
}) : _shortArray = shortArray,
_decodeElement = decodeElement,
_wantsUpdate = false,
_isUpdating = false,
_wantsCloseRecord = false,
super(const BlocBusyState(AsyncValue.loading())) {
// Make initial state update
_update();
@ -59,37 +57,21 @@ class DHTShortArrayCubit<T> extends Cubit<BlocBusyState<AsyncValue<IList<T>>>>
void _update() {
// Run at most one background update process
xxx convert to singleFuture with onBusy that sets wantsupdate
_wantsUpdate = true;
if (_isUpdating) {
return;
}
_isUpdating = true;
Future.delayed(Duration.zero, () async {
// Keep updating until we don't want to update any more
// Because this is async, we could get an update while we're
// still processing the last one
_sspUpdate.busyUpdate<T, AsyncValue<IList<T>>>(busy, (emit) async {
try {
do {
_wantsUpdate = false;
try {
final initialState = await _getElements();
final initialState = await _getElementsInner();
emit(AsyncValue.data(initialState));
} on Exception catch (e) {
emit(AsyncValue.error(e));
}
} while (_wantsUpdate);
} finally {
// Note that this update future has finished
_isUpdating = false;
}
});
}
// Get and decode the entire short array
Future<IList<T>> _getElements() async {
Future<IList<T>> _getElementsInner() async {
assert(isBusy, 'should only be called from a busy state');
var out = IList<T>();
for (var i = 0; i < _shortArray.length; i++) {
// Get the element bytes (throw if fails, array state is invalid)
@ -112,12 +94,13 @@ xxx convert to singleFuture with onBusy that sets wantsupdate
await super.close();
}
DHTShortArray get shortArray => _shortArray;
Future<R> operate<R>(Future<R> Function(DHTShortArray) closure) async =>
_operateMutex.protect(() async => closure(_shortArray));
final _operateMutex = Mutex();
late final DHTShortArray _shortArray;
final T Function(List<int> data) _decodeElement;
StreamSubscription<void>? _subscription;
bool _wantsUpdate;
bool _isUpdating;
bool _wantsCloseRecord;
bool _wantsCloseRecord = false;
final _sspUpdate = SingleStatelessProcessor();
}