Git-Mirrors/veilidchat
1
0
Fork 0
mirror of https://gitlab.com/veilid/veilidchat.git synced 2024-10-01 06:55:46 -04:00
Code Issues Packages Projects Releases Wiki Activity

beta warning dialog

Browse Source
This commit is contained in:
Christien Rioux 2024-07-24 15:20:29 -04:00
parent ba191d3903
commit 6080c2f0c6
26 changed files with 445 additions and 339 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
assets/i18n/en.json
View File

@ -8,6 +8,10 @@
"accounts": "Accounts",
"version": "Version"
},
"splash": {
"beta_title": "VeilidChat is BETA SOFTWARE",
"beta_text": "DO NOT USE THIS FOR ANYTHING IMPORTANT\n\nUntil 1.0 is released:\n\n• You should have no expectations of actual privacy, or guarantees of security.\n• You will likely lose accounts, contacts, and messages and need to recreate them.\n\nPlease read our BETA PARTICIPATION GUIDE located here:\n\n"
},
"pager": {
"chats": "Chats",
"contacts": "Contacts"
@ -99,7 +103,8 @@
},
"contacts_page": {
"contacts": "Contacts",
"invitations": "Invitations"
"invitations": "Invitations",
"loading_contacts": "Loading contacts..."
},
"add_contact_sheet": {
"new_contact": "New Contact",

11
lib/chat/cubits/reconciliation/author_input_source.dart
View File

@ -57,16 +57,11 @@ class AuthorInputSource {
// Get another input batch futher back
final nextWindow = await cubit.loadElementsFromReader(
reader, last + 1, (last + 1) - first);
final asErr = nextWindow.asError;
if (asErr != null) {
return AsyncValue.error(asErr.error, asErr.stackTrace);
}
final asLoading = nextWindow.asLoading;
if (asLoading != null) {
if (nextWindow == null) {
return const AsyncValue.loading();
}
_currentWindow = InputWindow(
elements: nextWindow.asData!.value, first: first, last: last);
_currentWindow =
InputWindow(elements: nextWindow, first: first, last: last);
return const AsyncValue.data(true);
});

264
lib/chat/views/chat_component_widget.dart
View File

@ -19,143 +19,66 @@ import '../chat.dart';
const onEndReachedThreshold = 0.75;
class ChatComponentWidget extends StatelessWidget {
const ChatComponentWidget._({required super.key});
// Builder wrapper function that takes care of state management requirements
static Widget builder(
{required TypedKey localConversationRecordKey, Key? key}) =>
Builder(builder: (context) {
// Get the account info
final accountInfo = context.watch<AccountInfoCubit>().state;
// Get the account record cubit
final accountRecordCubit = context.read<AccountRecordCubit>();
// Get the contact list cubit
final contactListCubit = context.watch<ContactListCubit>();
// Get the active conversation cubit
final activeConversationCubit = context
.select<ActiveConversationsBlocMapCubit, ActiveConversationCubit?>(
(x) => x.tryOperateSync(localConversationRecordKey,
closure: (cubit) => cubit));
if (activeConversationCubit == null) {
return waitingPage();
}
// Get the messages cubit
final messagesCubit = context.select<
ActiveSingleContactChatBlocMapCubit,
SingleContactMessagesCubit?>(
(x) => x.tryOperateSync(localConversationRecordKey,
closure: (cubit) => cubit));
if (messagesCubit == null) {
return waitingPage();
}
// Make chat component state
return BlocProvider(
key: key,
create: (context) => ChatComponentCubit.singleContact(
accountInfo: accountInfo,
accountRecordCubit: accountRecordCubit,
contactListCubit: contactListCubit,
activeConversationCubit: activeConversationCubit,
messagesCubit: messagesCubit,
),
child: ChatComponentWidget._(key: key));
});
const ChatComponentWidget(
{required super.key, required TypedKey localConversationRecordKey})
: _localConversationRecordKey = localConversationRecordKey;
/////////////////////////////////////////////////////////////////////
void _handleSendPressed(
ChatComponentCubit chatComponentCubit, types.PartialText message) {
final text = message.text;
if (text.startsWith('/')) {
chatComponentCubit.runCommand(text);
return;
}
chatComponentCubit.sendMessage(message);
}
// void _handleAttachmentPressed() async {
// //
// }
Future<void> _handlePageForward(
ChatComponentCubit chatComponentCubit,
WindowState<types.Message> messageWindow,
ScrollNotification notification) async {
print(
'_handlePageForward: messagesState.length=${messageWindow.length} messagesState.windowTail=${messageWindow.windowTail} messagesState.windowCount=${messageWindow.windowCount} ScrollNotification=$notification');
// Go forward a page
final tail = min(messageWindow.length,
messageWindow.windowTail + (messageWindow.windowCount ~/ 4)) %
messageWindow.length;
// Set follow
final follow = messageWindow.length == 0 ||
tail == 0; // xxx incorporate scroll position
// final scrollOffset = (notification.metrics.maxScrollExtent -
// notification.metrics.minScrollExtent) *
// (1.0 - onEndReachedThreshold);
// chatComponentCubit.scrollOffset = scrollOffset;
await chatComponentCubit.setWindow(
tail: tail, count: messageWindow.windowCount, follow: follow);
// chatComponentCubit.state.scrollController.position.jumpTo(
// chatComponentCubit.state.scrollController.offset + scrollOffset);
//chatComponentCubit.scrollOffset = 0;
}
Future<void> _handlePageBackward(
ChatComponentCubit chatComponentCubit,
WindowState<types.Message> messageWindow,
ScrollNotification notification,
) async {
print(
'_handlePageBackward: messagesState.length=${messageWindow.length} messagesState.windowTail=${messageWindow.windowTail} messagesState.windowCount=${messageWindow.windowCount} ScrollNotification=$notification');
// Go back a page
final tail = max(
messageWindow.windowCount,
(messageWindow.windowTail - (messageWindow.windowCount ~/ 4)) %
messageWindow.length);
// Set follow
final follow = messageWindow.length == 0 ||
tail == 0; // xxx incorporate scroll position
// final scrollOffset = -(notification.metrics.maxScrollExtent -
// notification.metrics.minScrollExtent) *
// (1.0 - onEndReachedThreshold);
// chatComponentCubit.scrollOffset = scrollOffset;
await chatComponentCubit.setWindow(
tail: tail, count: messageWindow.windowCount, follow: follow);
// chatComponentCubit.scrollOffset = scrollOffset;
// chatComponentCubit.state.scrollController.position.jumpTo(
// chatComponentCubit.state.scrollController.offset + scrollOffset);
//chatComponentCubit.scrollOffset = 0;
}
@override
Widget build(BuildContext context) {
final theme = Theme.of(context);
final scale = theme.extension<ScaleScheme>()!;
final scaleConfig = theme.extension<ScaleConfig>()!;
final textTheme = theme.textTheme;
// Get the account info
final accountInfo = context.watch<AccountInfoCubit>().state;
// Get the account record cubit
final accountRecordCubit = context.read<AccountRecordCubit>();
// Get the contact list cubit
final contactListCubit = context.watch<ContactListCubit>();
// Get the active conversation cubit
final activeConversationCubit = context
.select<ActiveConversationsBlocMapCubit, ActiveConversationCubit?>(
(x) => x.tryOperateSync(_localConversationRecordKey,
closure: (cubit) => cubit));
if (activeConversationCubit == null) {
return waitingPage();
}
// Get the messages cubit
final messagesCubit = context.select<ActiveSingleContactChatBlocMapCubit,
SingleContactMessagesCubit?>(
(x) => x.tryOperateSync(_localConversationRecordKey,
closure: (cubit) => cubit));
if (messagesCubit == null) {
return waitingPage();
}
// Make chat component state
return BlocProvider(
key: key,
create: (context) => ChatComponentCubit.singleContact(
accountInfo: accountInfo,
accountRecordCubit: accountRecordCubit,
contactListCubit: contactListCubit,
activeConversationCubit: activeConversationCubit,
messagesCubit: messagesCubit,
),
child: Builder(builder: _buildChatComponent));
}
/////////////////////////////////////////////////////////////////////
Widget _buildChatComponent(BuildContext context) {
final theme = Theme.of(context);
final scale = theme.extension<ScaleScheme>()!;
final scaleConfig = theme.extension<ScaleConfig>()!;
final textTheme = theme.textTheme;
final chatTheme = makeChatTheme(scale, scaleConfig, textTheme);
final errorChatTheme = (ChatThemeEditor(chatTheme)
..inputTextColor = scale.errorScale.primary
@ -323,4 +246,89 @@ class ChatComponentWidget extends StatelessWidget {
],
);
}
void _handleSendPressed(
ChatComponentCubit chatComponentCubit, types.PartialText message) {
final text = message.text;
if (text.startsWith('/')) {
chatComponentCubit.runCommand(text);
return;
}
chatComponentCubit.sendMessage(message);
}
// void _handleAttachmentPressed() async {
// //
// }
Future<void> _handlePageForward(
ChatComponentCubit chatComponentCubit,
WindowState<types.Message> messageWindow,
ScrollNotification notification) async {
print(
'_handlePageForward: messagesState.length=${messageWindow.length} messagesState.windowTail=${messageWindow.windowTail} messagesState.windowCount=${messageWindow.windowCount} ScrollNotification=$notification');
// Go forward a page
final tail = min(messageWindow.length,
messageWindow.windowTail + (messageWindow.windowCount ~/ 4)) %
messageWindow.length;
// Set follow
final follow = messageWindow.length == 0 ||
tail == 0; // xxx incorporate scroll position
// final scrollOffset = (notification.metrics.maxScrollExtent -
// notification.metrics.minScrollExtent) *
// (1.0 - onEndReachedThreshold);
// chatComponentCubit.scrollOffset = scrollOffset;
await chatComponentCubit.setWindow(
tail: tail, count: messageWindow.windowCount, follow: follow);
// chatComponentCubit.state.scrollController.position.jumpTo(
// chatComponentCubit.state.scrollController.offset + scrollOffset);
//chatComponentCubit.scrollOffset = 0;
}
Future<void> _handlePageBackward(
ChatComponentCubit chatComponentCubit,
WindowState<types.Message> messageWindow,
ScrollNotification notification,
) async {
print(
'_handlePageBackward: messagesState.length=${messageWindow.length} messagesState.windowTail=${messageWindow.windowTail} messagesState.windowCount=${messageWindow.windowCount} ScrollNotification=$notification');
// Go back a page
final tail = max(
messageWindow.windowCount,
(messageWindow.windowTail - (messageWindow.windowCount ~/ 4)) %
messageWindow.length);
// Set follow
final follow = messageWindow.length == 0 ||
tail == 0; // xxx incorporate scroll position
// final scrollOffset = -(notification.metrics.maxScrollExtent -
// notification.metrics.minScrollExtent) *
// (1.0 - onEndReachedThreshold);
// chatComponentCubit.scrollOffset = scrollOffset;
await chatComponentCubit.setWindow(
tail: tail, count: messageWindow.windowCount, follow: follow);
// chatComponentCubit.scrollOffset = scrollOffset;
// chatComponentCubit.state.scrollController.position.jumpTo(
// chatComponentCubit.state.scrollController.offset + scrollOffset);
//chatComponentCubit.scrollOffset = 0;
}
////////////////////////////////////////////////////////////////////////////
final TypedKey _localConversationRecordKey;
}

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

@ -13,7 +13,7 @@ import 'empty_contact_list_widget.dart';
class ContactListWidget extends StatefulWidget {
const ContactListWidget(
{required this.contactList, required this.disabled, super.key});
final IList<proto.Contact> contactList;
final IList<proto.Contact>? contactList;
final bool disabled;
@override
@ -46,13 +46,18 @@ class _ContactListWidgetState extends State<ContactListWidget>
title: translate('contacts_page.contacts'),
sliver: SliverFillRemaining(
child: SearchableList<proto.Contact>.sliver(
initialList: widget.contactList.toList(),
initialList: widget.contactList == null
? []
: widget.contactList!.toList(),
itemBuilder: (c) =>
ContactItemWidget(contact: c, disabled: widget.disabled)
.paddingLTRB(0, 4, 0, 0),
filter: (value) {
final lowerValue = value.toLowerCase();
return widget.contactList
if (widget.contactList == null) {
return [];
}
return widget.contactList!
.where((element) =>
element.nickname.toLowerCase().contains(lowerValue) ||
element.profile.name
@ -65,9 +70,13 @@ class _ContactListWidgetState extends State<ContactListWidget>
},
searchFieldHeight: 40,
spaceBetweenSearchAndList: 4,
emptyWidget: const EmptyContactListWidget(),
emptyWidget: widget.contactList == null
? waitingPage(
text: translate('contacts_page.loading_contacts'))
: const EmptyContactListWidget(),
defaultSuffixIconColor: scale.primaryScale.border,
closeKeyboardWhenScrolling: true,
searchFieldEnabled: widget.contactList != null,
inputDecoration: InputDecoration(
labelText: translate('contact_list.search'),
),

4
lib/conversation/cubits/active_single_contact_chat_bloc_map_cubit.dart
View File

@ -37,8 +37,8 @@ class _SingleContactChatState extends Equatable {
];
}
// Map of localConversationRecordKey to MessagesCubit
// Wraps a MessagesCubit to stream the latest messages to the state
// Map of localConversationRecordKey to SingleContactMessagesCubit
// Wraps a SingleContactMessagesCubit to stream the latest messages to the state
// Automatically follows the state of a ActiveConversationsBlocMapCubit.
class ActiveSingleContactChatBlocMapCubit extends BlocMapCubit<TypedKey,
SingleContactMessagesState, SingleContactMessagesCubit>

7
lib/layout/home/home_account_ready.dart
View File

@ -86,7 +86,7 @@ class _HomeAccountReadyState extends State<HomeAccountReady> {
if (activeChatLocalConversationKey == null) {
return const NoConversationWidget();
}
return ChatComponentWidget.builder(
return ChatComponentWidget(
localConversationRecordKey: activeChatLocalConversationKey,
key: ValueKey(activeChatLocalConversationKey));
}
@ -104,11 +104,6 @@ class _HomeAccountReadyState extends State<HomeAccountReady> {
final activeChat = context.watch<ActiveChatCubit>().state;
final hasActiveChat = activeChat != null;
// if (hasActiveChat) {
// _chatAnimationController.forward();
// } else {
// _chatAnimationController.reset();
// }
return LayoutBuilder(builder: (context, constraints) {
const leftColumnSize = 300.0;

37
lib/layout/home/home_screen.dart
View File

@ -1,9 +1,14 @@
import 'dart:async';
import 'dart:math';
import 'package:flutter/gestures.dart';
import 'package:flutter/material.dart';
import 'package:flutter_translate/flutter_translate.dart';
import 'package:flutter_zoom_drawer/flutter_zoom_drawer.dart';
import 'package:provider/provider.dart';
import 'package:quickalert/quickalert.dart';
import 'package:transitioned_indexed_stack/transitioned_indexed_stack.dart';
import 'package:url_launcher/url_launcher_string.dart';
import 'package:veilid_support/veilid_support.dart';
import '../../account_manager/account_manager.dart';
@ -36,6 +41,8 @@ class HomeScreenState extends State<HomeScreen>
.indexWhere((x) => x.superIdentity.recordKey == activeLocalAccount);
final canClose = activeIndex != -1;
unawaited(_doBetaDialog(context));
if (!canClose) {
await _zoomDrawerController.open!();
}
@ -43,6 +50,36 @@ class HomeScreenState extends State<HomeScreen>
super.initState();
}
Future<void> _doBetaDialog(BuildContext context) async {
await QuickAlert.show(
context: context,
title: translate('splash.beta_title'),
widget: RichText(
textAlign: TextAlign.center,
text: TextSpan(
children: <TextSpan>[
TextSpan(
text: translate('splash.beta_text'),
style: const TextStyle(
color: Colors.black87,
),
),
TextSpan(
text: 'https://veilid.com/chat/beta',
style: const TextStyle(
color: Colors.blue,
decoration: TextDecoration.underline,
),
recognizer: TapGestureRecognizer()
..onTap =
() => launchUrlString('https://veilid.com/chat/beta'),
),
],
),
),
type: QuickAlertType.warning);
}
Widget _buildAccountPage(
BuildContext context,
TypedKey superIdentityRecordKey,

5
lib/layout/home/main_pager/contacts_page.dart
View File

@ -43,8 +43,7 @@ class ContactsPageState extends State<ContactsPage> {
final ciState = context.watch<ContactListCubit>().state;
final ciBusy = ciState.busy;
final contactList =
ciState.state.asData?.value.map((x) => x.value).toIList() ??
const IListConst([]);
ciState.state.asData?.value.map((x) => x.value).toIList();
return CustomScrollView(slivers: [
if (contactInvitationRecordList.isNotEmpty)
@ -53,7 +52,7 @@ class ContactsPageState extends State<ContactsPage> {
sliver: ContactInvitationListWidget(
contactInvitationRecordList: contactInvitationRecordList,
disabled: cilBusy)),
ContactListWidget(contactList: contactList, disabled: ciBusy),
ContactListWidget(contactList: contactList, disabled: ciBusy)
]).paddingLTRB(8, 0, 8, 8);
}
}

4
lib/layout/splash.dart
View File

@ -1,5 +1,9 @@
import 'dart:async';
import 'package:flutter/material.dart';
import 'package:flutter_svg/flutter_svg.dart';
import 'package:flutter_translate/flutter_translate.dart';
import 'package:quickalert/quickalert.dart';
import 'package:radix_colors/radix_colors.dart';
import '../tools/tools.dart';

26
lib/theme/views/widget_helpers.dart
View File

@ -37,10 +37,12 @@ extension ModalProgressExt on Widget {
Widget buildProgressIndicator() => Builder(builder: (context) {
final theme = Theme.of(context);
final scale = theme.extension<ScaleScheme>()!;
return SpinKitFoldingCube(
color: scale.tertiaryScale.primary,
size: 80,
);
return FittedBox(
fit: BoxFit.scaleDown,
child: SpinKitFoldingCube(
color: scale.tertiaryScale.primary,
size: 80,
));
});
Widget waitingPage({String? text}) => Builder(builder: (context) {
@ -48,11 +50,17 @@ Widget waitingPage({String? text}) => Builder(builder: (context) {
final scale = theme.extension<ScaleScheme>()!;
return ColoredBox(
color: scale.tertiaryScale.appBackground,
child: Center(
child: Column(children: [
buildProgressIndicator().expanded(),
if (text != null) Text(text)
])));
child: Column(
crossAxisAlignment: CrossAxisAlignment.stretch,
mainAxisAlignment: MainAxisAlignment.center,
children: [
buildProgressIndicator(),
if (text != null)
Text(text,
textAlign: TextAlign.center,
style: theme.textTheme.bodySmall!
.copyWith(color: scale.tertiaryScale.appText))
]));
});
Widget debugPage(String text) => Builder(

78
packages/veilid_support/lib/dht_support/src/dht_log/dht_log_cubit.dart
View File

@ -37,7 +37,7 @@ typedef DHTLogState<T> = AsyncValue<DHTLogStateData<T>>;
typedef DHTLogBusyState<T> = BlocBusyState<DHTLogState<T>>;
class DHTLogCubit<T> extends Cubit<DHTLogBusyState<T>>
with BlocBusyWrapper<DHTLogState<T>> {
with BlocBusyWrapper<DHTLogState<T>>, RefreshableCubit {
DHTLogCubit({
required Future<DHTLog> Function() open,
required T Function(List<int> data) decodeElement,
@ -52,7 +52,7 @@ class DHTLogCubit<T> extends Cubit<DHTLogBusyState<T>>
_log = await open();
_wantsCloseRecord = true;
break;
} on VeilidAPIExceptionTryAgain {
} on DHTExceptionNotAvailable {
// Wait for a bit
await asyncSleep();
}
@ -91,6 +91,7 @@ class DHTLogCubit<T> extends Cubit<DHTLogBusyState<T>>
await _refreshNoWait(forceRefresh: forceRefresh);
}
@override
Future<void> refresh({bool forceRefresh = false}) async {
await _initWait();
await _refreshNoWait(forceRefresh: forceRefresh);
@ -101,68 +102,51 @@ class DHTLogCubit<T> extends Cubit<DHTLogBusyState<T>>
Future<void> _refreshInner(void Function(AsyncValue<DHTLogStateData<T>>) emit,
{bool forceRefresh = false}) async {
late final AsyncValue<IList<OnlineElementState<T>>> avElements;
late final int length;
await _log.operate((reader) async {
final window = await _log.operate((reader) async {
length = reader.length;
avElements =
await loadElementsFromReader(reader, _windowTail, _windowSize);
return loadElementsFromReader(reader, _windowTail, _windowSize);
});
final err = avElements.asError;
if (err != null) {
emit(AsyncValue.error(err.error, err.stackTrace));
if (window == null) {
setWantsRefresh();
return;
}
final loading = avElements.asLoading;
if (loading != null) {
emit(const AsyncValue.loading());
return;
}
final window = avElements.asData!.value;
emit(AsyncValue.data(DHTLogStateData(
length: length,
window: window,
windowTail: _windowTail,
windowSize: _windowSize,
follow: _follow)));
setRefreshed();
}
// Tail is one past the last element to load
Future<AsyncValue<IList<OnlineElementState<T>>>> loadElementsFromReader(
Future<IList<OnlineElementState<T>>?> loadElementsFromReader(
DHTLogReadOperations reader, int tail, int count,
{bool forceRefresh = false}) async {
try {
final length = reader.length;
if (length == 0) {
return const AsyncValue.data(IList.empty());
}
final end = ((tail - 1) % length) + 1;
final start = (count < end) ? end - count : 0;
// If this is writeable get the offline positions
Set<int>? offlinePositions;
if (_log.writer != null) {
offlinePositions = await reader.getOfflinePositions();
if (offlinePositions == null) {
return const AsyncValue.loading();
}
}
// Get the items
final allItems = (await reader.getRange(start,
length: end - start, forceRefresh: forceRefresh))
?.indexed
.map((x) => OnlineElementState(
value: _decodeElement(x.$2),
isOffline: offlinePositions?.contains(x.$1) ?? false))
.toIList();
if (allItems == null) {
return const AsyncValue.loading();
}
return AsyncValue.data(allItems);
} on Exception catch (e, st) {
return AsyncValue.error(e, st);
final length = reader.length;
if (length == 0) {
return const IList.empty();
}
final end = ((tail - 1) % length) + 1;
final start = (count < end) ? end - count : 0;
// If this is writeable get the offline positions
Set<int>? offlinePositions;
if (_log.writer != null) {
offlinePositions = await reader.getOfflinePositions();
}
// Get the items
final allItems = (await reader.getRange(start,
length: end - start, forceRefresh: forceRefresh))
?.indexed
.map((x) => OnlineElementState(
value: _decodeElement(x.$2),
isOffline: offlinePositions?.contains(x.$1) ?? false))
.toIList();
return allItems;
}
void _update(DHTLogUpdate upd) {

6
packages/veilid_support/lib/dht_support/src/dht_log/dht_log_read.dart
View File

@ -47,11 +47,13 @@ class _DHTLogRead implements DHTLogReadOperations {
final chunks = Iterable<int>.generate(length)
.slices(kMaxDHTConcurrency)
.map((chunk) =>
chunk.map((pos) => get(pos + start, forceRefresh: forceRefresh)));
.map((chunk) => chunk
.map((pos) async => get(pos + start, forceRefresh: forceRefresh)));
for (final chunk in chunks) {
final elems = await chunk.wait;
// If any element was unavailable, return null
if (elems.contains(null)) {
return null;
}

143
packages/veilid_support/lib/dht_support/src/dht_log/dht_log_spine.dart
View File

@ -296,7 +296,7 @@ class _DHTLogSpine {
segmentKeyBytes);
}
Future<DHTShortArray> _openOrCreateSegment(int segmentNumber) async {
Future<DHTShortArray?> _openOrCreateSegment(int segmentNumber) async {
assert(_spineMutex.isLocked, 'should be in mutex here');
assert(_spineRecord.writer != null, 'should be writable');
@ -306,51 +306,56 @@ class _DHTLogSpine {
final subkey = l.subkey;
final segment = l.segment;
var subkeyData = await _spineRecord.get(subkey: subkey);
subkeyData ??= _makeEmptySubkey();
while (true) {
final segmentKey = _getSegmentKey(subkeyData!, segment);
if (segmentKey == null) {
// Create a shortarray segment
final segmentRec = await DHTShortArray.create(
debugName: '${_spineRecord.debugName}_spine_${subkey}_$segment',
stride: _segmentStride,
crypto: _spineRecord.crypto,
parent: _spineRecord.key,
routingContext: _spineRecord.routingContext,
writer: _spineRecord.writer,
);
var success = false;
try {
// Write it back to the spine record
_setSegmentKey(subkeyData, segment, segmentRec.recordKey);
subkeyData =
await _spineRecord.tryWriteBytes(subkeyData, subkey: subkey);
// If the write was successful then we're done
if (subkeyData == null) {
// Return it
success = true;
return segmentRec;
}
} finally {
if (!success) {
await segmentRec.close();
await segmentRec.delete();
try {
var subkeyData = await _spineRecord.get(subkey: subkey);
subkeyData ??= _makeEmptySubkey();
while (true) {
final segmentKey = _getSegmentKey(subkeyData!, segment);
if (segmentKey == null) {
// Create a shortarray segment
final segmentRec = await DHTShortArray.create(
debugName: '${_spineRecord.debugName}_spine_${subkey}_$segment',
stride: _segmentStride,
crypto: _spineRecord.crypto,
parent: _spineRecord.key,
routingContext: _spineRecord.routingContext,
writer: _spineRecord.writer,
);
var success = false;
try {
// Write it back to the spine record
_setSegmentKey(subkeyData, segment, segmentRec.recordKey);
subkeyData =
await _spineRecord.tryWriteBytes(subkeyData, subkey: subkey);
// If the write was successful then we're done
if (subkeyData == null) {
// Return it
success = true;
return segmentRec;
}
} finally {
if (!success) {
await segmentRec.close();
await segmentRec.delete();
}
}
} else {
// Open a shortarray segment
final segmentRec = await DHTShortArray.openWrite(
segmentKey,
_spineRecord.writer!,
debugName: '${_spineRecord.debugName}_spine_${subkey}_$segment',
crypto: _spineRecord.crypto,
parent: _spineRecord.key,
routingContext: _spineRecord.routingContext,
);
return segmentRec;
}
} else {
// Open a shortarray segment
final segmentRec = await DHTShortArray.openWrite(
segmentKey,
_spineRecord.writer!,
debugName: '${_spineRecord.debugName}_spine_${subkey}_$segment',
crypto: _spineRecord.crypto,
parent: _spineRecord.key,
routingContext: _spineRecord.routingContext,
);
return segmentRec;
// Loop if we need to try again with the new data from the network
}
// Loop if we need to try again with the new data from the network
} on DHTExceptionNotAvailable {
return null;
}
}
@ -364,34 +369,38 @@ class _DHTLogSpine {
final segment = l.segment;
// See if we have the segment key locally
TypedKey? segmentKey;
var subkeyData = await _spineRecord.get(
subkey: subkey, refreshMode: DHTRecordRefreshMode.local);
if (subkeyData != null) {
segmentKey = _getSegmentKey(subkeyData, segment);
}
if (segmentKey == null) {
// If not, try from the network
subkeyData = await _spineRecord.get(
subkey: subkey, refreshMode: DHTRecordRefreshMode.network);
if (subkeyData == null) {
return null;
try {
TypedKey? segmentKey;
var subkeyData = await _spineRecord.get(
subkey: subkey, refreshMode: DHTRecordRefreshMode.local);
if (subkeyData != null) {
segmentKey = _getSegmentKey(subkeyData, segment);
}
segmentKey = _getSegmentKey(subkeyData, segment);
if (segmentKey == null) {
return null;
// If not, try from the network
subkeyData = await _spineRecord.get(
subkey: subkey, refreshMode: DHTRecordRefreshMode.network);
if (subkeyData == null) {
return null;
}
segmentKey = _getSegmentKey(subkeyData, segment);
if (segmentKey == null) {
return null;
}
}
}
// Open a shortarray segment
final segmentRec = await DHTShortArray.openRead(
segmentKey,
debugName: '${_spineRecord.debugName}_spine_${subkey}_$segment',
crypto: _spineRecord.crypto,
parent: _spineRecord.key,
routingContext: _spineRecord.routingContext,
);
return segmentRec;
// Open a shortarray segment
final segmentRec = await DHTShortArray.openRead(
segmentKey,
debugName: '${_spineRecord.debugName}_spine_${subkey}_$segment',
crypto: _spineRecord.crypto,
parent: _spineRecord.key,
routingContext: _spineRecord.routingContext,
);
return segmentRec;
} on DHTExceptionNotAvailable {
return null;
}
}
_DHTLogSegmentLookup _lookupSegment(int segmentNumber) {

22
packages/veilid_support/lib/dht_support/src/dht_log/dht_log_write.dart
View File

@ -17,7 +17,7 @@ class _DHTLogWrite extends _DHTLogRead implements DHTLogWriteOperations {
}
final lookup = await _spine.lookupPosition(pos);
if (lookup == null) {
throw DHTExceptionInvalidData();
throw const DHTExceptionInvalidData();
}
// Write item to the segment
@ -26,7 +26,7 @@ class _DHTLogWrite extends _DHTLogRead implements DHTLogWriteOperations {
final success =
await write.tryWriteItem(lookup.pos, newValue, output: output);
if (!success) {
throw DHTExceptionOutdated();
throw const DHTExceptionOutdated();
}
}));
} on DHTExceptionOutdated {
@ -45,12 +45,12 @@ class _DHTLogWrite extends _DHTLogRead implements DHTLogWriteOperations {
}
final aLookup = await _spine.lookupPosition(aPos);
if (aLookup == null) {
throw DHTExceptionInvalidData();
throw const DHTExceptionInvalidData();
}
final bLookup = await _spine.lookupPosition(bPos);
if (bLookup == null) {
await aLookup.close();
throw DHTExceptionInvalidData();
throw const DHTExceptionInvalidData();
}
// Swap items in the segments
@ -65,20 +65,20 @@ class _DHTLogWrite extends _DHTLogRead implements DHTLogWriteOperations {
if (bItem.value == null) {
final aItem = await aWrite.get(aLookup.pos);
if (aItem == null) {
throw DHTExceptionInvalidData();
throw const DHTExceptionInvalidData();
}
await sb.operateWriteEventual((bWrite) async {
final success = await bWrite
.tryWriteItem(bLookup.pos, aItem, output: bItem);
if (!success) {
throw DHTExceptionOutdated();
throw const DHTExceptionOutdated();
}
});
}
final success =
await aWrite.tryWriteItem(aLookup.pos, bItem.value!);
if (!success) {
throw DHTExceptionOutdated();
throw const DHTExceptionOutdated();
}
})));
}
@ -101,7 +101,7 @@ class _DHTLogWrite extends _DHTLogRead implements DHTLogWriteOperations {
await write.clear();
} else if (lookup.pos != write.length) {
// We should always be appending at the length
throw DHTExceptionInvalidData();
throw const DHTExceptionInvalidData();
}
return write.add(value);
}));
@ -122,7 +122,7 @@ class _DHTLogWrite extends _DHTLogRead implements DHTLogWriteOperations {
final lookup = await _spine.lookupPosition(insertPos + valueIdx);
if (lookup == null) {
throw DHTExceptionInvalidData();
throw const DHTExceptionInvalidData();
}
final sacount = min(remaining, DHTShortArray.maxElements - lookup.pos);
@ -137,7 +137,7 @@ class _DHTLogWrite extends _DHTLogRead implements DHTLogWriteOperations {
await write.clear();
} else if (lookup.pos != write.length) {
// We should always be appending at the length
throw DHTExceptionInvalidData();
throw const DHTExceptionInvalidData();
}
return write.addAll(sublistValues);
}));
@ -152,7 +152,7 @@ class _DHTLogWrite extends _DHTLogRead implements DHTLogWriteOperations {
await dws();
if (!success) {
throw DHTExceptionOutdated();
throw const DHTExceptionOutdated();
}
}

33
packages/veilid_support/lib/dht_support/src/dht_record/dht_record.dart
View File

@ -134,11 +134,25 @@ class DHTRecord implements DHTDeleteable<DHTRecord> {
return null;
}
final valueData = await _routingContext.getDHTValue(key, subkey,
forceRefresh: refreshMode._forceRefresh);
var retry = kDHTTryAgainTries;
ValueData? valueData;
while (true) {
try {
valueData = await _routingContext.getDHTValue(key, subkey,
forceRefresh: refreshMode._forceRefresh);
break;
} on VeilidAPIExceptionTryAgain {
retry--;
if (retry == 0) {
throw const DHTExceptionNotAvailable();
}
await asyncSleep();
}
}
if (valueData == null) {
return null;
}
// See if this get resulted in a newer sequence number
if (refreshMode == DHTRecordRefreshMode.update &&
lastSeq != null &&
@ -415,10 +429,10 @@ class DHTRecord implements DHTDeleteable<DHTRecord> {
Timestamp? expiration,
int? count}) async {
// Set up watch requirements which will get picked up by the next tick
final oldWatchState = watchState;
watchState =
final oldWatchState = _watchState;
_watchState =
_WatchState(subkeys: subkeys, expiration: expiration, count: count);
if (oldWatchState != watchState) {
if (oldWatchState != _watchState) {
_sharedDHTRecordData.needsWatchStateUpdate = true;
}
}
@ -476,8 +490,8 @@ class DHTRecord implements DHTDeleteable<DHTRecord> {
/// Takes effect on the next DHTRecordPool tick
Future<void> cancelWatch() async {
// Tear down watch requirements
if (watchState != null) {
watchState = null;
if (_watchState != null) {
_watchState = null;
_sharedDHTRecordData.needsWatchStateUpdate = true;
}
}
@ -503,7 +517,7 @@ class DHTRecord implements DHTDeleteable<DHTRecord> {
{required bool local,
required Uint8List? data,
required List<ValueSubkeyRange> subkeys}) {
final ws = watchState;
final ws = _watchState;
if (ws != null) {
final watchedSubkeys = ws.subkeys;
if (watchedSubkeys == null) {
@ -551,6 +565,5 @@ class DHTRecord implements DHTDeleteable<DHTRecord> {
final _mutex = Mutex();
int _openCount;
StreamController<DHTRecordWatchChange>? _watchController;
@internal
_WatchState? watchState;
_WatchState? _watchState;
}

3
packages/veilid_support/lib/dht_support/src/dht_record/dht_record_cubit.dart
View File

@ -29,8 +29,7 @@ abstract class DHTRecordCubit<T> extends Cubit<AsyncValue<T>> {
_record = await open();
_wantsCloseRecord = true;
break;
} on VeilidAPIExceptionKeyNotFound {
} on VeilidAPIExceptionTryAgain {
} on DHTExceptionNotAvailable {
// Wait for a bit
await asyncSleep();
}

21
packages/veilid_support/lib/dht_support/src/dht_record/dht_record_pool.dart
View File

@ -21,8 +21,11 @@ part 'dht_record_pool_private.dart';
/// Maximum number of concurrent DHT operations to perform on the network
const int kMaxDHTConcurrency = 8;
/// Number of times to retry a 'key not found'
const int kDHTKeyNotFoundRetry = 3;
/// Total number of times to try in a 'VeilidAPIExceptionKeyNotFound' loop
const int kDHTKeyNotFoundTries = 3;
/// Total number of times to try in a 'VeilidAPIExceptionTryAgain' loop
const int kDHTTryAgainTries = 3;
typedef DHTRecordPoolLogger = void Function(String message);
@ -280,12 +283,12 @@ class DHTRecordPool with TableDBBackedJson<DHTRecordPoolAllocations> {
for (final rec in openedRecordInfo.records) {
// See if the watch had an expiration and if it has expired
// otherwise the renewal will keep the same parameters
final watchState = rec.watchState;
final watchState = rec._watchState;
if (watchState != null) {
final exp = watchState.expiration;
if (exp != null && exp.value < now) {
// Has expiration, and it has expired, clear watch state
rec.watchState = null;
rec._watchState = null;
}
}
}
@ -392,7 +395,7 @@ class DHTRecordPool with TableDBBackedJson<DHTRecordPoolAllocations> {
if (openedRecordInfo == null) {
// Fresh open, just open the record
var retry = kDHTKeyNotFoundRetry;
var retry = kDHTKeyNotFoundTries;
late final DHTRecordDescriptor recordDescriptor;
while (true) {
try {
@ -403,7 +406,7 @@ class DHTRecordPool with TableDBBackedJson<DHTRecordPoolAllocations> {
await asyncSleep();
retry--;
if (retry == 0) {
rethrow;
throw DHTExceptionNotAvailable();
}
}
}
@ -705,7 +708,7 @@ class DHTRecordPool with TableDBBackedJson<DHTRecordPoolAllocations> {
var cancelWatch = true;
for (final rec in records) {
final ws = rec.watchState;
final ws = rec._watchState;
if (ws != null) {
cancelWatch = false;
final wsCount = ws.count;
@ -762,9 +765,9 @@ class DHTRecordPool with TableDBBackedJson<DHTRecordPoolAllocations> {
static void _updateWatchRealExpirations(Iterable<DHTRecord> records,
Timestamp realExpiration, Timestamp renewalTime) {
for (final rec in records) {
final ws = rec.watchState;
final ws = rec._watchState;
if (ws != null) {
rec.watchState = _WatchState(
rec._watchState = _WatchState(
subkeys: ws.subkeys,
expiration: ws.expiration,
count: ws.count,

6
packages/veilid_support/lib/dht_support/src/dht_short_array/dht_short_array.dart
View File

@ -68,7 +68,7 @@ class DHTShortArray implements DHTDeleteable<DHTShortArray> {
}
});
return dhtShortArray;
} on Exception catch (_) {
} on Exception {
await dhtRecord.close();
await pool.deleteRecord(dhtRecord.key);
rethrow;
@ -89,7 +89,7 @@ class DHTShortArray implements DHTDeleteable<DHTShortArray> {
final dhtShortArray = DHTShortArray._(headRecord: dhtRecord);
await dhtShortArray._head.operate((head) => head._loadHead());
return dhtShortArray;
} on Exception catch (_) {
} on Exception {
await dhtRecord.close();
rethrow;
}
@ -113,7 +113,7 @@ class DHTShortArray implements DHTDeleteable<DHTShortArray> {
final dhtShortArray = DHTShortArray._(headRecord: dhtRecord);
await dhtShortArray._head.operate((head) => head._loadHead());
return dhtShortArray;
} on Exception catch (_) {
} on Exception {
await dhtRecord.close();
rethrow;
}

14
packages/veilid_support/lib/dht_support/src/dht_short_array/dht_short_array_cubit.dart
View File

@ -8,6 +8,7 @@ import 'package:fast_immutable_collections/fast_immutable_collections.dart';
import 'package:meta/meta.dart';
import '../../../veilid_support.dart';
import '../interfaces/refreshable_cubit.dart';
@immutable
class DHTShortArrayElementState<T> extends Equatable {
@ -24,7 +25,7 @@ typedef DHTShortArrayState<T> = AsyncValue<IList<DHTShortArrayElementState<T>>>;
typedef DHTShortArrayBusyState<T> = BlocBusyState<DHTShortArrayState<T>>;
class DHTShortArrayCubit<T> extends Cubit<DHTShortArrayBusyState<T>>
with BlocBusyWrapper<DHTShortArrayState<T>> {
with BlocBusyWrapper<DHTShortArrayState<T>>, RefreshableCubit {
DHTShortArrayCubit({
required Future<DHTShortArray> Function() open,
required T Function(List<int> data) decodeElement,
@ -39,7 +40,7 @@ class DHTShortArrayCubit<T> extends Cubit<DHTShortArrayBusyState<T>>
_shortArray = await open();
_wantsCloseRecord = true;
break;
} on VeilidAPIExceptionTryAgain {
} on DHTExceptionNotAvailable {
// Wait for a bit
await asyncSleep();
}
@ -57,6 +58,7 @@ class DHTShortArrayCubit<T> extends Cubit<DHTShortArrayBusyState<T>>
});
}
@override
Future<void> refresh({bool forceRefresh = false}) async {
await _initWait();
await _refreshNoWait(forceRefresh: forceRefresh);
@ -87,9 +89,13 @@ class DHTShortArrayCubit<T> extends Cubit<DHTShortArrayBusyState<T>>
.toIList();
return allItems;
});
if (newState != null) {
emit(AsyncValue.data(newState));
if (newState == null) {
// Mark us as needing refresh
setWantsRefresh();
return;
}
emit(AsyncValue.data(newState));
setRefreshed();
} on Exception catch (e) {
emit(AsyncValue.error(e));
}

2
packages/veilid_support/lib/dht_support/src/dht_short_array/dht_short_array_head.dart
View File

@ -91,7 +91,7 @@ class _DHTShortArrayHead {
if (!await _writeHead()) {
// Failed to write head means head got overwritten so write should
// be considered failed
throw DHTExceptionOutdated();
throw const DHTExceptionOutdated();
}
onUpdatedHead?.call();

36
packages/veilid_support/lib/dht_support/src/dht_short_array/dht_short_array_read.dart
View File

@ -17,21 +17,25 @@ class _DHTShortArrayRead implements DHTShortArrayReadOperations {
throw IndexError.withLength(pos, length);
}
final lookup = await _head.lookupPosition(pos, false);
try {
final lookup = await _head.lookupPosition(pos, false);
final refresh = forceRefresh || _head.positionNeedsRefresh(pos);
final outSeqNum = Output<int>();
final out = lookup.record.get(
subkey: lookup.recordSubkey,
refreshMode: refresh
? DHTRecordRefreshMode.network
: DHTRecordRefreshMode.cached,
outSeqNum: outSeqNum);
if (outSeqNum.value != null) {
_head.updatePositionSeq(pos, false, outSeqNum.value!);
final refresh = forceRefresh || _head.positionNeedsRefresh(pos);
final outSeqNum = Output<int>();
final out = await lookup.record.get(
subkey: lookup.recordSubkey,
refreshMode: refresh
? DHTRecordRefreshMode.network
: DHTRecordRefreshMode.cached,
outSeqNum: outSeqNum);
if (outSeqNum.value != null) {
_head.updatePositionSeq(pos, false, outSeqNum.value!);
}
return out;
} on DHTExceptionNotAvailable {
// If any element is not available, return null
return null;
}
return out;
}
(int, int) _clampStartLen(int start, int? len) {
@ -56,11 +60,13 @@ class _DHTShortArrayRead implements DHTShortArrayReadOperations {
final chunks = Iterable<int>.generate(length)
.slices(kMaxDHTConcurrency)
.map((chunk) =>
chunk.map((pos) => get(pos + start, forceRefresh: forceRefresh)));
.map((chunk) => chunk
.map((pos) async => get(pos + start, forceRefresh: forceRefresh)));
for (final chunk in chunks) {
final elems = await chunk.wait;
// If any element was unavailable, return null
if (elems.contains(null)) {
return null;
}

18
packages/veilid_support/lib/dht_support/src/interfaces/exceptions.dart
View File

@ -1,15 +1,21 @@
class DHTExceptionOutdated implements Exception {
DHTExceptionOutdated(
const DHTExceptionOutdated(
[this.cause = 'operation failed due to newer dht value']);
String cause;
final String cause;
}
class DHTExceptionInvalidData implements Exception {
DHTExceptionInvalidData([this.cause = 'dht data structure is corrupt']);
String cause;
const DHTExceptionInvalidData([this.cause = 'dht data structure is corrupt']);
final String cause;
}
class DHTExceptionCancelled implements Exception {
DHTExceptionCancelled([this.cause = 'operation was cancelled']);
String cause;
const DHTExceptionCancelled([this.cause = 'operation was cancelled']);
final String cause;
}
class DHTExceptionNotAvailable implements Exception {
const DHTExceptionNotAvailable(
[this.cause = 'request could not be completed at this time']);
final String cause;
}

1
packages/veilid_support/lib/dht_support/src/interfaces/interfaces.dart
View File

@ -6,3 +6,4 @@ export 'dht_random_read.dart';
export 'dht_random_write.dart';
export 'dht_truncate.dart';
export 'exceptions.dart';
export 'refreshable_cubit.dart';

16
packages/veilid_support/lib/dht_support/src/interfaces/refreshable_cubit.dart Normal file
View File

@ -0,0 +1,16 @@
abstract mixin class RefreshableCubit {
Future<void> refresh({bool forceRefresh = false});
void setWantsRefresh() {
_wantsRefresh = true;
}
void setRefreshed() {
_wantsRefresh = false;
}
bool get wantsRefresh => _wantsRefresh;
////////////////////////////////////////////////////////////////////////////
bool _wantsRefresh = false;
}

2
pubspec.lock
View File

@ -1576,7 +1576,7 @@ packages:
source: hosted
version: "1.1.0"
url_launcher:
dependency: transitive
dependency: "direct main"
description:
name: url_launcher
sha256: "21b704ce5fa560ea9f3b525b43601c678728ba46725bab9b01187b4831377ed3"

1
pubspec.yaml
View File

@ -96,6 +96,7 @@ dependencies:
stack_trace: ^1.11.1
stream_transform: ^2.1.0
transitioned_indexed_stack: ^1.0.2
url_launcher: ^6.3.0
uuid: ^4.4.0
veilid:
# veilid: ^0.0.1