+QueryObserver

+Query
+notifyManager
+onlineManager
+QueryCache
This commit is contained in:
Kingkor Roy Tirtho
2022-02-10 18:29:42 +06:00
parent f1415f55f4
commit 3ec81de606
51 changed files with 2503 additions and 4083 deletions
@@ -1,95 +0,0 @@
import 'package:fl_query/src/core/_data_class.dart';
import 'package:fl_query/fl_query.dart';
import 'package:fl_query/src/core/result_parser.dart';
/// TODO refactor into [Request] container
/// Base options.
abstract class BaseOptions<TParsed> extends MutableDataClass {
BaseOptions({
required this.document,
this.variables = const {},
this.operationName,
ResultParserFn<TParsed>? parserFn,
Context? context,
FetchPolicy? fetchPolicy,
ErrorPolicy? errorPolicy,
CacheRereadPolicy? cacheRereadPolicy,
this.optimisticResult,
}) : policies = Policies(
fetch: fetchPolicy,
error: errorPolicy,
cacheReread: cacheRereadPolicy,
),
context = context ?? Context(),
parserFn = parserFn ??
((d) => throw UnimplementedError(
"Please provide a parser function to support result parsing.",
));
/// Document containing at least one [OperationDefinitionNode]
DocumentNode document;
/// Name of the executable definition
///
/// Must be specified if [document] contains more than one [OperationDefinitionNode]
String? operationName;
/// A map going from variable name to variable value, where the variables are used
/// within the GraphQL query.
Map<String, dynamic> variables;
/// An optimistic result to eagerly add to the operation stream
Object? optimisticResult;
/// Specifies the [Policies] to be used during execution.
Policies policies;
FetchPolicy? get fetchPolicy => policies.fetch;
ErrorPolicy? get errorPolicy => policies.error;
CacheRereadPolicy? get cacheRereadPolicy => policies.cacheReread;
/// Context to be passed to link execution chain.
Context context;
ResultParserFn<TParsed> parserFn;
// TODO consider inverting this relationship
/// Resolve these options into a request
Request get asRequest => Request(
operation: Operation(
document: document,
operationName: operationName,
),
variables: variables,
context: context,
);
@override
List<Object?> get properties => [
document,
operationName,
variables,
optimisticResult,
policies,
context,
];
OperationType get type {
final definitions =
document.definitions.whereType<OperationDefinitionNode>().toList();
if (operationName != null) {
definitions.removeWhere(
(node) => node.name!.value != operationName,
);
}
// TODO differentiate error types, add exception
assert(definitions.length == 1);
return definitions.first.type;
}
bool get isQuery => type == OperationType.query;
bool get isMutation => type == OperationType.mutation;
bool get isSubscription => type == OperationType.subscription;
}
@@ -1,29 +0,0 @@
import 'package:meta/meta.dart';
import "package:collection/collection.dart";
/// Helper for making mutable data classes with
/// a [properties] based [equal] helper
///
/// NOTE: I (@micimize) settled on this helper instead of truly immutable classes
/// because I didn't want to deal with the issue of `copyWith(field: null)`,
/// but also didn't want to commit to adding a true dataclass generator
/// like `freezed` or `built_value` yet. I consider this a stopgap,
/// and think we should eventually have a truly immutable API
abstract class MutableDataClass {
const MutableDataClass();
/// identifying properties for the inheriting class
@protected
List<Object?> get properties;
/// [properties] based deep equality check
bool equal(MutableDataClass other) =>
identical(this, other) ||
(runtimeType == other.runtimeType &&
const ListEquality<Object?>(
DeepCollectionEquality(),
).equals(
other.properties,
properties,
));
}
@@ -1,103 +0,0 @@
import 'package:fl_query/fl_query.dart';
import 'package:fl_query/src/core/query_key.dart';
/// Internal writeQuery wrapper
typedef _IntWriteQuery = void Function(
QueryKey queryKey, Map<String, dynamic>? data);
extension InternalQueryWriteHandling on QueryManager {
/// Merges exceptions into `queryResult` and
/// returns `true` on success.
///
/// This is named `*OrSetExceptionOnQueryResult` because it is very imperative,
/// and edits the [queryResult] inplace.
bool _writeQueryOrSetExceptionOnQueryResult(
QueryKey queryKey,
Map<String, dynamic>? data,
QueryResult? queryResult, {
required _IntWriteQuery writeQuery,
}) {
try {
writeQuery(queryKey, data);
return true;
} on CacheMisconfigurationException catch (failure) {
queryResult!.exception = coalesceErrors(
exception: queryResult.exception,
linkException: failure,
);
}
return false;
}
/// Part of [InternalQueryWriteHandling], and not exposed outside the
/// library.
///
/// Returns `true` if a reread should be attempted to incorporate potential optimistic data.
///
/// If we have no data, we skip caching, thus taking [ErrorPolicy.none]
/// into account.
///
/// networked wrapper for [_writeQueryOrSetExceptionOnQueryResult]
/// NOTE: mapFetchResultToQueryResult must be called beforehand
bool attemptCacheWriteFromResponse(
Policies policies,
Request request,
Response response,
QueryResult? queryResult,
) =>
(policies.fetch == FetchPolicy.noCache || queryResult!.data == null)
? false
: _writeQueryOrSetExceptionOnQueryResult(
request,
response.data,
queryResult,
writeQuery: (req, data) => cache.writeQuery(req, data: data!),
onPartial: (failure) => UnexpectedResponseStructureException(
failure,
queryKey: request,
parsedResponse: response,
),
) &&
policies.mergeOptimisticData;
/// Part of [InternalQueryWriteHandling], and not exposed outside the
/// library.
///
/// client-side wrapper for [_writeQueryOrSetExceptionOnQueryResult]
bool attemptCacheWriteFromClient(
Request request,
Map<String, dynamic>? data,
QueryResult queryResult, {
required _IntWriteQuery writeQuery,
}) =>
_writeQueryOrSetExceptionOnQueryResult(
request,
data,
queryResult,
writeQuery: writeQuery,
onPartial: (failure) => MismatchedDataStructureException(
failure,
queryKey: request,
data: data,
),
);
/// Reread the request into the result from the cache,
/// adding a [CacheMissException] if it fails to do so
void attempCacheRereadIntoResult(Request request, QueryResult? queryResult) {
// normalize results if previously written
final rereadData = cache.readQuery(request);
if (rereadData == null) {
queryResult!.exception = coalesceErrors(
exception: queryResult.exception,
linkException: CacheMissException(
'Round trip cache re-read failed: cache.readQuery(request) returned null',
request,
expectedData: queryResult.data,
),
);
} else {
queryResult!.data = rereadData;
}
}
}
-6
View File
@@ -1,6 +0,0 @@
export 'package:fl_query/src/core/observable_query.dart';
export 'package:fl_query/src/core/query_manager.dart';
export 'package:fl_query/src/core/query_options.dart';
export 'package:fl_query/src/core/mutation_options.dart';
export 'package:fl_query/src/core/query_result.dart';
export 'package:fl_query/src/core/policies.dart';
@@ -1,84 +0,0 @@
import 'dart:async';
import 'package:fl_query/fl_query.dart';
import 'package:fl_query/src/core/_query_write_handling.dart';
/// Fetch more results and then merge them with [previousResult]
/// according to [FetchMoreOptions.updateQuery]
///
/// Will add results if [ObservableQuery.queryId] is supplied,
/// and broadcast any cache changes
///
/// This is the **Internal Implementation**,
/// used by [ObservableQuery] and [GraphQLCLient.fetchMore]
Future<QueryResult<TParsed>> fetchMoreImplementation<TParsed>(
FetchMoreOptions fetchMoreOptions, {
required QueryOptions<TParsed> originalOptions,
required QueryManager queryManager,
required QueryResult<TParsed> previousResult,
String? queryId,
}) async {
// fetch more and update
final document = (fetchMoreOptions.document ?? originalOptions.document);
final request = originalOptions.asRequest;
final combinedOptions = QueryOptions<TParsed>(
fetchPolicy: FetchPolicy.noCache,
errorPolicy: originalOptions.errorPolicy,
document: document,
variables: {
...originalOptions.variables,
...fetchMoreOptions.variables,
},
);
QueryResult<TParsed> fetchMoreResult =
await queryManager.query(combinedOptions);
try {
// combine the query with the new query, using the function provided by the user
final data = fetchMoreOptions.updateQuery(
previousResult.data,
fetchMoreResult.data,
)!;
fetchMoreResult.data = data;
if (originalOptions.fetchPolicy != FetchPolicy.noCache) {
queryManager.attemptCacheWriteFromClient(
request,
data,
fetchMoreResult,
writeQuery: (req, data) => queryManager.cache.writeQuery(
req,
data: data!,
),
);
}
// will add to a stream with `queryId` and rebroadcast if appropriate
queryManager.addQueryResult(
request,
queryId,
fetchMoreResult,
);
} catch (error) {
if (fetchMoreResult.hasException) {
// because the updateQuery failure might have been because of these errors,
// we just add them to the old errors
previousResult.exception = coalesceErrors(
exception: previousResult.exception,
graphqlErrors: fetchMoreResult.exception!.graphqlErrors,
linkException: fetchMoreResult.exception!.linkException,
);
return previousResult;
} else {
// TODO merge results OperationException
rethrow;
}
}
return fetchMoreResult;
}
+419
View File
@@ -0,0 +1,419 @@
import 'package:fl_query/src/core/query.dart';
import 'package:fl_query/src/core/query_key.dart';
import 'package:fl_query/src/core/retryer.dart';
typedef QueryMeta<T> = Map<String, T>;
typedef QueryKeyHashFunction = String Function(QueryKey queryKey);
typedef QueryFunction<T, TPageParam> = Function(QueryFunctionContext context);
typedef GetPreviousPageParamFunction<TQueryFnData> = Function(
TQueryFnData firstPage,
List<TQueryFnData> allPages,
);
typedef GetNextPageParamFunction<TQueryFnData> = Function(
TQueryFnData firstPage,
List<TQueryFnData> allPages,
);
class QueryOptions<TQueryFnData, TError, TData> {
ShouldRetryFunction<TError>? retry;
RetryDelayFunction<TError>? retryDelay;
Duration? cacheTime;
bool Function(TData? oldData, TData newData)? isDataEqual;
QueryFunction? queryFn;
QueryKey? queryKey;
/// Basically [QueryKey.key] in short form
String? queryHash;
QueryKeyHashFunction? queryKeyHashFn;
TData? initialData;
DateTime? initialDataUpdatedAt;
QueryBehavior<TQueryFnData, TError, TData>? behavior;
/// Set this to `false` to disable structural sharing between query results\
/// Defaults to `true`.
bool? structuralSharing;
/// This function can be set to automatically get the previous cursor for infinite queries.
/// The result will also be used to determine the value of `hasPreviousPage`.
GetPreviousPageParamFunction<TQueryFnData>? getPreviousPageParam;
/// This function can be set to automatically get the next cursor for
/// infinite queries.
/// The result will also be used to determine the value of
/// `hasNextPage`.
GetNextPageParamFunction<TQueryFnData>? getNextPageParam;
bool? defaulted;
/// Additional payload to be stored on each query.
/// Use this property to pass information that can be used in other places.
QueryMeta? meta;
QueryOptions({
this.retry,
this.retryDelay,
this.queryKey,
this.queryKeyHashFn,
this.cacheTime,
this.isDataEqual,
this.queryFn,
this.defaulted,
this.initialData,
this.initialDataUpdatedAt,
this.meta,
this.queryHash,
this.structuralSharing,
this.getPreviousPageParam,
this.getNextPageParam,
});
QueryOptions.fromJson(Map<String, dynamic> json) {
queryKey = json["queryKey"];
queryKeyHashFn = json["queryKeyHashFn"];
cacheTime = json["cacheTime"];
isDataEqual = json["isDataEqual"];
queryFn = json["queryFn"];
queryHash = json["queryHash"];
initialData = json["initialData"];
initialDataUpdatedAt = json["initialDataUpdatedAt"];
meta = json["meta"];
structuralSharing = json["structuralSharing"];
defaulted = json["defaulted"];
retry = json["retry"];
retryDelay = json["retryDelay"];
behavior = json["behavior"];
getPreviousPageParam = json["getPreviousPageParam"];
getNextPageParam = json["getNextPageParam"];
}
Map<String, dynamic> toJson() {
return {
"queryKey": queryKey,
"queryKeyHashFn": queryKeyHashFn,
"cacheTime": cacheTime,
"isDataEqual": isDataEqual,
"queryFn": queryFn,
"queryHash": queryHash,
"initialData": initialData,
"initialDataUpdatedAt": initialDataUpdatedAt,
"meta": meta,
"structuralSharing": structuralSharing,
"defaulted": defaulted,
"retry": retry,
"retryDelay": retryDelay,
"behavior": behavior,
"getPreviousPageParam": getPreviousPageParam,
"getNextPageParam": getNextPageParam,
};
}
}
class QueryFilters {
bool? active;
bool? exact;
bool? inactive;
bool Function(Query query)? predicate;
bool? queryKey;
bool? stale;
bool? fetching;
QueryFilters({
this.active,
this.exact,
this.inactive,
this.predicate,
this.queryKey,
this.stale,
this.fetching,
});
Map<String, dynamic> toJson() {
return {
"active": active,
"exact": exact,
"inactive": inactive,
"queryKey": queryKey,
"stale": stale,
"fetching": fetching,
"predicate": predicate,
};
}
}
class RefetchPageFilters<TPageData> {
bool Function(TPageData lastPage, int index, List<TPageData> allPages)?
refetchPage;
}
class RefetchQueryFilters<TPageData>
implements QueryFilters, RefetchPageFilters<TPageData> {
@override
bool? active;
@override
bool? exact;
@override
bool? fetching;
@override
bool? inactive;
@override
bool Function(Query query)? predicate;
@override
bool? queryKey;
@override
bool Function(TPageData lastPage, int index, List<TPageData> allPages)?
refetchPage;
@override
bool? stale;
RefetchQueryFilters({
this.active,
this.exact,
this.inactive,
this.predicate,
this.queryKey,
this.stale,
this.fetching,
this.refetchPage,
});
@override
Map<String, dynamic> toJson() {
return {
"active": active,
"exact": exact,
"inactive": inactive,
"queryKey": queryKey,
"stale": stale,
"fetching": fetching,
"predicate": predicate,
};
}
}
class ResultOptions {
bool? throwOnError;
}
class RefetchOptions implements ResultOptions {
bool? cancelRefetch;
@override
bool? throwOnError;
}
enum QueryStatus {
idle,
loading,
error,
success,
}
class QueryObserverResult<TData, TError> {
TData? data;
DateTime? dataUpdatedAt;
TError? error;
DateTime? errorUpdatedAt;
int failureCount;
bool isError;
bool isFetched;
bool isFetchedAfterMount;
bool isFetching;
bool isIdle;
bool isLoading;
bool isLoadingError;
bool isPlaceholderData;
bool isPreviousData;
bool isRefetchError;
bool isRefetching;
bool isStale;
bool isSuccess;
Future<QueryObserverResult<TData, TError>> Function<TPageData>({
RefetchOptions options,
RefetchQueryFilters<TPageData> filters,
}) refetch;
void Function() remove;
QueryStatus status;
QueryObserverResult({
required this.failureCount,
required this.isError,
required this.isFetched,
required this.isFetchedAfterMount,
required this.isFetching,
required this.isIdle,
required this.isLoading,
required this.isLoadingError,
required this.isPlaceholderData,
required this.isPreviousData,
required this.isRefetchError,
required this.isRefetching,
required this.isStale,
required this.isSuccess,
required this.refetch,
required this.remove,
required this.status,
this.data,
this.error,
this.dataUpdatedAt,
this.errorUpdatedAt,
}) {
String errorLabel =
"[QueryObserverResult.QueryObserverResult] status = `$status` but parent has wrong set of properties";
if (status == QueryStatus.idle &&
(data != null ||
error != null ||
isError ||
!isIdle ||
isLoading ||
isLoadingError ||
isRefetchError ||
isSuccess)) throw Exception(errorLabel);
if (status == QueryStatus.loading &&
(data != null ||
error != null ||
isError ||
isIdle ||
!isLoading ||
isLoadingError ||
isRefetchError ||
isSuccess != false)) throw Exception(errorLabel);
if (status == QueryStatus.error &&
((!(error is TError)) || !isError || isIdle || isLoading || isSuccess))
throw Exception(errorLabel);
if (status == QueryStatus.success &&
(!(data is TData) ||
error != null ||
isError ||
isIdle ||
isLoading ||
isLoadingError ||
isRefetchError ||
!isSuccess)) throw Exception(errorLabel);
}
Map<String, dynamic> toJson() {
final Map<String, dynamic> data = {
'data': this.data,
'dataUpdatedAt': dataUpdatedAt,
'error': error,
'errorUpdatedAt': errorUpdatedAt,
'failureCount': failureCount,
'isError': isError,
'isFetched': isFetched,
'isFetchedAfterMount': isFetchedAfterMount,
'isFetching': isFetching,
'isIdle': isIdle,
'isLoading': isLoading,
'isLoadingError': isLoadingError,
'isPlaceholderData': isPlaceholderData,
'isPreviousData': isPreviousData,
'isRefetchError': isRefetchError,
'isRefetching': isRefetching,
'isStale': isStale,
'isSuccess': isSuccess,
'refetch': refetch,
'remove': remove,
'status': status,
};
return data;
}
}
typedef RefetchIntervalFunction<TQueryFnData, TError, TQueryData, TData>
= Duration? Function(
TData? data,
Query<TQueryFnData, TError, TQueryData> query,
);
enum RefetchOnReconnect {
on,
off,
always,
}
enum RefetchOnMount {
on,
off,
always,
}
class QueryObserverOptions<TQueryFnData, TError, TData, TQueryData>
extends QueryOptions<TQueryFnData, TError, TQueryData> {
bool? enabled;
Duration? staleTime;
RefetchIntervalFunction<TQueryFnData, TError, TQueryData, TData>?
refetchInterval;
bool? refetchIntervalInBackground;
RefetchOnReconnect? refetchOnReconnect;
RefetchOnMount? refetchOnMount;
bool? retryOnMount;
OnData? onSuccess;
OnError? onError;
void Function(TData? data, [TError? error])? onSettled;
bool Function(TError error)? useErrorBoundary;
TData Function(TQueryData? data)? select;
bool? suspense;
bool? keepPreviousData;
TQueryData? placeholderData;
bool? optimisticResults;
/*List<String>|'tracked'?*/ dynamic notifyOnChangeProps;
List<String>? notifyOnChangePropsExclusions;
QueryObserverOptions({
this.enabled,
this.staleTime,
this.refetchInterval,
this.refetchIntervalInBackground,
this.refetchOnReconnect,
this.refetchOnMount,
this.retryOnMount,
this.onSuccess,
this.onError,
this.onSettled,
this.useErrorBoundary,
this.select,
this.suspense,
this.keepPreviousData,
this.placeholderData,
this.optimisticResults,
QueryKey? queryKey,
QueryKeyHashFunction? queryKeyHashFn,
Duration? cacheTime,
bool Function(TQueryData? oldData, TQueryData newData)? isDataEqual,
QueryFunction? queryFn,
String? queryHash,
TQueryData? initialData,
DateTime? initialDataUpdatedAt,
QueryMeta? meta,
bool? structuralSharing,
bool? defaulted,
}) : super(
queryKey: queryKey,
queryKeyHashFn: queryKeyHashFn,
cacheTime: cacheTime,
isDataEqual: isDataEqual,
queryFn: queryFn,
queryHash: queryHash,
initialData: initialData,
initialDataUpdatedAt: initialDataUpdatedAt,
meta: meta,
structuralSharing: structuralSharing,
defaulted: defaulted,
);
}
class QueryFunctionContext<TPageParam> {
QueryKey queryKey;
/* AbortSignal */ dynamic? signal;
TPageParam? pageParam;
QueryMeta? meta;
QueryFunctionContext({
required this.queryKey,
this.signal,
this.pageParam,
this.meta,
});
}
@@ -1,141 +0,0 @@
// ignore_for_file: deprecated_member_use_from_same_package
import 'dart:async';
import 'package:fl_query/src/cache/cache.dart';
import 'package:fl_query/src/core/_base_options.dart';
import 'package:fl_query/src/core/observable_query.dart';
import 'package:fl_query/src/core/result_parser.dart';
import 'package:fl_query/src/exceptions.dart';
import 'package:fl_query/src/core/query_result.dart';
import 'package:fl_query/src/utilities/helpers.dart';
import 'package:fl_query/src/core/policies.dart';
typedef OnMutationCompleted = FutureOr<void> Function(dynamic data);
typedef OnMutationUpdate = FutureOr<void> Function(
JSONDataProxy cache,
QueryResult? result,
);
typedef OnError = FutureOr<void> Function(OperationException? error);
class MutationOptions<TParsed> extends BaseOptions<TParsed> {
MutationOptions({
required DocumentNode document,
String? operationName,
Map<String, dynamic> variables = const {},
FetchPolicy? fetchPolicy,
ErrorPolicy? errorPolicy,
CacheRereadPolicy? cacheRereadPolicy,
Context? context,
Object? optimisticResult,
this.onCompleted,
this.update,
this.onError,
ResultParserFn<TParsed>? parserFn,
}) : super(
fetchPolicy: fetchPolicy,
errorPolicy: errorPolicy,
cacheRereadPolicy: cacheRereadPolicy,
document: document,
operationName: operationName,
variables: variables,
context: context,
optimisticResult: optimisticResult,
parserFn: parserFn,
);
final OnMutationCompleted? onCompleted;
final OnMutationUpdate? update;
final OnError? onError;
@override
List<Object?> get properties =>
[...super.properties, onCompleted, update, onError];
}
/// Handles execution of mutation `update`, `onCompleted`, and `onError` callbacks
class MutationCallbackHandler {
final MutationOptions options;
final QueryCache cache;
final String queryId;
MutationCallbackHandler({
required this.options,
required this.cache,
required this.queryId,
});
// callbacks will be called against each result in the stream,
// which should then rebroadcast queries with the appropriate optimism
Iterable<OnData> get callbacks =>
<OnData?>[onCompleted, update, onError].where(notNull).cast<OnData>();
// Todo: probably move this to its own class
OnData? get onCompleted {
if (options.onCompleted != null) {
return (QueryResult? result) {
if (!result!.isLoading && !result.isOptimistic) {
return options.onCompleted!(result.data);
}
};
}
return null;
}
OnData? get onError {
if (options.onError != null) {
return (QueryResult? result) {
if (!result!.isLoading &&
result.hasException &&
options.errorPolicy != ErrorPolicy.ignore) {
return options.onError!(result.exception);
}
};
}
return null;
}
/// The optimistic cache layer id `update` will write to
/// is a "child patch" of the default optimistic patch
/// created by the query manager
String get _patchId => '${queryId}.update';
/// apply the user's patch
void _optimisticUpdate(QueryResult? result) {
final String patchId = _patchId;
// this is also done in query_manager, but better safe than sorry
cache.recordOptimisticTransaction(
(JSONDataProxy cache) {
options.update!(cache, result);
return cache;
},
patchId,
);
}
// optimistic patches will be cleaned up by the query_manager
// cleanup is handled by heirarchical optimism -
// as in, because our patch id is prefixed with '${observableQuery.queryId}.',
// it will be discarded along with the observableQuery.queryId patch
// TODO this results in an implicit coupling with the patch id system
OnData? get update {
if (options.update != null) {
// dereference all variables that might be needed if the widget is disposed
final OnMutationUpdate? widgetUpdate = options.update;
final OnData optimisticUpdate = _optimisticUpdate;
// wrap update logic to handle optimism
FutureOr<void> updateOnData(QueryResult? result) {
if (result!.isOptimistic) {
return optimisticUpdate(result);
} else {
return widgetUpdate!(cache, result);
}
}
return updateOnData;
}
return null;
}
}
@@ -0,0 +1,95 @@
// TYPES
import 'package:fl_query/src/core/utils.dart';
typedef NotifyCallback = void Function();
typedef NotifyFunction = void Function(void Function() callback);
typedef BatchNotifyFunction = void Function(void Function() callback);
class _NotifyManager {
List<NotifyCallback> _queue;
int _transactions;
late NotifyFunction _notifyFn;
late BatchNotifyFunction _batchNotifyFn;
_NotifyManager()
: _queue = [],
_transactions = 0 {
_notifyFn = (void Function() callback) {
callback();
};
_batchNotifyFn = (void Function() callback) {
callback();
};
}
T batch<T>(T Function() callback) {
T result;
_transactions++;
try {
result = callback();
} finally {
_transactions--;
if (_transactions == 0) {
flush();
}
}
return result;
}
schedule(NotifyCallback callback) {
if (_transactions > 0) {
_queue.add(callback);
} else {
scheduleMicrotask((val) {
_notifyFn(callback);
});
}
}
/// All calls to the wrapped function will be batched.
T batchCalls<T extends void Function(List? args)>(T callback) {
void fn(List? args) {
schedule(() {
callback(args);
});
}
;
return fn as T;
}
void flush() {
var queue = _queue;
_queue = [];
if (queue.isNotEmpty) {
scheduleMicrotask((val) {
_batchNotifyFn(() {
queue.forEach((fn) {
_notifyFn(fn);
});
});
});
}
}
///Use this method to set a custom notify function.
void setNotifyFunction(NotifyFunction fn) {
_notifyFn = fn;
}
/// Use this method to set a custom function to batch notifications
/// together into a single tick.
/// By default React Query will use the batch function provided by
/// ReactDOM or React Native.
void setBatchNotifyFunction(BatchNotifyFunction fn) {
_batchNotifyFn = fn;
}
}
// SINGLETON
_NotifyManager notifyManager = new _NotifyManager();
@@ -1,386 +0,0 @@
import 'dart:async';
import 'package:fl_query/fl_query.dart';
import 'package:meta/meta.dart';
import 'package:fl_query/src/core/fetch_more.dart';
import 'package:fl_query/src/scheduler/scheduler.dart';
/// Side effect to register for execution when data is received
typedef OnData = FutureOr<void> Function(QueryResult? result);
/// Lifecycle states for [ObservableQuery.lifecycle]
enum QueryLifecycle {
/// No results have been requested or fetched
unexecuted,
/// Results are being fetched, and will be side-effect free
pending,
/// Polling for results periodically
polling,
/// Was polling but [ObservableQuery.stopPolling()] was called
pollingStopped,
/// Results are being fetched, and will trigger
/// the callbacks registered with [ObservableQuery.onData]
sideEffectsPending,
/// Pending side effects are preventing [ObservableQuery.close],
/// and the [ObservableQuery] will be discarded after fetch completes
/// and side effects are resolved.
sideEffectsBlocking,
/// The operation was executed and is not [polling]
completed,
/// [ObservableQuery.close] was called and all activity
/// from this [ObservableQuery] has ceased.
closed
}
/// An Observable/Stream-based API for both queries and mutations.
///
/// Returned from [GraphQLClient.watchQuery] for use in reactive programming,
/// for instance in `graphql_flutter` widgets.
/// It is modelled closely after [Apollo's ObservableQuery][apollo_oq]
///
/// [ObservableQuery]'s core api/usage is to [fetchResults], then listen to the [stream].
/// [fetchResults] will be called on instantiation if [options.eagerlyFetchResults] is set,
/// which in turn defaults to [options.fetchResults].
///
/// Beyond that, [ObservableQuery] is a bit of a kitchen sink:
/// * There are [refetch] and [fetchMore] methods for fetching more results
/// * An [onData] method for registering callbacks (namely for mutations)
/// * [lifecycle] for tracking polling, side effect, an inflight execution state
/// * [latestResult]  the most recent result from this operation
///
/// And a handful of internally leveraged methods.
///
/// [apollo_oq]: https://www.apollographql.com/docs/react/v3.0-beta/api/core/ObservableQuery/
class ObservableQuery<TParsed> {
ObservableQuery({
required this.queryManager,
required this.options,
}) : queryId = queryManager.generateQueryId().toString() {
if (options.eagerlyFetchResults) {
_latestWasEagerlyFetched = true;
fetchResults();
}
controller = StreamController<QueryResult<TParsed>>.broadcast(
onListen: onListen,
);
}
// set to true when eagerly fetched to prevent back-to-back queries
bool _latestWasEagerlyFetched = false;
/// The identity of this query within the [QueryManager]
final String queryId;
@protected
final QueryManager queryManager;
@protected
QueryScheduler? get scheduler => queryManager.scheduler;
/// callbacks registered with [onData]
List<OnData> _onDataCallbacks = [];
/// call [queryManager.maybeRebroadcastQueries] after all other [_onDataCallbacks]
///
/// Automatically appended as an [OnData]
FutureOr<void> _maybeRebroadcast(QueryResult? result) =>
queryManager.maybeRebroadcastQueries(exclude: this);
/// The most recently seen result from this operation's stream
QueryResult<TParsed>? latestResult;
QueryLifecycle lifecycle = QueryLifecycle.unexecuted;
WatchQueryOptions<TParsed> options;
late StreamController<QueryResult<TParsed>> controller;
Stream<QueryResult<TParsed>> get stream => controller.stream;
bool get isCurrentlyPolling => lifecycle == QueryLifecycle.polling;
bool get isRefetchSafe {
if (!options.isQuery) {
return false;
}
switch (lifecycle) {
case QueryLifecycle.completed:
case QueryLifecycle.polling:
case QueryLifecycle.pollingStopped:
return true;
case QueryLifecycle.pending:
case QueryLifecycle.closed:
case QueryLifecycle.unexecuted:
case QueryLifecycle.sideEffectsPending:
case QueryLifecycle.sideEffectsBlocking:
return false;
}
}
/// Attempts to refetch _on the network_, throwing error if not refetch safe
///
/// **NOTE:** overrides any present non-network-only [FetchPolicy],
/// as refetching from the `cache` does not make sense.
Future<QueryResult<TParsed>?> refetch() {
if (isRefetchSafe) {
addResult(QueryResult.loading(
data: latestResult?.data,
parserFn: options.parserFn,
));
return queryManager.refetchQuery<TParsed>(queryId);
}
throw Exception('Query is not refetch safe');
}
/// Whether it is safe to rebroadcast results due to cache
/// changes based on policies and [lifecycle].
///
/// Called internally by the [QueryManager]
bool get isRebroadcastSafe {
if (!options.policies.allowsRebroadcasting) {
return false;
}
switch (lifecycle) {
case QueryLifecycle.pending:
case QueryLifecycle.completed:
case QueryLifecycle.polling:
case QueryLifecycle.pollingStopped:
return true;
case QueryLifecycle.unexecuted: // this might be ok
case QueryLifecycle.closed:
case QueryLifecycle.sideEffectsPending:
case QueryLifecycle.sideEffectsBlocking:
return false;
}
}
void onListen() {
if (_latestWasEagerlyFetched) {
_latestWasEagerlyFetched = false;
// eager results are resolved synchronously,
// so we have to add them manually now that
// the stream is available
if (!controller.isClosed && latestResult != null) {
controller.add(latestResult!);
}
return;
}
if (options.fetchResults) {
fetchResults();
}
}
/// Fetch results based on [options.fetchPolicy]
///
/// Will [startPolling] if [options.pollInterval] is set
MultiSourceResult<TParsed> fetchResults() {
final MultiSourceResult<TParsed> allResults =
queryManager.fetchQueryAsMultiSourceResult(queryId, options);
latestResult ??= allResults.eagerResult;
if (allResults.networkResult == null) {
// This path is only possible for cacheFirst and cacheOnly fetch policies.
lifecycle = QueryLifecycle.completed;
} else {
// if onData callbacks have been registered,
// they are waited on by default
lifecycle = _onDataCallbacks.isNotEmpty
? QueryLifecycle.sideEffectsPending
: QueryLifecycle.pending;
}
if (options.pollInterval != null && options.pollInterval! > Duration.zero) {
startPolling(options.pollInterval);
}
return allResults;
}
/// fetch more results and then merge them with the [latestResult]
/// according to [FetchMoreOptions.updateQuery].
///
/// The results will then be added to to stream for listeners to react to,
/// such as for triggering `grahphql_flutter` widget rebuilds
///
/// **NOTE**: with the addition of strict data structure checking in v4,
/// it is easy to make mistakes in writing [updateQuery].
///
/// To mitigate this, [FetchMoreOptions.partial] has been provided.
Future<QueryResult<TParsed>> fetchMore(
FetchMoreOptions fetchMoreOptions) async {
addResult(QueryResult.loading(
data: latestResult?.data,
parserFn: options.parserFn,
));
return fetchMoreImplementation(
fetchMoreOptions,
originalOptions: options,
queryManager: queryManager,
previousResult: latestResult!,
queryId: queryId,
);
}
/// Add a [result] to the [stream] unless it was created
/// before [lasestResult].
///
/// Copies the [QueryResult.source] from the [latestResult]
/// if it is set to `null`.
///
/// Called internally by the [QueryManager]
void addResult(QueryResult<TParsed> result, {bool fromRebroadcast = false}) {
// don't overwrite results due to some async/optimism issue
if (latestResult != null &&
latestResult!.timestamp.isAfter(result.timestamp)) {
return;
}
if (options.carryForwardDataOnException && result.hasException) {
result.data ??= latestResult?.data;
}
if (lifecycle == QueryLifecycle.pending && result.isConcrete) {
lifecycle = QueryLifecycle.completed;
}
latestResult = result;
// TODO should callbacks be applied before or after streaming
if (!controller.isClosed) {
controller.add(result);
}
if (result.isNotLoading) {
_applyCallbacks(result, fromRebroadcast: fromRebroadcast);
}
}
// most mutation behavior happens here
/// Register [callbacks] to trigger when [stream] has new results
/// where [QueryResult.isNotLoading]
///
/// Will deregister [callbacks] after calling them on the first
/// result that [QueryResult.isConcrete],
/// handling the resolution of [lifecycle] from
/// [QueryLifecycle.sideEffectsBlocking] to [QueryLifecycle.completed]
/// as appropriate
void onData(Iterable<OnData> callbacks) => _onDataCallbacks.addAll(callbacks);
/// Applies [onData] callbacks at the end of [addResult]
///
/// [fromRebroadcast] is used to avoid the super-edge case of infinite rebroadcasts
/// (not sure if it's even possible)
void _applyCallbacks(
QueryResult? result, {
bool fromRebroadcast = false,
}) async {
final callbacks = [
..._onDataCallbacks,
if (!fromRebroadcast) _maybeRebroadcast
];
for (final callback in callbacks) {
await callback(result);
}
if (lifecycle == QueryLifecycle.closed) {
// .close(force: true) was called
return;
}
if (result!.isConcrete) {
// avoid removing new callbacks
_onDataCallbacks.removeWhere((cb) => callbacks.contains(cb));
// if there are new callbacks, there is maybe another inflight mutation
if (_onDataCallbacks.isEmpty) {
if (lifecycle == QueryLifecycle.sideEffectsBlocking) {
lifecycle = QueryLifecycle.completed;
close();
}
// the mutation has been completed, but disposal has not been requested
if (lifecycle == QueryLifecycle.sideEffectsPending) {
lifecycle = QueryLifecycle.completed;
}
}
}
}
/// Poll the server periodically for results.
///
/// Will be called by [fetchResults] automatically if [options.pollInterval] is set
void startPolling(Duration? pollInterval) {
if (options.fetchPolicy == FetchPolicy.cacheFirst ||
options.fetchPolicy == FetchPolicy.cacheOnly) {
throw Exception(
'Queries that specify the cacheFirst and cacheOnly fetch policies cannot also be polling queries.',
);
}
if (isCurrentlyPolling) {
scheduler!.stopPollingQuery(queryId);
}
options.pollInterval = pollInterval;
lifecycle = QueryLifecycle.polling;
scheduler!.startPollingQuery(options, queryId);
}
void stopPolling() {
if (isCurrentlyPolling) {
scheduler!.stopPollingQuery(queryId);
options.pollInterval = null;
lifecycle = QueryLifecycle.pollingStopped;
}
}
set variables(Map<String, dynamic> variables) =>
options.variables = variables;
/// [onData] callbacks have het to be run
///
/// inlcudes `lifecycle == QueryLifecycle.sideEffectsBlocking`
bool get sideEffectsArePending =>
(lifecycle == QueryLifecycle.sideEffectsPending ||
lifecycle == QueryLifecycle.sideEffectsBlocking);
/// Closes the query or mutation, or else queues it for closing.
///
/// To preserve Mutation side effects, [close] checks the [lifecycle],
/// queuing the stream for closing if [sideEffectsArePending].
/// You can override this check with `force: true`.
///
/// Returns a [FutureOr] of the resultant lifecycle, either
/// [QueryLifecycle.sideEffectsBlocking] or [QueryLifecycle.closed]
FutureOr<QueryLifecycle> close({
bool force = false,
bool fromManager = false,
}) async {
if (lifecycle == QueryLifecycle.sideEffectsPending && !force) {
lifecycle = QueryLifecycle.sideEffectsBlocking;
// stop closing because we're waiting on something
return lifecycle;
}
// `fromManager` is used by the query manager when it wants to close a query to avoid infinite loops
if (!fromManager) {
queryManager.closeQuery(this, fromQuery: true);
}
stopPolling();
await controller.close();
lifecycle = QueryLifecycle.closed;
return QueryLifecycle.closed;
}
}
@@ -0,0 +1,72 @@
import 'package:fl_query/src/core/subscribable.dart';
import 'package:internet_connection_checker/internet_connection_checker.dart';
typedef SetupFn = void Function()? Function(
void Function([bool? online]) setOnline);
class _OnlineManager extends Subscribable {
bool? _online;
void Function()? _cleanup;
SetupFn? _setup;
_OnlineManager() {
_setup = (listener) {
var subscription = InternetConnectionChecker()
.onStatusChange
.listen((status) => listener());
return () {
subscription.cancel();
};
};
}
@override
void onSubscribe() {
if (_cleanup == null) {
setEventListener(_setup!);
}
}
@override
void onUnsubscribe() {
if (!hasListeners()) {
_cleanup?.call();
_cleanup = null;
}
}
setEventListener(SetupFn setup) {
_setup = setup;
_cleanup?.call();
_cleanup = setup(([bool? online]) {
if (online != null) {
setOnline(online);
} else {
onOnline();
}
});
}
void setOnline(bool? online) {
_online = online;
if (online != null && online) {
onOnline();
}
}
void onOnline() {
listeners.forEach((listener) {
listener();
});
}
Future<bool> isOnline() {
if (_online != null) {
return Future.value(_online!);
}
return InternetConnectionChecker().hasConnection;
}
}
_OnlineManager onlineManager = _OnlineManager();
@@ -1,323 +0,0 @@
import 'package:fl_query/fl_query.dart';
import 'package:meta/meta.dart';
import "package:collection/collection.dart";
/// [FetchPolicy] determines where the client may return a result from.
///
/// * [cacheFirst]: return result from cache. Only fetch from network if cached result is not available.
/// * [cacheAndNetwork]: return result from cache first (if it exists), then return network result once it's available.
/// * [cacheOnly]: return result from cache if available, fail otherwise.
/// * [noCache]: return result from network, fail if network call doesn't succeed, don't save to cache.
/// * [networkOnly]: return result from network, fail if network call doesn't succeed, save to cache.
///
/// The default `fetchPolicy` for each method are:
/// * `watchQuery`: [cacheAndNetwork]
/// * `watchMutation`: [cacheAndNetwork]
/// * `query`: [cacheFirst]
/// * `mutation`: [networkOnly]
/// * `subscribe`: [networkOnly]
///
/// These can be overriden at client construction time by passing
/// a [DefaultPolicies] instance to `defaultPolicies`.
enum FetchPolicy {
/// Return result from cache. Only fetch from network if cached result is not available.
cacheFirst,
/// Return result from cache first (if it exists), then return network result once it's available.
cacheAndNetwork,
/// Return result from cache if available, fail otherwise.
cacheOnly,
/// Return result from network, fail if network call doesn't succeed, don't save to cache.
noCache,
/// Return result from network, fail if network call doesn't succeed, save to cache.
networkOnly,
}
// TODO investigate the relationship between optimistic results
// and policy in flutter
bool shouldRespondEagerlyFromCache(FetchPolicy? fetchPolicy) =>
fetchPolicy == FetchPolicy.cacheFirst ||
fetchPolicy == FetchPolicy.cacheAndNetwork ||
fetchPolicy == FetchPolicy.cacheOnly;
bool shouldStopAtCache(FetchPolicy? fetchPolicy) =>
fetchPolicy == FetchPolicy.cacheFirst ||
fetchPolicy == FetchPolicy.cacheOnly;
bool willAlwaysExecuteOnNetwork(FetchPolicy? policy) {
switch (policy) {
case FetchPolicy.noCache:
case FetchPolicy.networkOnly:
return true;
case FetchPolicy.cacheFirst:
case FetchPolicy.cacheAndNetwork:
case FetchPolicy.cacheOnly:
case null:
return false;
}
}
/// [ErrorPolicy] determines the level of events for GraphQL Errors in the execution result. The options are:
///
/// While the default for all client methods is [none],
/// [all] is recommended for notifying your users of potential issues.
///
/// * [none] (default): Any GraphQL Errors are treated the same as network errors and any data is ignored from the response.
/// * [ignore]: Ignore allows you to read any data that is returned alongside GraphQL Errors,
/// but doesn't save the errors or report them to your UI.
/// * [all]: Saves both data and errors into the `cache` so your UI can use them.
/// It is recommended for notifying your users of potential issues,
/// while still showing as much data as possible from your server.
///
/// **NOTE**: [ErrorPolicy] only effects **GraphQL Errors**.
/// Client side and network exceptions are added to a [QueryResult] as they occur,
/// and can co-exist alongside data.
enum ErrorPolicy {
/// Any GraphQL Errors are treated the same as network errors and any data is ignored from the response. (default)
none,
/// Ignore allows you to read any data that is returned alongside GraphQL Errors,
/// but doesn't save the errors or report them to your UI.
ignore,
/// Saves both data and errors into the `cache` so your UI can use them.
///
/// It is recommended for notifying your users of potential issues,
/// while still showing as much data as possible from your server.
all,
}
/// [CacheRereadPolicy] determines whether and how cache data will be merged into
/// the final [QueryResult] `data` before it is returned.
///
/// It _does not_ effect `optimisticResults` added to [QueryOptions], etc.
///
/// * [mergeOptimistic]: Merge relevant optimistic data from the cache before returning.
/// * [ignoreOptimistic]: Ignore optimistic data, but still allow for non-optimistic cache rebroadcasts
/// **if applicable**.
/// * [ignoreAll]: Ignore all cache data besides the result, and never rebroadcast the result,
/// even if the underlying cache data changes.
///
/// The default `cacheRereadPolicy` for each method are:
/// * `watchQuery`: [mergeOptimistic]
/// * `watchMutation`: [ignoreAll]
/// * `query`: [mergeOptimistic]
/// * `mutation`: [ignoreAll]
/// * `subscribe`: [mergeOptimistic]
enum CacheRereadPolicy {
/// Merge relevant optimistic data from the cache before returning.
mergeOptimistic,
/// Ignore optimistic data, but still allow for non-optimistic cache rebroadcasts
/// **if applicable**.
ignoreOptimisitic,
/// Ignore all cache data besides the result, and never rebroadcast the result,
/// even if the underlying cache data changes.
ignoreAll,
}
/// Container for supplying [fetch], [error], and [cacheReread] policies.
///
/// If any are `null`, the appropriate policy will be selected from [DefaultPolicies]
@immutable
class Policies {
/// Specifies the [FetchPolicy] to be used.
final FetchPolicy? fetch;
/// Specifies the [ErrorPolicy] to be used.
final ErrorPolicy? error;
/// Specifies the [CacheRereadPolicy] to be used.
final CacheRereadPolicy? cacheReread;
bool get mergeOptimisticData =>
cacheReread == CacheRereadPolicy.mergeOptimistic;
Policies({
this.fetch,
this.error,
this.cacheReread,
});
Policies.safe(
FetchPolicy this.fetch,
ErrorPolicy this.error,
CacheRereadPolicy this.cacheReread,
);
Policies withOverrides([Policies? overrides]) => Policies.safe(
overrides?.fetch ?? fetch!,
overrides?.error ?? error!,
overrides?.cacheReread ?? cacheReread!,
);
Policies copyWith({FetchPolicy? fetch, ErrorPolicy? error}) =>
Policies(fetch: fetch, error: error, cacheReread: cacheReread);
operator ==(Object other) =>
identical(this, other) ||
(other is Policies &&
fetch == other.fetch &&
error == other.error &&
cacheReread == other.cacheReread);
@override
int get hashCode => const ListEquality<Object?>(
DeepCollectionEquality(),
).hash([fetch, error, cacheReread]);
/// Returns `false` if either [fetch] or [cacheReread] policies have disabled rebroadcast.
bool get allowsRebroadcasting => !(fetch == FetchPolicy.noCache ||
cacheReread == CacheRereadPolicy.ignoreAll);
@override
String toString() =>
'Policies(fetch: $fetch, error: $error, cacheReread: $cacheReread)';
}
/// The default [Policies] to set for each client action.
@immutable
class DefaultPolicies {
/// The default [Policies] for watchQuery.
/// Defaults to
/// ```
/// Policies(
/// FetchPolicy.cacheAndNetwork,
/// ErrorPolicy.none,
/// CacheRereadPolicy.mergeOptimistic,
/// )
/// ```
final Policies watchQuery;
/// The default [Policies] for watchMutation.
/// Defaults to
/// ```
/// Policies(
/// FetchPolicy.networkOnly,
/// ErrorPolicy.none,
/// CacheRereadPolicy.ignoreAll,
/// )
/// ```
final Policies watchMutation;
/// The default [Policies] for query.
/// Defaults to
/// ```
/// Policies(
/// FetchPolicy.cacheFirst,
/// ErrorPolicy.none,
/// CacheRereadPolicy.mergeOptimistic,
/// )
/// ```
final Policies query;
/// The default [Policies] for mutate.
/// Defaults to
/// ```
/// Policies(
/// FetchPolicy.networkOnly,
/// ErrorPolicy.none,
/// CacheRereadPolicy.ignore,
/// )
/// ```
final Policies mutate;
/// The default [Policies] for subscribe.
/// Defaults to
/// ```
/// Policies(
/// FetchPolicy.networkOnly,
/// ErrorPolicy.none,
/// CacheRereadPolicy.mergeOptimistic,
/// )
/// ```
///
/// The subscription spec is very flexible, so we default to `FetchPolicy.networkOnly`
/// to avoid breaking some use-cases by default.
///
/// `FetchPolicy.cacheOnly` is invalid for subscriptions. This is because `FetchPolicy` changes do
/// little to change subscription behavior, only determining
/// whether an eager result is first read from the cache.
final Policies subscribe;
DefaultPolicies({
Policies? watchQuery,
Policies? watchMutation,
Policies? query,
Policies? mutate,
Policies? subscribe,
}) : watchQuery = _watchQueryDefaults.withOverrides(watchQuery),
watchMutation = _mutateDefaults.withOverrides(watchMutation),
query = _queryDefaults.withOverrides(query),
mutate = _mutateDefaults.withOverrides(mutate),
subscribe = _subscribeDefaults.withOverrides(subscribe);
static final _watchQueryDefaults = Policies.safe(
FetchPolicy.cacheAndNetwork,
ErrorPolicy.none,
CacheRereadPolicy.mergeOptimistic,
);
static final _queryDefaults = Policies.safe(
FetchPolicy.cacheFirst,
ErrorPolicy.none,
CacheRereadPolicy.mergeOptimistic,
);
static final _mutateDefaults = Policies.safe(
FetchPolicy.networkOnly,
ErrorPolicy.none,
CacheRereadPolicy.ignoreAll,
);
static final _subscribeDefaults = Policies.safe(
FetchPolicy.networkOnly,
ErrorPolicy.none,
CacheRereadPolicy.mergeOptimistic,
);
DefaultPolicies copyWith({
Policies? watchQuery,
Policies? query,
Policies? watchMutation,
Policies? mutate,
Policies? subscribe,
}) =>
DefaultPolicies(
watchQuery: watchQuery,
query: query,
watchMutation: watchMutation,
mutate: mutate,
subscribe: subscribe,
);
List<Object> _getChildren() => [
watchQuery,
query,
watchMutation,
mutate,
subscribe,
];
@override
bool operator ==(Object o) =>
identical(this, o) ||
(o is DefaultPolicies &&
const ListEquality<Object?>(
DeepCollectionEquality(),
).equals(
o._getChildren(),
_getChildren(),
));
@override
int get hashCode => const ListEquality<Object?>(
DeepCollectionEquality(),
).hash(
_getChildren(),
);
}
+642
View File
@@ -0,0 +1,642 @@
import 'dart:async';
import 'dart:math';
import 'package:fl_query/src/core/models.dart';
import 'package:fl_query/src/core/notify_manager.dart';
import 'package:fl_query/src/core/query_cache.dart';
import 'package:fl_query/src/core/query_key.dart';
import 'package:fl_query/src/core/query_observer.dart';
import 'package:fl_query/src/core/retryer.dart';
import 'package:fl_query/src/core/utils.dart';
import 'package:meta/meta.dart';
import 'package:collection/collection.dart';
class FetchOptions {
bool? cancelRefetch;
dynamic meta;
FetchOptions({this.cancelRefetch, this.meta});
}
class FetchContext<TQueryFnData, TError, TData> {
FutureOr Function() fetchFn;
FetchOptions? fetchOptions;
QueryOptions<TQueryFnData, TError, TData> options;
QueryKey queryKey;
QueryState<TData, TError> state;
QueryMeta? meta;
FetchContext({
required this.fetchFn,
required this.options,
required this.queryKey,
required this.state,
this.meta,
this.fetchOptions,
});
}
class QueryBehavior<TQueryFnData, TError, TData> {
void Function(FetchContext<TQueryFnData, TError, TData> context) onFetch;
QueryBehavior({required this.onFetch});
}
class QueryState<TData, TError> {
TData? data;
TError? error;
QueryStatus status;
DateTime? dataUpdatedAt;
int dataUpdateCount;
DateTime? errorUpdatedAt;
int errorUpdateCount;
int fetchFailureCount;
dynamic fetchMeta;
bool isFetching;
bool isInvalidated;
bool isPaused;
QueryState({
required this.status,
required this.dataUpdatedAt,
required this.dataUpdateCount,
required this.errorUpdatedAt,
required this.errorUpdateCount,
required this.fetchFailureCount,
required this.fetchMeta,
required this.isFetching,
required this.isInvalidated,
required this.isPaused,
this.data,
this.error,
});
QueryState.fromJson(Map<String, dynamic> json)
: data = json["data"],
error = json["error"],
status = json["status"],
dataUpdatedAt = json["dataUpdatedAt"],
dataUpdateCount = json["dataUpdateCount"],
errorUpdatedAt = json["errorUpdatedAt"],
errorUpdateCount = json["errorUpdateCount"],
fetchFailureCount = json["fetchFailureCount"],
fetchMeta = json["fetchMeta"],
isFetching = json["isFetching"],
isInvalidated = json["isInvalidated"],
isPaused = json["isPaused"];
Map<String, dynamic> toJson() {
return {
"data": data,
"error": error,
"status": status,
"dataUpdatedAt": dataUpdatedAt,
"dataUpdateCount": dataUpdateCount,
"errorUpdatedAt": errorUpdatedAt,
"errorUpdateCount": errorUpdateCount,
"fetchFailureCount": fetchFailureCount,
"fetchMeta": fetchMeta,
"isFetching": isFetching,
"isInvalidated": isInvalidated,
"isPaused": isPaused,
};
}
}
enum ActionType {
failed,
fetch,
success,
error,
invalidate,
pause,
resume,
setState,
}
class SetStateOptions {
Object? meta;
SetStateOptions({this.meta});
Map<String, dynamic> toJson() {
return {"meta": meta};
}
}
class Action<TData, TError> {
ActionType type;
Object? meta;
TData? data;
DateTime? dataUpdatedAt;
TError? error;
QueryState<TData, TError>? state;
SetStateOptions? setStateOptions;
Action(
this.type, {
this.meta,
this.data,
this.dataUpdatedAt,
this.error,
this.state,
this.setStateOptions,
}) {
if (type == ActionType.error && error == null)
throw Exception(
"[Action.Action] property `error` can't be null when `type` = `$type`");
if (type == ActionType.setState && state == null)
throw Exception(
"[Action.Action] property `state` can't be null when `type` = `$type`");
}
Map<String, dynamic> toJson() {
return {
"type": type,
"meta": meta,
"data": data,
"dataUpdatedAt": dataUpdatedAt,
"error": error,
"state": state,
"setStateOptions": setStateOptions,
};
}
}
class Query<TQueryFnData, TError, TData> {
QueryKey queryKey;
String queryHash;
late QueryOptions<TQueryFnData, TError, TData> options;
late QueryState<TData, TError> initialState;
QueryState<TData, TError>? revertState;
late QueryState<TData, TError> state;
Duration? cacheTime;
QueryMeta? meta;
QueryCache _cache;
Future<TData>? _future;
Timer? _gcTimeout;
Retryer<TData, TError>? _retryer;
List<QueryObserver> _observers;
QueryOptions<TQueryFnData, TError, TData>? _defaultOptions;
bool _abortSignalConsumed;
bool _hadObservers;
Query({
required this.queryKey,
required this.queryHash,
required QueryCache cache,
QueryOptions<TQueryFnData, TError, TData>? options,
QueryOptions<TQueryFnData, TError, TData>? defaultOptions,
QueryState<TData, TError>? state,
QueryMeta? meta,
}) : _abortSignalConsumed = false,
_hadObservers = false,
_defaultOptions = defaultOptions,
_observers = [],
_cache = cache {
_setOptions(options);
initialState = state ?? _getDefaultState(this.options);
this.state = initialState;
this.meta = meta;
_scheduleGc();
}
void _scheduleGc() {
this._clearGcTimeout();
if (this.cacheTime != null) {
_gcTimeout = Timer(cacheTime!, () {
this._optionalRemove();
});
}
}
void _clearGcTimeout() {
_gcTimeout?.cancel();
_gcTimeout = null;
}
void _optionalRemove() {
if (_observers.isEmpty) {
if (state.isFetching) {
if (_hadObservers) {
_scheduleGc();
}
} else {
_cache.remove(this);
}
}
}
void _setOptions(QueryOptions<TQueryFnData, TError, TData>? options) {
this.options = QueryOptions.fromJson({
...(_defaultOptions?.toJson() ?? {}),
...(options?.toJson() ?? {}),
});
meta = options?.meta;
/// Default to [5 minutes] if cache time isn't set
cacheTime = Duration(
milliseconds: max(
cacheTime?.inMilliseconds ?? 0,
this.options.cacheTime?.inMilliseconds ?? 5 * 60 * 1000,
));
}
QueryState<TData, TError> _getDefaultState(
QueryOptions<TQueryFnData, TError, TData> options) {
var data = options.initialData;
bool hasData = data != null;
DateTime? initialDataUpdatedAt =
hasData ? options.initialDataUpdatedAt : null;
return QueryState(
data: data,
dataUpdateCount: 0,
dataUpdatedAt: hasData ? initialDataUpdatedAt ?? DateTime.now() : null,
error: null,
errorUpdateCount: 0,
errorUpdatedAt: null,
fetchFailureCount: 0,
fetchMeta: null,
isFetching: false,
isInvalidated: false,
isPaused: false,
status: hasData ? QueryStatus.success : QueryStatus.idle,
);
}
TData setData(
DataUpdateFunction<TData?, TData> updater, {
DateTime? updatedAt,
}) {
var prevData = this.state.data;
var data = updater(prevData);
// Use prev data if an isDataEqual function is defined and returns `true`
if (this.options.isDataEqual?.call(prevData, data) == true) {
data = prevData as TData;
} else if (this.options.structuralSharing != false) {
// Structurally share data between prev and new data if needed
data = replaceEqualDeep<TData>(prevData ?? {} as TData, data);
}
// Set data and mark it as cached
_dispatch(Action(
ActionType.success,
data: data,
dataUpdatedAt: updatedAt,
));
return data;
}
void setState(
QueryState<TData, TError> state, [
SetStateOptions? setStateOptions,
]) {
_dispatch(Action(
ActionType.setState,
state: state,
setStateOptions: setStateOptions,
));
}
Future<void> cancel({bool? revert, bool? silent}) {
var future = _future;
_retryer?.cancel(revert: revert, silent: silent);
return future != null ? future.then(noop).catchError(noop) : Future.value();
}
void reset() {
destroy();
setState(initialState);
}
destroy() {
_clearGcTimeout();
cancel(silent: true);
}
bool isActive() {
return _observers.any((observer) => observer.options.enabled != false);
}
bool isFetching() {
return this.state.isFetching;
}
Future<TData> fetch([
QueryOptions<TQueryFnData, TError, TData>? options,
ObserverFetchOptions? fetchOptions,
]) {
if (this.state.isFetching) {
if (this.state.dataUpdatedAt != null &&
fetchOptions?.cancelRefetch == true) {
// Silently cancel current fetch if the user wants to cancel re-fetches
this.cancel(silent: true);
} else if (_future != null) {
// make sure that retries that were potentially cancelled due to unmounts can continue
_retryer?.continueRetry();
// Return current promise if we are already fetching
return _future!;
}
}
// Update config if passed, otherwise the config from the last execution is used
if (options != null) {
_setOptions(options);
}
// Use the options from the first observer with a query function if no function is found.
// This can happen when the query is hydrated or created with setQueryData.
if (this.options.queryFn == null) {
var observer =
_observers.firstWhereOrNull((x) => x.options.queryFn != null);
if (observer != null) {
_setOptions(QueryOptions(
queryKey: observer.options.queryKey,
queryKeyHashFn: observer.options.queryKeyHashFn,
cacheTime: observer.options.cacheTime,
isDataEqual: observer.options.isDataEqual,
queryFn: observer.options.queryFn,
queryHash: observer.options.queryHash,
initialData: observer.options.initialData,
initialDataUpdatedAt: observer.options.initialDataUpdatedAt,
meta: observer.options.meta,
structuralSharing: observer.options.structuralSharing,
defaulted: observer.options.defaulted,
));
}
}
QueryFunctionContext queryFnContext = QueryFunctionContext(
queryKey: queryKey,
meta: meta,
);
/// !!LANGUAGE LIMITATION!! There's no equivalent of [AbortController]
/// the [get] can be implemented using Dart's getter but it'd be
/// useless since there's no equivalent of AbortController.
/// Have to find a better way to control ABORTION
// Object.defineProperty(queryFnContext, 'signal', {
// enumerable: true,
// get: () {
// if (abortController) {
// this.abortSignalConsumed = true
// return abortController.signal
// }
// return undefined
// },
// })
// Create fetch function
fetchFn() {
if (this.options.queryFn == null) {
return Future.error('Missing queryFn');
}
_abortSignalConsumed = false;
return options?.queryFn?.call(queryFnContext);
}
// Trigger behavior hook
FetchContext<TQueryFnData, TError, TData> context = FetchContext(
fetchOptions: fetchOptions,
options: this.options,
queryKey: queryKey,
state: this.state,
fetchFn: fetchFn,
meta: this.meta,
);
this.options.behavior?.onFetch(context);
// Store state in case the current fetch needs to be reverted
this.revertState = this.state;
// Set to fetching state if not already in it
if (!this.state.isFetching ||
this.state.fetchMeta != context.fetchOptions?.meta) {
_dispatch(Action(ActionType.fetch, meta: context.fetchOptions?.meta));
}
_retryer = Retryer(
fn: context.fetchFn as FutureOr<TData> Function(),
// abort: abortController?.abort?.bind(abortController),
onSuccess: (data) {
this.setData((_) => data);
// Notify cache callback
_cache.onData?.call(data, this);
// Remove query after fetching if cache time is 0
if (this.cacheTime == null || this.cacheTime == Duration.zero) {
_optionalRemove();
}
},
onError: (TError error) {
// Optimistically update state if needed
if (!(isCancelledError(error) && (error as dynamic)?.silent == true)) {
_dispatch(Action(ActionType.error, error: error));
}
if (!isCancelledError(error)) {
// Notify cache callback
_cache.onError?.call(error, this);
// Log error
// getLogger().error(error);
}
// Remove query after fetching if cache time is 0
if (this.cacheTime == null || this.cacheTime == Duration.zero) {
_optionalRemove();
}
},
onFail: (failureCount, error) {
_dispatch(Action(ActionType.failed));
},
onPause: () {
_dispatch(Action(ActionType.pause));
},
onContinue: () {
_dispatch(Action(ActionType.resume));
},
retry: context.options.retry,
retryDelay: context.options.retryDelay,
);
this._future = _retryer!.future;
return this._future!;
}
void _dispatch(Action<TData, TError> action) {
this.state = this.reducer(this.state, action);
notifyManager.batch(() {
_observers.forEach((observer) {
observer.onQueryUpdate(action);
});
_cache.notify(QueryCacheNotifyEvent(
QueryCacheNotifyEventType.queryUpdated,
this,
action: action,
));
});
}
void addObserver(QueryObserver observer) {
if (_observers.indexOf(observer) == -1) {
_observers.add(observer);
_hadObservers = true;
// Stop the query from being garbage collected
_clearGcTimeout();
_cache.notify(QueryCacheNotifyEvent(
QueryCacheNotifyEventType.observerAdded,
this,
observer: observer,
));
}
}
void removeObserver(QueryObserver observer) {
if (_observers.indexOf(observer) != -1) {
_observers = _observers.where((x) => x != observer).toList();
if (_observers.isEmpty) {
// If the transport layer does not support cancellation
// we'll let the query continue so the result can be cached
if (_retryer != null) {
if (_retryer?.isTransportCancelable == true || _abortSignalConsumed) {
_retryer?.cancel(revert: true);
} else {
_retryer?.cancelRetry();
}
}
if (cacheTime != null) {
_scheduleGc();
} else {
_cache.remove(this);
}
}
_cache.notify(QueryCacheNotifyEvent(
QueryCacheNotifyEventType.observerRemoved,
this,
observer: observer,
));
}
}
int getObserversCount() {
return _observers.length;
}
void invalidate() {
if (!this.state.isInvalidated) {
_dispatch(Action(ActionType.invalidate));
}
}
bool isStale() {
return (this.state.isInvalidated ||
this.state.dataUpdatedAt == null ||
_observers.any((observer) => observer.getCurrentResult().isStale));
}
bool isStaleByTime(Duration? staleTime) {
return (this.state.isInvalidated ||
this.state.dataUpdatedAt == null ||
timeUntilStale(this.state.dataUpdatedAt!, staleTime) == Duration.zero);
}
void onOnline() {
var observer = _observers
.firstWhereOrNull((x) => x.shouldFetchCurrentQueryOnReconnect());
if (observer != null) {
observer.refetch();
}
// Continue fetch if currently paused
_retryer?.continueFn();
}
@protected
QueryState<TData, TError> reducer(
QueryState<TData, TError> state,
Action<TData, TError> action,
) {
switch (action.type) {
case ActionType.failed:
return QueryState.fromJson({
...state.toJson(),
"fetchFailureCount": state.fetchFailureCount + 1,
});
case ActionType.fetch:
return QueryState.fromJson({
...state.toJson(),
"fetchFailureCount": 0,
"fetchMeta": action.meta,
"isFetching": true,
"isPaused": false,
if (state.dataUpdatedAt == null)
...({
"error": null,
"status": QueryStatus.loading,
})
});
case ActionType.success:
return QueryState.fromJson({
...state.toJson(),
"data": action.data,
"dataUpdateCount": state.dataUpdateCount + 1,
"dataUpdatedAt": action.dataUpdatedAt ?? DateTime.now(),
"error": null,
"fetchFailureCount": 0,
"isFetching": false,
"isInvalidated": false,
"isPaused": false,
"status": QueryStatus.success,
});
case ActionType.error:
var error = action.error as dynamic;
if (isCancelledError(error) &&
error?.revert == true &&
revertState != null) {
return QueryState.fromJson(revertState!.toJson());
}
return QueryState.fromJson({
...state.toJson(),
"error": error as TError,
"errorUpdateCount": state.errorUpdateCount + 1,
"errorUpdatedAt": DateTime.now(),
"fetchFailureCount": state.fetchFailureCount + 1,
"isFetching": false,
"isPaused": false,
"status": QueryStatus.error,
});
case ActionType.invalidate:
return QueryState.fromJson({
...state.toJson(),
"isInvalidated": true,
});
case ActionType.pause:
return QueryState.fromJson({
...state.toJson(),
"isPaused": true,
});
case ActionType.resume:
return QueryState.fromJson({
...state.toJson(),
"isPaused": false,
});
case ActionType.setState:
return QueryState.fromJson({
...state.toJson(),
...(action.state?.toJson() ?? {}),
});
default:
return state;
}
}
}
@@ -0,0 +1,174 @@
import 'package:fl_query/src/core/models.dart';
import 'package:fl_query/src/core/notify_manager.dart';
import 'package:fl_query/src/core/query.dart';
import 'package:fl_query/src/core/query_client.dart';
import 'package:fl_query/src/core/query_key.dart';
import 'package:fl_query/src/core/subscribable.dart';
import 'package:fl_query/src/core/utils.dart';
import 'package:collection/collection.dart';
enum QueryCacheNotifyEventType {
observerResultsUpdated,
observerRemoved,
observerAdded,
queryUpdated,
queryRemoved,
queryAdded
}
class QueryCacheNotifyEvent {
Query query;
Object? observer;
Object? action;
QueryCacheNotifyEventType type;
QueryCacheNotifyEvent(
this.type,
this.query, {
this.observer,
this.action,
}) {
if ([
QueryCacheNotifyEventType.observerAdded,
QueryCacheNotifyEventType.observerRemoved
].contains(type) &&
observer == null)
throw Exception(
"[QueryCacheNotifyEvent.constructor] property `observer` can't be `null` for `QueryCacheNotifyEventType.observerAdded` & `QueryCacheNotifyEventType.observerRemoved`");
if (type == QueryCacheNotifyEventType.queryUpdated && action == null)
throw Exception(
"[QueryCacheNotifyEvent.constructor] property `action` can't be `null` for `QueryCacheNotifyEventType.queryUpdated`");
}
}
typedef QueryCacheListener = void Function(QueryCacheNotifyEvent? event);
typedef QueryCacheOnError = void Function(dynamic error, Query query);
typedef QueryCacheOnData = void Function(dynamic data, Query query);
typedef QueryHashMap = Map<String, Query>;
class QueryCache extends Subscribable<QueryCacheListener> {
List<Query> _queries;
QueryHashMap _queriesMap;
QueryCacheOnError? onError;
QueryCacheOnData? onData;
QueryCache({
this.onData,
this.onError,
}) : _queries = [],
_queriesMap = {},
super();
Query<TQueryFnData, TError, TData> build<TQueryFnData, TError, TData>(
QueryClient client, QueryOptions<TQueryFnData, TError, TData> options,
[QueryState<TData, TError>? state]) {
QueryKey queryKey = options.queryKey!;
String queryHash =
options.queryHash ?? hashQueryKeyByOptions(queryKey, options);
Query<TQueryFnData, TError, TData>? query =
get<TQueryFnData, TError, TData>(queryHash);
if (query == null) {
query = Query(
cache: this,
queryKey: queryKey,
queryHash: queryHash,
options: client.defaultQueryOptions(options),
state: state,
defaultOptions: client.getQueryDefaults(queryKey),
meta: options.meta,
);
add(query);
}
return query;
}
void add(Query query) {
if (!_queriesMap.containsKey(query.queryHash)) {
_queriesMap[query.queryHash] = query;
_queries.add(query);
notify(
QueryCacheNotifyEvent(
QueryCacheNotifyEventType.queryAdded,
query,
),
);
}
}
void remove(Query query) {
Query? queryInMap = _queriesMap[query.queryHash];
if (queryInMap == null) return;
query.destroy();
_queries = _queries.where((x) => x != query).toList();
if (queryInMap == query) {
_queriesMap.remove(query.queryHash);
}
notify(QueryCacheNotifyEvent(
QueryCacheNotifyEventType.queryRemoved,
query,
));
}
void clear() {
notifyManager.batch(() {
for (var query in _queries) {
remove(query);
}
});
}
Query<TQueryFnData, TError, TData>? get<TQueryFnData, TError, TData>(
String queryHash) {
return _queriesMap[queryHash] as Query<TQueryFnData, TError, TData>?;
}
List<Query> getAll() {
return _queries;
}
Query? find<TQueryFnData, TError, TData>(
QueryKey queryKey,
QueryFilters queryFilters,
) {
queryFilters.exact ??= true;
return _queries
.firstWhereOrNull((query) => matchQuery(queryFilters, query));
}
List<Query> findAll(QueryKey? queryKey, [QueryFilters? filters]) {
if (queryKey == null && filters == null)
throw Exception(
"[QueryCache.findAll] both `queryKey` & `filters` can't be null");
bool filterIsEmpty =
filters?.toJson().entries.every((map) => map.value == null) ?? false;
return filterIsEmpty
? _queries
: _queries.where((query) => matchQuery(filters!, query)).toList();
}
void notify(QueryCacheNotifyEvent event) {
notifyManager.batch(() {
for (var listener in listeners) {
listener(event);
}
});
}
@override
void onSubscribe() {}
@override
void onUnsubscribe() {}
/// Dummy function just to keep the API similar to react-query
void onFocus() {}
void onOnline() {
notifyManager.batch(() {
_queries.forEach((query) {
query.onOnline();
});
});
}
}
@@ -0,0 +1,48 @@
import 'package:fl_query/src/core/models.dart';
import 'package:fl_query/src/core/query_cache.dart';
class QueryClient {
Object? options;
QueryCache _queryCache;
QueryCache _mutationCache;
QueryClient({
QueryCache? queryCache,
QueryCache? mutationCache,
this.options,
}) : _queryCache = queryCache ?? QueryCache(),
_mutationCache = mutationCache ?? QueryCache() {}
QueryObserverOptions<TQueryFnData, TError, TData, TQueryData>
defaultQueryOptions<TQueryFnData, TError, TData, TQueryData>(
QueryObserverOptions<TQueryFnData, TError, TData, TQueryData>?
options) {}
QueryObserverOptions<TQueryFnData, TError, TData, TQueryData>
defaultQueryObserverOptions<TQueryFnData, TError, TData, TQueryData>(
QueryObserverOptions<TQueryFnData, TError, TData, TQueryData>?
options) {
return this.defaultQueryOptions(options);
}
fetchQuery() {}
getQueryData() {}
setQueryData() {}
getQueryState() {}
invalidateQueries() {}
refetchQueries() {}
cancelQueries() {}
removeQueries() {}
resetQueries() {}
bool get isFetching => false;
bool get isMutating => false;
getDefaultOptions() {}
setDefaultOptions() {}
getQueryDefaults() {}
setQueryDefaults() {}
getMutationDefaults() {}
setMutationDefaults() {}
QueryCache getQueryCache() {}
getMutationCache() {}
clear() {}
}
@@ -9,6 +9,7 @@ class QueryKey {
QueryKey.parse(String keyStr) : _key = keyStr.split(".");
String get key => _key.map((k) => k.replaceAll(".", "")).join(".");
List<String> get keyAsList => _key;
@override
String toString() {
@@ -1,547 +0,0 @@
import 'dart:async';
import 'package:meta/meta.dart';
import 'package:collection/collection.dart';
import 'package:fl_query/src/core/result_parser.dart';
import 'package:fl_query/src/cache/cache.dart';
import 'package:fl_query/src/core/observable_query.dart';
import 'package:fl_query/src/core/_base_options.dart';
import 'package:fl_query/src/core/mutation_options.dart';
import 'package:fl_query/src/core/query_options.dart';
import 'package:fl_query/src/core/query_result.dart';
import 'package:fl_query/src/core/policies.dart';
import 'package:fl_query/src/exceptions.dart';
import 'package:fl_query/src/scheduler/scheduler.dart';
import 'package:fl_query/src/core/_query_write_handling.dart';
bool Function(dynamic a, dynamic b) _deepEquals =
const DeepCollectionEquality().equals;
class QueryManager {
QueryManager({
required this.link,
required this.cache,
this.alwaysRebroadcast = false,
}) {
scheduler = QueryScheduler(
queryManager: this,
);
}
final Link link;
final QueryCache cache;
/// Whether to skip deep equality checks in [maybeRebroadcastQueries]
final bool alwaysRebroadcast;
QueryScheduler? scheduler;
static final _oneOffOpId = '0';
int idCounter = 1;
/// [ObservableQuery] registry
Map<String, ObservableQuery> queries = <String, ObservableQuery>{};
/// prevents rebroadcasting for some intensive bulk operation like [refetchSafeQueries]
bool rebroadcastLocked = false;
ObservableQuery<TParsed> watchQuery<TParsed>(
WatchQueryOptions<TParsed> options) {
final ObservableQuery<TParsed> observableQuery = ObservableQuery<TParsed>(
queryManager: this,
options: options,
);
setQuery(observableQuery);
return observableQuery;
}
Stream<QueryResult<TParsed>> subscribe<TParsed>(
SubscriptionOptions<TParsed> options) async* {
assert(
options.fetchPolicy != FetchPolicy.cacheOnly,
"Cannot subscribe with FetchPolicy.cacheOnly: $options",
);
final request = options.asRequest;
// Add optimistic or cache-based result to the stream if any
if (options.optimisticResult != null) {
// TODO optimisticResults for streams just skip the cache for now
yield QueryResult.optimistic(
data: options.optimisticResult as Map<String, dynamic>?,
parserFn: options.parserFn,
);
} else if (shouldRespondEagerlyFromCache(options.fetchPolicy)) {
final cacheResult = cache.readQuery(
request,
optimistic: options.policies.mergeOptimisticData,
);
if (cacheResult != null) {
yield QueryResult(
source: QueryResultSource.cache,
data: cacheResult,
parserFn: options.parserFn,
);
}
}
try {
yield* link.queryKey(request).map((response) {
QueryResult<TParsed>? queryResult;
bool rereadFromCache = false;
try {
queryResult = mapFetchResultToQueryResult(
response,
options,
source: QueryResultSource.network,
);
rereadFromCache = attemptCacheWriteFromResponse(
options.policies,
request,
response,
queryResult,
);
} catch (failure, trace) {
// we set the source to indicate where the source of failure
queryResult ??= QueryResult(
source: QueryResultSource.network,
parserFn: options.parserFn,
);
queryResult.exception = coalesceErrors(
exception: queryResult.exception,
linkException: translateFailure(failure, trace),
);
}
if (rereadFromCache) {
// normalize results if previously written
attempCacheRereadIntoResult(request, queryResult);
}
return queryResult;
}).transform(StreamTransformer.fromHandlers(
handleError: (err, trace, sink) => sink.add(_wrapFailure(
err,
trace,
options.parserFn,
)),
));
} catch (ex, trace) {
yield* Stream.fromIterable([
_wrapFailure(
ex,
trace,
options.parserFn,
)
]);
}
}
Future<QueryResult<TParsed>> query<TParsed>(
QueryOptions<TParsed> options) async {
final result = await fetchQuery(_oneOffOpId, options);
maybeRebroadcastQueries();
return result;
}
Future<QueryResult<TParsed>> mutate<TParsed>(
MutationOptions<TParsed> options) async {
final result = await fetchQuery(_oneOffOpId, options);
// once the mutation has been process successfully, execute callbacks
// before returning the results
final mutationCallbacks = MutationCallbackHandler(
cache: cache,
options: options,
queryId: _oneOffOpId,
);
final callbacks = mutationCallbacks.callbacks;
for (final callback in callbacks) {
await callback(result);
}
/// wait until callbacks complete to rebroadcast
maybeRebroadcastQueries();
return result;
}
Future<QueryResult<TParsed>> fetchQuery<TParsed>(
String queryId,
BaseOptions<TParsed> options,
) async {
final MultiSourceResult<TParsed> allResults =
fetchQueryAsMultiSourceResult(queryId, options);
return allResults.networkResult ?? allResults.eagerResult;
}
/// Wrap both the `eagerResult` and `networkResult` future in a `MultiSourceResult`
/// if the cache policy precludes a network request, `networkResult` will be `null`
MultiSourceResult<TParsed> fetchQueryAsMultiSourceResult<TParsed>(
String queryId,
BaseOptions<TParsed> options,
) {
// create a new request to execute
final request = options.asRequest;
final QueryResult<TParsed> eagerResult = _resolveQueryEagerly(
request,
queryId,
options,
);
// _resolveQueryEagerly handles cacheOnly,
// so if we're loading + cacheFirst we continue to network
return MultiSourceResult(
parserFn: options.parserFn,
eagerResult: eagerResult,
networkResult:
(shouldStopAtCache(options.fetchPolicy) && !eagerResult.isLoading)
? null
: _resolveQueryOnNetwork(request, queryId, options),
);
}
/// Resolve the query on the network,
/// negotiating any necessary cache edits / optimistic cleanup
Future<QueryResult<TParsed>> _resolveQueryOnNetwork<TParsed>(
Request request,
String queryId,
BaseOptions<TParsed> options,
) async {
Response response;
QueryResult<TParsed>? queryResult;
bool rereadFromCache = false;
try {
// execute the request through the provided link(s)
response = await link.queryKey(request).first;
queryResult = mapFetchResultToQueryResult(
response,
options,
source: QueryResultSource.network,
);
rereadFromCache = attemptCacheWriteFromResponse(
options.policies,
request,
response,
queryResult,
);
} catch (failure, trace) {
// we set the source to indicate where the source of failure
queryResult ??= QueryResult(
source: QueryResultSource.network,
parserFn: options.parserFn,
);
queryResult.exception = coalesceErrors(
exception: queryResult.exception,
linkException: translateFailure(failure, trace),
);
}
// cleanup optimistic results
cache.removeOptimisticPatch(queryId);
if (rereadFromCache) {
// normalize results if previously written
attempCacheRereadIntoResult(request, queryResult);
}
// one off operations do not have an ObservableQuery to add to
if (queryId != _oneOffOpId) {
addQueryResult(request, queryId, queryResult);
}
return queryResult;
}
/// Add an eager cache response to the stream if possible,
/// based on `fetchPolicy` and `optimisticResults`
QueryResult<TParsed> _resolveQueryEagerly<TParsed>(
Request request,
String queryId,
BaseOptions<TParsed> options,
) {
QueryResult<TParsed> queryResult = QueryResult.loading(
parserFn: options.parserFn,
);
try {
if (options.optimisticResult != null) {
queryResult = _getOptimisticQueryResult(
request,
queryId: queryId,
optimisticResult: options.optimisticResult,
options: options,
);
}
// if we haven't already resolved results optimistically,
// we attempt to resolve the from the cache
if (shouldRespondEagerlyFromCache(options.fetchPolicy) &&
!queryResult.isOptimistic) {
final dynamic data = cache.readQuery(request, optimistic: false);
// we only push an eager query with data
if (data != null) {
queryResult = QueryResult(
data: data,
source: QueryResultSource.cache,
parserFn: options.parserFn,
);
}
if (options.fetchPolicy == FetchPolicy.cacheOnly &&
queryResult.isLoading) {
queryResult = QueryResult(
source: QueryResultSource.cache,
parserFn: options.parserFn,
exception: OperationException(
linkException: CacheMissException(
'Could not resolve the given request against the cache. (FetchPolicy.cacheOnly)',
request,
),
),
);
}
}
} catch (failure, trace) {
queryResult.exception = coalesceErrors(
exception: queryResult.exception,
linkException: translateFailure(failure, trace),
);
}
// If not a regular eager cache resolution,
// will either be loading, or optimistic.
//
// if there's an optimistic result, we add it regardless of fetchPolicy.
// This is undefined-ish behavior/edge case, but still better than just
// ignoring a provided optimisticResult.
// Would probably be better to add it ignoring the cache in such cases
//
// one off operations do not have an ObservableQuery to add to
if (queryId != _oneOffOpId) {
addQueryResult(request, queryId, queryResult);
}
return queryResult;
}
/// Refetch the [ObservableQuery] referenced by [queryId],
/// overriding any present non-network-only [FetchPolicy].
Future<QueryResult<TParsed>?> refetchQuery<TParsed>(String queryId) {
final WatchQueryOptions<TParsed> options =
queries[queryId]!.options.copy() as WatchQueryOptions<TParsed>;
if (!willAlwaysExecuteOnNetwork(options.fetchPolicy)) {
options.policies = options.policies.copyWith(
fetch: FetchPolicy.networkOnly,
);
}
// create a new request to execute
final request = options.asRequest;
return _resolveQueryOnNetwork(request, queryId, options);
}
@experimental
Future<List<QueryResult?>> refetchSafeQueries() async {
rebroadcastLocked = true;
final results = await Future.wait(
queries.values.where((q) => q.isRefetchSafe).map((q) => q.refetch()),
);
rebroadcastLocked = false;
maybeRebroadcastQueries();
return results;
}
ObservableQuery? getQuery(String? queryId) {
if (queries.containsKey(queryId)) {
return queries[queryId!];
}
return null;
}
/// Add a result to the [ObservableQuery] specified by `queryId`, if it exists.
///
/// Will [maybeRebroadcastQueries] from [ObservableQuery.addResult] if the [cache] has flagged the need to.
///
/// Queries are registered via [setQuery] and [watchQuery]
void addQueryResult<TParsed>(
Request request,
String? queryId,
QueryResult<TParsed> queryResult,
) {
final ObservableQuery<TParsed>? observableQuery =
getQuery(queryId) as ObservableQuery<TParsed>?;
if (observableQuery != null && !observableQuery.controller.isClosed) {
observableQuery.addResult(queryResult);
}
}
/// Create an optimstic result for the query specified by `queryId`, if it exists
QueryResult<TParsed> _getOptimisticQueryResult<TParsed>(
Request request, {
required String queryId,
required Object? optimisticResult,
required BaseOptions<TParsed> options,
}) {
QueryResult<TParsed> queryResult = QueryResult(
source: QueryResultSource.optimisticResult,
parserFn: options.parserFn,
);
attemptCacheWriteFromClient(
request,
optimisticResult as Map<String, dynamic>?,
queryResult,
writeQuery: (req, data) => cache.recordOptimisticTransaction(
(proxy) => proxy..writeQuery(req, data: data!),
queryId,
),
);
if (!queryResult.hasException) {
queryResult.data = cache.readQuery(
request,
optimistic: true,
);
}
return queryResult;
}
/// Rebroadcast cached queries with changed underlying data if [cache.broadcastRequested] or [force].
///
/// Push changed data from cache to query streams.
/// [exclude] is used to skip a query if it was recently executed
/// (normally the query that caused the rebroadcast)
///
/// Returns whether a broadcast was executed, which depends on the state of the cache.
/// If there are multiple in-flight cache updates, we wait until they all complete
///
/// **Note on internal implementation details**:
/// There is sometimes confusion on when this is called, but rebroadcasts are requested
/// from every [addQueryResult] where `result.isNotLoading` as an [OnData] callback from [ObservableQuery].
bool maybeRebroadcastQueries({ObservableQuery? exclude, bool force = false}) {
if (rebroadcastLocked && !force) {
return false;
}
final shouldBroadast = cache.shouldBroadcast(claimExecution: true);
if (!shouldBroadast && !force) {
return false;
}
for (ObservableQuery query in queries.values) {
if (query != exclude && query.isRebroadcastSafe) {
final cachedData = cache.readQuery(
query.options.asRequest,
optimistic: query.options.policies.mergeOptimisticData,
);
if (_cachedDataHasChangedFor(query, cachedData)) {
query.addResult(
mapFetchResultToQueryResult(
Response(data: cachedData),
query.options,
source: QueryResultSource.cache,
),
fromRebroadcast: true,
);
}
}
}
return true;
}
bool _cachedDataHasChangedFor(
ObservableQuery query,
Map<String, dynamic>? cachedData,
) =>
cachedData != null &&
(alwaysRebroadcast || !_deepEquals(query.latestResult!.data, cachedData));
void setQuery(ObservableQuery observableQuery) {
queries[observableQuery.queryId] = observableQuery;
}
void closeQuery(ObservableQuery observableQuery, {bool fromQuery = false}) {
if (!fromQuery) {
observableQuery.close(fromManager: true);
}
queries.remove(observableQuery.queryId);
}
int generateQueryId() {
final int requestId = idCounter;
idCounter++;
return requestId;
}
QueryResult<TParsed> mapFetchResultToQueryResult<TParsed>(
Response response,
BaseOptions<TParsed> options, {
required QueryResultSource source,
}) {
List<GraphQLError>? errors;
dynamic data;
// check if there are errors and apply the error policy if so
// in a nutshell: `ignore` swallows errors, `none` swallows data
if (response.errors != null && response.errors!.isNotEmpty) {
switch (options.errorPolicy) {
case ErrorPolicy.all:
// handle both errors and data
errors = response.errors;
data = response.data;
break;
case ErrorPolicy.ignore:
// ignore errors
data = response.data;
break;
case ErrorPolicy.none:
default:
// TODO not actually sure if apollo even casts graphql errors in `none` mode,
// it's also kind of legacy
errors = response.errors;
break;
}
} else {
data = response.data;
}
return QueryResult(
data: data,
context: response.context,
source: source,
exception: coalesceErrors(graphqlErrors: errors),
parserFn: options.parserFn,
);
}
}
QueryResult<TParsed> _wrapFailure<TParsed>(
dynamic ex,
trace,
ResultParserFn<TParsed> parserFn,
) =>
QueryResult(
// we set the source to indicate where the source of failure
source: QueryResultSource.network,
exception: coalesceErrors(linkException: translateFailure(ex, trace)),
parserFn: parserFn,
);
@@ -0,0 +1,660 @@
/// `TQueryData`, `TQueryFnData`, `TData` should be [Map]s for shallow/deep equality checks
/// Or these can be data classes that have `toJson` method & `fromJson`
/// constructor. This also requires the data-class to be passed to the
/// [Query] constructor parameters e.g ([dataType])
import 'dart:async';
import 'package:fl_query/src/core/models.dart';
import 'package:fl_query/src/core/notify_manager.dart';
import 'package:fl_query/src/core/query.dart';
import 'package:fl_query/src/core/query_cache.dart';
import 'package:fl_query/src/core/query_client.dart';
import 'package:fl_query/src/core/retryer.dart';
import 'package:fl_query/src/core/subscribable.dart';
import 'package:fl_query/src/core/utils.dart';
import 'package:meta/meta.dart';
typedef QueryObserverListener<TData, TError> = void Function(
QueryObserverResult<TData, TError> result);
class NotifyOptions {
bool? cache;
bool? listeners;
bool? onError;
bool? onSuccess;
NotifyOptions({this.cache, this.listeners, this.onError, this.onSuccess});
Map<String, dynamic> toJson() {
final Map<String, dynamic> data = new Map<String, dynamic>();
data['cache'] = this.cache;
data['listeners'] = this.listeners;
data['onError'] = this.onError;
data['onSuccess'] = this.onSuccess;
return data;
}
NotifyOptions.fromJson(Map<String, dynamic> json) {
cache = json['cache'];
listeners = json['listeners'];
onError = json['onError'];
onSuccess = json['onSuccess'];
}
}
class ObserverFetchOptions extends FetchOptions {
bool? throwOnError;
ObserverFetchOptions({
this.throwOnError,
bool? cancelRefetch,
dynamic meta,
}) : super(cancelRefetch: cancelRefetch, meta: meta);
}
class SelectQuery<TQueryData, TData> {
TData Function(TQueryData data) fn;
TData result;
SelectQuery(this.fn, this.result);
}
class QueryObserver<TQueryFnData, TError, TData, TQueryData>
extends Subscribable<QueryObserverListener> {
late QueryObserverOptions<TQueryFnData, TError, TData, TQueryData> options;
QueryClient _client;
Query<TQueryFnData, TError, TQueryData>? _currentQuery;
late QueryState<TQueryData, TError> _currentQueryInitialState;
late QueryObserverResult<TData, TError> _currentResult;
QueryState<TQueryData, TError>? _currentResultState;
QueryObserverOptions<TQueryFnData, TError, TData, TQueryData>?
_currentResultOptions;
QueryObserverResult<TData, TError>? _previousQueryResult;
Exception? _previousSelectError;
SelectQuery? _previousSelect;
Timer? _staleTimeout;
Timer? _refetchInterval;
Duration? _currentRefetchInterval;
/// List of tracked keys/properties of [QueryObserverResult]
late List<String> _trackedProps;
QueryObserver(this._client, options)
: _trackedProps = [],
_previousSelectError = null {
this.setOptions(options);
}
bool shouldFetchCurrentQueryOnReconnect() {
return shouldFetchOnReconnect(_currentQuery!, this.options);
}
@override
void onSubscribe() {
if (listeners.length == 1) {
_currentQuery?.addObserver(this);
if (_currentQuery != null &&
shouldFetchOnMount(_currentQuery!, options)) {
_executeFetch();
}
_updateTimers();
}
}
@override
void onUnsubscribe() {
if (listeners.isEmpty) {
this.destroy();
}
}
void destroy() {
listeners = [];
_clearTimers();
_currentQuery?.removeObserver(this);
}
setOptions(
QueryObserverOptions<TQueryFnData, TError, TData, TQueryData>? options, [
NotifyOptions? notifyOptions,
]) {
var prevOptions = this.options;
var prevQuery = _currentQuery;
this.options = this._client.defaultQueryObserverOptions(options);
this.options.queryKey ??= prevOptions.queryKey;
_updateQuery();
bool mounted = hasListeners();
if (mounted &&
_currentQuery != null &&
prevQuery != null &&
shouldFetchOptionally(
_currentQuery!, prevQuery, this.options, prevOptions)) {
_executeFetch();
}
;
}
QueryObserverResult<TData, TError> getOptimisticResult(
QueryObserverOptions<TQueryFnData, TError, TData, TQueryData> options,
) {
final defaultedOptions = _client.defaultQueryObserverOptions(options);
final query = _client.getQueryCache().build(_client, defaultedOptions);
return createResult(query, defaultedOptions);
}
QueryObserverResult<TData, TError> getCurrentResult() {
return _currentResult;
}
/// There's nothing similar to JS [defineProperty] in dart native
/// objects thus modifying the underlying property `get` method is
/// impossible so [trackProp] can't be implemented at the moment
/// At least not following this procedure
QueryObserverResult<TData, TError> trackResult(
QueryObserverResult<TData, TError> result,
QueryObserverOptions<TQueryFnData, TError, TData, TQueryData>
defaultedOptions,
) {
// final Map<String, dynamic> trackedResult = <String, dynamic>{};
// const trackProp = (key: keyof QueryObserverResult) => {
// if (!this.trackedProps.includes(key)) {
// this.trackedProps.push(key)
// }
// }
// Object.keys(result).forEach(key => {
// Object.defineProperty(trackedResult, key, {
// configurable: false,
// enumerable: true,
// get: () => {
// trackProp(key as keyof QueryObserverResult)
// return result[key as keyof QueryObserverResult]
// },
// })
// })
// if (defaultedOptions.useErrorBoundary || defaultedOptions.suspense) {
// trackProp('error')
// }
// return trackedResult
throw UnimplementedError("COULD NOT IMPLEMENT DUE TO LANGUAGE LIMITATIONS");
}
Future<QueryObserverResult<TData, TError>> getNextResult(
ResultOptions? options) {
var completer = Completer<QueryObserverResult<TData, TError>>();
var unsubscribe;
unsubscribe = subscribe((result) {
if (!result.isFetching) {
unsubscribe?.call();
if (result.isError && options?.throwOnError == true) {
if (!completer.isCompleted)
completer.completeError(result.error as Object);
} else {
if (!completer.isCompleted)
completer.complete(
result as QueryObserverResult<TData, TError>,
);
}
}
});
return completer.future;
}
Query<TQueryFnData, TError, TQueryData> getCurrentQuery() {
return _currentQuery!;
}
Future<QueryObserverResult<TData, TError>> fetchOptimistic(
QueryObserverOptions<TQueryFnData, TError, TData, TQueryData> options) {
var defaultedOptions = _client.defaultQueryObserverOptions(options);
var query = _client.getQueryCache().build(_client, defaultedOptions);
return query.fetch().then((val) {
return createResult(query, defaultedOptions);
});
}
@protected
Future<QueryObserverResult<TData, TError>> fetch(
ObserverFetchOptions fetchOptions,
) {
return _executeFetch(fetchOptions).then((val) {
updateResult();
return _currentResult;
});
}
Future<TQueryData?> _executeFetch([ObserverFetchOptions? fetchOptions]) {
// Make sure we reference the latest query as the current one might have been removed
_updateQuery();
// Fetch
Future<TQueryData?> future = _currentQuery!.fetch(
this.options,
fetchOptions,
);
if (fetchOptions?.throwOnError != null) {
future = future.catchError((e) => e);
}
return future;
}
bool _shouldNotifyListeners(QueryObserverResult<TData, TError> result,
[QueryObserverResult<TData, TError>? prevResult]) {
if (prevResult == null) return true;
if (!options.notifyOnChangeProps &&
options.notifyOnChangePropsExclusions == null) {
return true;
}
if (options.notifyOnChangeProps == 'tracked' && _trackedProps.isEmpty) {
return true;
}
List<String>? includedProps = options.notifyOnChangeProps == 'tracked'
? _trackedProps
: options.notifyOnChangeProps;
Map<String, dynamic> resultMap = result.toJson();
Map<String, dynamic> prevResultMap = prevResult.toJson();
return resultMap.keys.any((key) {
var changed = resultMap[key] != prevResultMap[key];
bool? isIncluded = includedProps?.any((x) => x == key);
bool isExcluded =
options.notifyOnChangePropsExclusions?.any((x) => x == key) ?? false;
return changed &&
!isExcluded &&
(includedProps == null || isIncluded == true);
});
}
void updateResult([NotifyOptions? notifyOptions]) {
QueryObserverResult<TData, TError>? prevResult = _currentResult;
if (_currentQuery != null)
_currentResult = this.createResult(_currentQuery!, this.options);
_currentResultState = _currentQuery?.state;
_currentResultOptions = this.options;
// Only notify if something has changed
if (shallowEqualMap(_currentResult.toJson(), prevResult.toJson())) {
return;
}
NotifyOptions defaultNotifyOptions = NotifyOptions(cache: true);
if (notifyOptions?.listeners != false &&
_shouldNotifyListeners(_currentResult, prevResult)) {
defaultNotifyOptions.listeners = true;
}
_notify(NotifyOptions.fromJson({
...defaultNotifyOptions.toJson(),
...(notifyOptions?.toJson() ?? {}),
}));
}
void _updateQuery() {
var query = this._client.getQueryCache().build(this._client, this.options);
if (query == _currentQuery) return;
var prevQuery = _currentQuery;
_currentQuery = query;
_currentQueryInitialState = query.state;
_previousQueryResult = _currentResult;
if (hasListeners()) {
prevQuery?.removeObserver(this);
query.addObserver(this);
}
}
void onQueryUpdate(Action<TData, TError> action) {
final NotifyOptions notifyOptions = NotifyOptions();
if (action.type == 'success') {
notifyOptions.onSuccess = true;
} else if (action.type == 'error' && !isCancelledError(action.error)) {
notifyOptions.onError = true;
}
updateResult(notifyOptions);
if (this.hasListeners()) {
_updateTimers();
}
}
QueryObserverResult<TData, TError> createResult(
Query<TQueryFnData, TError, TQueryData> query,
QueryObserverOptions<TQueryFnData, TError, TData, TQueryData> options,
) {
var prevQuery = _currentQuery;
var prevOptions = this.options;
var prevResult = _currentResult;
var prevResultState = _currentResultState;
var prevResultOptions = _currentResultOptions;
bool queryChange = query != prevQuery;
var queryInitialState =
queryChange ? query.state : _currentQueryInitialState;
var prevQueryResult = queryChange ? _currentResult : _previousQueryResult;
var state = query.state;
var dataUpdatedAt = state.dataUpdatedAt;
var error = state.error;
var errorUpdatedAt = state.errorUpdatedAt;
var isFetching = state.isFetching;
var status = state.status;
bool isPreviousData = false;
bool isPlaceholderData = false;
TData? data;
// Optimistically set result in fetching state if needed
if (options.optimisticResults == true) {
var mounted = hasListeners();
var fetchOnMount = !mounted && shouldFetchOnMount(query, options);
bool fetchOptionally = mounted &&
prevQuery != null &&
shouldFetchOptionally(query, prevQuery, options, prevOptions);
if (fetchOnMount || fetchOptionally) {
isFetching = true;
if (dataUpdatedAt == null) {
status = QueryStatus.loading;
}
}
}
// Keep previous data if needed
if (prevQueryResult != null &&
options.keepPreviousData == true &&
state.dataUpdateCount == 0 &&
prevQueryResult.isSuccess == true &&
status != QueryStatus.error) {
data = prevQueryResult.data;
dataUpdatedAt = prevQueryResult.dataUpdatedAt;
status = prevQueryResult.status;
isPreviousData = true;
}
// Select data if needed
else if (options.select != null && state.data != null) {
if (prevResult != null &&
state.data == prevResultState?.data &&
options.select == _previousSelect?.fn &&
_previousSelectError == null) {
data = _previousSelect?.result;
} else {
try {
data = options.select?.call(state.data);
if (options.structuralSharing != false) {
data = replaceEqualDeep(prevResult?.data, data);
}
if (options.select != null && data != null) {
_previousSelect = SelectQuery<TQueryData, TData>(
options.select!,
data,
);
}
_previousSelectError = null;
} catch (selectError) {
// getLogger().error(selectError);
error = selectError as TError;
_previousSelectError = selectError as Exception;
errorUpdatedAt = DateTime.now();
status = QueryStatus.error;
}
}
}
// Use query data
else {
data = state.data as TData;
}
if (options.placeholderData != null &&
data == null &&
(status == QueryStatus.loading || status == QueryStatus.idle)) {
var placeholderData;
if (prevResult?.isPlaceholderData == true &&
options.placeholderData == prevResultOptions?.placeholderData) {
placeholderData = prevResult.data;
} else {
placeholderData = options.placeholderData;
if (options.select != null && placeholderData != null) {
try {
placeholderData = options.select?.call(placeholderData);
if (options.structuralSharing != false) {
placeholderData =
replaceEqualDeep(prevResult?.data, placeholderData);
}
_previousSelectError = null;
} catch (selectError) {
// getLogger().error(selectError);
error = selectError as TError;
_previousSelectError = selectError as Exception;
errorUpdatedAt = DateTime.now();
status = QueryStatus.error;
}
}
}
if (placeholderData != null) {
status = QueryStatus.success;
data = placeholderData as TData;
isPlaceholderData = true;
}
}
QueryObserverResult<TData, TError> result = QueryObserverResult(
status: status,
dataUpdatedAt: dataUpdatedAt,
isLoading: status == QueryStatus.loading,
isSuccess: status == QueryStatus.success,
isError: status == QueryStatus.error,
isIdle: status == QueryStatus.idle,
data: data,
error: error,
failureCount: state.fetchFailureCount,
isFetched: state.dataUpdateCount > 0 || state.errorUpdateCount > 0,
isFetchedAfterMount:
state.dataUpdateCount > queryInitialState.dataUpdateCount ||
state.errorUpdateCount > queryInitialState.errorUpdateCount,
isFetching: isFetching,
isRefetching: isFetching && status != QueryStatus.loading,
isLoadingError:
status == QueryStatus.error && state.dataUpdatedAt == null,
isPlaceholderData: isPlaceholderData,
isPreviousData: isPreviousData,
isRefetchError: status == 'error' && state.dataUpdatedAt != 0,
isStale: isStale(query, options),
refetch: this.refetch,
remove: this.remove,
);
return result;
}
void _notify(NotifyOptions notifyOptions) {
notifyManager.batch(() {
// First trigger the configuration callbacks
if (notifyOptions.onSuccess == true) {
this.options.onSuccess?.call(_currentResult.data!);
this.options.onSettled?.call(_currentResult.data!);
} else if (notifyOptions.onError == true) {
this.options.onError?.call(_currentResult.error!);
this.options.onSettled?.call(null, _currentResult.error!);
}
// Then trigger the listeners
if (notifyOptions.listeners == true) {
this.listeners.forEach((listener) {
listener(_currentResult);
});
}
// Then the cache listeners
if (notifyOptions.cache == true && _currentQuery != null) {
_client.getQueryCache().notify(
QueryCacheNotifyEvent(
QueryCacheNotifyEventType.observerResultsUpdated,
_currentQuery as Query,
),
);
}
});
}
Duration? _computeRefetchInterval() {
return this.options.refetchInterval != null && _currentQuery != null
? this.options.refetchInterval!(_currentResult.data, _currentQuery!)
: null;
}
void _updateTimers() {
_updateStaleTimeout();
_updateRefetchInterval(_computeRefetchInterval());
}
void _updateStaleTimeout() {
_clearStaleTimeout();
if (_currentResult.isStale ||
options.staleTime == null ||
_currentResult.dataUpdatedAt == null) return;
// The timeout is sometimes triggered 1 ms before the stale time
// expiration. To mitigate this issue we always add 1 ms to the
// timeout.
Duration time = Duration(
milliseconds:
timeUntilStale(_currentResult.dataUpdatedAt!, this.options.staleTime)
.inMilliseconds +
1,
);
_staleTimeout = Timer(time, () {
if (!_currentResult.isStale) {
this.updateResult();
}
});
}
_updateRefetchInterval(Duration? nextInterval) {
_clearRefetchInterval();
_currentRefetchInterval = nextInterval;
if (this.options.enabled == false ||
_currentRefetchInterval == null ||
_currentRefetchInterval == Duration.zero) return;
_refetchInterval = Timer.periodic(_currentRefetchInterval!, (t) {
if (this.options.refetchIntervalInBackground == true) {
_executeFetch();
}
});
}
void _clearTimers() {
_clearStaleTimeout();
_clearRefetchInterval();
}
void _clearStaleTimeout() {
_staleTimeout?.cancel();
_staleTimeout = null;
}
void _clearRefetchInterval() {
_refetchInterval?.cancel();
_refetchInterval = null;
}
void remove() {
_client.getQueryCache().remove(_currentQuery as Query);
_clearTimers();
_currentQuery?.removeObserver(this);
}
Future<QueryObserverResult<TData, TError>> refetch<TPageData>({
RefetchQueryFilters<TPageData>? filters,
RefetchOptions? options,
}) {
return fetch(
ObserverFetchOptions(
cancelRefetch: options?.cancelRefetch,
meta: filters?.toJson(),
throwOnError: options?.throwOnError,
),
);
}
}
bool shouldLoadOnMount<TQueryFnData, TError, TData, TQueryData>(
Query<TQueryFnData, TError, TQueryData> query,
QueryObserverOptions<TQueryFnData, TError, TData, TQueryData> options,
) {
return (options.enabled != false &&
query.state.dataUpdatedAt != null &&
!(query.state.status == QueryStatus.error &&
options.retryOnMount == false));
}
bool shouldRefetchOnMount<TQueryFnData, TError, TData, TQueryData>(
Query<TQueryFnData, TError, TQueryData> query,
QueryObserverOptions<TQueryFnData, TError, TData, TQueryData> options,
) {
return (options.enabled != false &&
query.state.dataUpdatedAt != null &&
(options.refetchOnMount == RefetchOnMount.always ||
(options.refetchOnMount != RefetchOnMount.off &&
isStale(query, options))));
}
bool shouldFetchOnMount<TQueryFnData, TError, TData, TQueryData>(
Query<TQueryFnData, TError, TQueryData> query,
QueryObserverOptions<TQueryFnData, TError, TData, TQueryData> options,
) {
return (shouldLoadOnMount(query, options) ||
shouldRefetchOnMount(query, options));
}
bool shouldFetchOnReconnect<TQueryFnData, TError, TData, TQueryData>(
Query<TQueryFnData, TError, TQueryData> query,
QueryObserverOptions<TQueryFnData, TError, TData, TQueryData> options,
) {
return (options.enabled != false &&
(options.refetchOnReconnect == RefetchOnReconnect.always ||
(options.refetchOnReconnect != RefetchOnReconnect.off &&
isStale(query, options))));
}
bool shouldFetchOptionally<TQueryFnData, TError, TData, TQueryData>(
Query<TQueryFnData, TError, TQueryData> query,
Query<TQueryFnData, TError, TQueryData> prevQuery,
QueryObserverOptions<TQueryFnData, TError, TData, TQueryData> options,
QueryObserverOptions<TQueryFnData, TError, TData, TQueryData> prevOptions,
) {
return (options.enabled != false &&
(query != prevQuery || prevOptions.enabled == false) &&
(options.suspense != true || query.state.status != QueryStatus.error) &&
isStale(query, options));
}
bool isStale<TQueryFnData, TError, TData, TQueryData>(
Query<TQueryFnData, TError, TQueryData> query,
QueryObserverOptions<TQueryFnData, TError, TData, TQueryData> options,
) {
return query.isStaleByTime(options.staleTime);
}
@@ -1,191 +0,0 @@
// ignore_for_file: deprecated_member_use_from_same_package
import 'package:fl_query/src/core/_base_options.dart';
import 'package:fl_query/src/core/result_parser.dart';
import 'package:fl_query/src/utilities/helpers.dart';
import 'package:fl_query/fl_query.dart';
/// Query options.
class QueryOptions<TParsed> extends BaseOptions<TParsed> {
QueryOptions({
required DocumentNode document,
String? operationName,
Map<String, dynamic> variables = const {},
FetchPolicy? fetchPolicy,
ErrorPolicy? errorPolicy,
CacheRereadPolicy? cacheRereadPolicy,
Object? optimisticResult,
this.pollInterval,
Context? context,
ResultParserFn<TParsed>? parserFn,
}) : super(
fetchPolicy: fetchPolicy,
errorPolicy: errorPolicy,
cacheRereadPolicy: cacheRereadPolicy,
document: document,
operationName: operationName,
variables: variables,
context: context,
optimisticResult: optimisticResult,
parserFn: parserFn,
);
/// The time interval on which this query should be re-fetched from the server.
Duration? pollInterval;
@override
List<Object?> get properties => [...super.properties, pollInterval];
WatchQueryOptions<TParsed> asWatchQueryOptions({bool fetchResults = true}) =>
WatchQueryOptions(
document: document,
operationName: operationName,
variables: variables,
fetchPolicy: fetchPolicy,
errorPolicy: errorPolicy,
cacheRereadPolicy: cacheRereadPolicy,
pollInterval: pollInterval,
fetchResults: fetchResults,
context: context,
optimisticResult: optimisticResult,
parserFn: this.parserFn,
);
}
class SubscriptionOptions<TParsed> extends BaseOptions<TParsed> {
SubscriptionOptions({
required DocumentNode document,
String? operationName,
Map<String, dynamic> variables = const {},
FetchPolicy? fetchPolicy,
ErrorPolicy? errorPolicy,
CacheRereadPolicy? cacheRereadPolicy,
Object? optimisticResult,
Context? context,
ResultParserFn<TParsed>? parserFn,
}) : super(
fetchPolicy: fetchPolicy,
errorPolicy: errorPolicy,
cacheRereadPolicy: cacheRereadPolicy,
document: document,
operationName: operationName,
variables: variables,
context: context,
optimisticResult: optimisticResult,
parserFn: parserFn,
);
/// An optimistic first result to eagerly add to the subscription stream
Object? optimisticResult;
}
class WatchQueryOptions<TParsed> extends QueryOptions<TParsed> {
WatchQueryOptions({
required DocumentNode document,
String? operationName,
Map<String, dynamic> variables = const {},
FetchPolicy? fetchPolicy,
ErrorPolicy? errorPolicy,
CacheRereadPolicy? cacheRereadPolicy,
Object? optimisticResult,
Duration? pollInterval,
this.fetchResults = false,
this.carryForwardDataOnException = true,
bool? eagerlyFetchResults,
Context? context,
ResultParserFn<TParsed>? parserFn,
}) : eagerlyFetchResults = eagerlyFetchResults ?? fetchResults,
super(
document: document,
operationName: operationName,
variables: variables,
fetchPolicy: fetchPolicy,
errorPolicy: errorPolicy,
cacheRereadPolicy: cacheRereadPolicy,
pollInterval: pollInterval,
context: context,
optimisticResult: optimisticResult,
parserFn: parserFn,
);
/// Whether or not to fetch results
bool fetchResults;
/// Whether to [fetchResults] immediately on instantiation.
/// Defaults to [fetchResults].
bool eagerlyFetchResults;
/// carry forward previous data in the result of errors and no data.
/// defaults to `true`.
bool carryForwardDataOnException;
@override
List<Object?> get properties =>
[...super.properties, fetchResults, eagerlyFetchResults];
WatchQueryOptions<TParsed> copy() => WatchQueryOptions<TParsed>(
document: document,
operationName: operationName,
variables: variables,
fetchPolicy: fetchPolicy,
errorPolicy: errorPolicy,
cacheRereadPolicy: cacheRereadPolicy,
optimisticResult: optimisticResult,
pollInterval: pollInterval,
fetchResults: fetchResults,
eagerlyFetchResults: eagerlyFetchResults,
carryForwardDataOnException: carryForwardDataOnException,
context: context,
parserFn: parserFn,
);
}
/// options for fetchMore operations
///
/// **NOTE**: with the addition of strict data structure checking in v4,
/// it is easy to make mistakes in writing [updateQuery].
///
/// To mitigate this, [FetchMoreOptions.partial] has been provided.
class FetchMoreOptions {
FetchMoreOptions({
this.document,
this.variables = const {},
required this.updateQuery,
});
/// Automatically merge the results of [updateQuery] into `previousResultData`.
///
/// This is useful if you only want to, say, extract some list data
/// from the newly fetched result, and don't want to worry about
/// structural inconsistencies while merging.
static FetchMoreOptions partial({
DocumentNode? document,
Map<String, dynamic> variables = const {},
required UpdateQuery updateQuery,
}) =>
FetchMoreOptions(
document: document,
variables: variables,
updateQuery: partialUpdater(updateQuery),
);
DocumentNode? document;
Map<String, dynamic> variables;
/// Strategy for merging the fetchMore result data
/// with the result data already in the cache
UpdateQuery updateQuery;
/// Wrap an [UpdateQuery] in a [deeplyMergeLeft] of the `previousResultData`.
static UpdateQuery partialUpdater(UpdateQuery update) =>
(previous, fetched) => deeplyMergeLeft(
[previous, update(previous, fetched)],
);
}
/// merge fetchMore result data with earlier result data
typedef Map<String, dynamic>? UpdateQuery(
Map<String, dynamic>? previousResultData,
Map<String, dynamic>? fetchMoreResultData,
);
@@ -1,155 +0,0 @@
import 'dart:async' show FutureOr;
import 'package:fl_query/fl_query.dart';
import 'package:fl_query/src/core/result_parser.dart';
/// The source of the result data contained
///
/// * [loading]: No data has been specified from any source
/// for the _most recent_ operation
/// * [cache]: A result has been eagerly resolved from the cache
/// * [optimisticResult]: An optimistic result has been specified
/// May include eager results from the cache.
/// * [network]: The query has been resolved on the network
///
/// Both [optimisticResult] and [cache] sources are considered "Eager" results.
enum QueryResultSource {
/// No data has been specified from any source for the _most recent_ operation
loading,
/// A result has been eagerly resolved from the cache
cache,
/// An optimistic result has been specified.
/// May include eager results from the cache
optimisticResult,
/// The query has been resolved on the network
network,
}
extension Getters on QueryResultSource {
/// Whether this result source is considered "eager" (is [cache] or [optimisticResult])
bool get isEager => _eagerSources.contains(this);
}
final _eagerSources = {
QueryResultSource.cache,
QueryResultSource.optimisticResult
};
/// A single operation result
class QueryResult<TParsed> {
QueryResult({
this.data,
this.exception,
this.context = const Context(),
required this.parserFn,
required this.source,
}) : timestamp = DateTime.now();
/// Unexecuted singleton, used as a placeholder for mutations,
/// etc.
static final unexecuted = QueryResult(
source: null,
parserFn: (d) =>
throw UnimplementedError("Unexecuted query data can not be parsed."),
)..timestamp = DateTime.fromMillisecondsSinceEpoch(0);
factory QueryResult.loading({
Map<String, dynamic>? data,
required ResultParserFn<TParsed> parserFn,
}) =>
QueryResult(
data: data,
source: QueryResultSource.loading,
parserFn: parserFn,
);
factory QueryResult.optimistic({
Map<String, dynamic>? data,
required ResultParserFn<TParsed> parserFn,
}) =>
QueryResult(
data: data,
source: QueryResultSource.optimisticResult,
parserFn: parserFn,
);
DateTime timestamp;
/// The source of the result data.
///
/// `null` when unexecuted.
/// Will be set when encountering an error during any execution attempt
QueryResultSource? source;
/// Response data
Map<String, dynamic>? data;
/// Response context. Defaults to an empty `Context()`
Context context;
OperationException? exception;
ResultParserFn<TParsed> parserFn;
/// [data] has yet to be specified from any source
/// for the _most recent_ operation
/// (including [QueryResultSource.optimisticResult])
///
/// **NOTE:** query updating methods like `fetchMore` and `refetch` will send
/// an [isLoading], so it is best practice to check both `isLoading && data != null`
/// before assuming there is no data that should be displayed.
bool get isLoading => source == QueryResultSource.loading;
/// [data] been specified (including [QueryResultSource.optimisticResult])
bool get isNotLoading => !isLoading;
/// [data] has been specified as an [QueryResultSource.optimisticResult]
///
/// May include eager results from the cache.
bool get isOptimistic => source == QueryResultSource.optimisticResult;
/// [data] has been specified and is **not** an [QueryResultSource.optimisticResult]
///
/// shorthand for `!isLoading && !isOptimistic`
bool get isConcrete => !isLoading && !isOptimistic;
/// Whether the response includes an [exception]
bool get hasException => (exception != null);
/// If a parserFn is provided, this getter can be used to fetch the parsed data.
TParsed? get parsedData {
final data = this.data;
final parserFn = this.parserFn;
if (data == null) {
return null;
}
return parserFn(data);
}
@override
String toString() => 'QueryResult('
'source: $source, '
'data: $data, '
'context: $context, '
'exception: $exception, '
'timestamp: $timestamp'
')';
}
class MultiSourceResult<TParsed> {
MultiSourceResult({
QueryResult<TParsed>? eagerResult,
this.networkResult,
required ResultParserFn<TParsed> parserFn,
}) : eagerResult = eagerResult ?? QueryResult.loading(parserFn: parserFn),
assert(
eagerResult!.source != QueryResultSource.network,
'An eager result cannot be gotten from the network',
);
QueryResult<TParsed> eagerResult;
FutureOr<QueryResult<TParsed>>? networkResult;
}
@@ -1 +0,0 @@
typedef ResultParserFn<TResult> = TResult Function(Map<String, dynamic> data);
+196
View File
@@ -0,0 +1,196 @@
import 'dart:async';
import 'dart:math' show pow, min;
import 'package:fl_query/src/core/online_manager.dart';
typedef ShouldRetryFunction<TError> = int Function(
int failureCount,
TError error,
);
typedef RetryDelayFunction<TError> = double Function(
int failureCount,
TError error,
);
double defaultRetryDelay(int failureCount) {
return min(pow(1000 * 2, failureCount), 30000).toDouble();
}
abstract class Cancelable {
void cancel();
}
bool isCancelable(value) {
return value is Cancelable;
}
class CancelledError {
bool? revert;
bool? silent;
CancelledError({this.revert, this.silent});
}
bool isCancelledError(value) {
return value is CancelledError;
}
typedef OnError<TError> = void Function(TError error);
typedef OnData<TData> = void Function(TData data);
class Retryer<TData, TError> {
late void Function({bool? revert, bool? silent}) cancel;
late void Function() cancelRetry;
late void Function() continueRetry;
late void Function() continueFn;
late Future<TData> future;
int failureCount;
bool isPaused;
bool isResolved;
bool isTransportCancelable;
// config options for the retryer
FutureOr<TData> Function() fn;
void Function()? _abort;
OnError<TError>? onError;
OnData<TData>? onSuccess;
void Function(int failureCount, TError error)? onFail;
void Function()? onPause;
void Function()? onContinue;
ShouldRetryFunction<TError>? retry;
RetryDelayFunction<TError>? retryDelay;
Retryer({
required this.fn,
void Function()? abort,
this.onError,
this.onSuccess,
this.onFail,
this.onPause,
this.onContinue,
this.retry,
this.retryDelay,
}) : _abort = abort,
failureCount = 0,
isPaused = false,
isResolved = false,
isTransportCancelable = false {
bool cancelRetry = false;
void Function({bool? revert, bool? silent})? cancelFn;
void Function([dynamic value])? continueFn;
cancel = ({bool? revert, bool? silent}) {
cancelFn?.call();
};
this.cancelRetry = () {
cancelRetry = true;
};
this.continueRetry = () {
cancelRetry = false;
};
this.continueFn = () => continueFn?.call();
Completer<TData> completer = Completer<TData>();
this.future = completer.future;
resolve(value) {
if (!this.isResolved) {
this.isResolved = true;
onSuccess?.call(value);
continueFn?.call();
if (!completer.isCompleted) completer.complete(value);
}
}
reject(value) {
if (!this.isResolved) {
this.isResolved = true;
onError?.call(value);
continueFn?.call();
if (!completer.isCompleted) completer.completeError(value);
}
}
pause() {
Completer completer = Completer();
if (!completer.isCompleted) continueFn = completer.complete;
this.isPaused = true;
onPause?.call();
return completer.future.then((val) {
continueFn = null;
this.isPaused = false;
onContinue?.call();
});
}
run() {
// Do nothing if already resolved
if (this.isResolved) {
return;
}
var promiseOrValue;
// Execute query
try {
promiseOrValue = fn();
} catch (error) {
promiseOrValue = Future.error(error);
}
// Create callback to cancel this fetch
cancelFn = ({bool? revert, bool? silent}) {
if (!this.isResolved) {
reject(new CancelledError(revert: revert, silent: silent));
abort?.call();
// Cancel transport if supported
if (isCancelable(promiseOrValue)) {
try {
promiseOrValue.cancel();
} catch (error) {}
}
}
};
// Check if the transport layer support cancellation
this.isTransportCancelable = isCancelable(promiseOrValue);
Future.value(promiseOrValue).then(resolve).catchError((error) {
// Stop if the fetch is already resolved
if (this.isResolved) return;
// Do we need to retry the request?
int _retry = retry?.call(failureCount, error) ?? 3;
double _retryDelay = retryDelay?.call(failureCount, error) ??
defaultRetryDelay(failureCount);
bool shouldRetry = _retry > 0 && _retry > failureCount;
if (cancelRetry || !shouldRetry) {
// We are done if the query does not need to be retried
reject(error);
return;
}
this.failureCount++;
// Notify on fail
onFail?.call(this.failureCount, error);
Future.delayed(Duration(milliseconds: _retryDelay.toInt()))
.then((val) async {
if (!await onlineManager.isOnline()) {
return pause();
}
}).then((val) {
if (cancelRetry) {
reject(error);
} else {
run();
}
});
});
}
// Start loop
run();
}
}
@@ -0,0 +1,30 @@
import 'package:flutter/material.dart';
abstract class Subscribable<TListener extends Function> {
@protected
List<TListener> listeners;
Subscribable() : listeners = [];
void Function() subscribe(TListener? listener) {
var callback = listener ?? (() => null);
listeners.add(callback as TListener);
onSubscribe();
return () {
listeners = listeners.where((x) => x != callback).toList();
onUnsubscribe();
};
}
bool hasListeners() {
return listeners.isNotEmpty;
}
@protected
void onSubscribe();
@protected
void onUnsubscribe();
}
+148
View File
@@ -0,0 +1,148 @@
import 'package:fl_query/src/core/models.dart';
import 'package:fl_query/src/core/query.dart';
import 'package:fl_query/src/core/query_key.dart';
/**
* Schedules a microtask.
* This can be useful to schedule state updates after rendering.
*/
void scheduleMicrotask(void Function(dynamic val) callback) {
Future.value()
.then(callback)
.catchError((error) => Future.delayed(Duration.zero, () => throw error));
}
/// Default query keys hash function.
/// Dummy function just to fill the gaps for original react-query like
/// function body signatures
/// It is not required as a Standardized [QueryKey] data-class is used to
/// create the queryKey
String hashQueryKeyByOptions(
QueryKey queryKey,
QueryOptions? options,
) {
return options?.queryKeyHashFn?.call(queryKey) ?? queryKey.key;
}
enum QueryStatusFilter {
all,
active,
inactive,
none,
}
QueryStatusFilter mapQueryStatusFilter(
bool? active,
bool? inactive,
) {
if ((active == true && inactive == true) ||
(active == null && inactive == null)) {
return QueryStatusFilter.all;
} else if (active == false && inactive == false) {
return QueryStatusFilter.none;
} else {
// At this point, active|inactive can only be true|false or false|true
// so, when only one value is provided, the missing one has to be the negated value
bool isActive = active ?? !(inactive ?? false);
return isActive ? QueryStatusFilter.active : QueryStatusFilter.inactive;
}
}
bool matchQuery(QueryFilters filters, Query query, [QueryKey? queryKey]) {
if (queryKey != null) {
if (filters.exact! &&
query.queryHash != hashQueryKeyByOptions(queryKey, query.options))
return false;
else if (query.queryKey.key != queryKey) return false;
}
QueryStatusFilter queryStatusFilter =
mapQueryStatusFilter(filters.active, filters.inactive);
if (queryStatusFilter == QueryStatusFilter.none) {
return false;
} else if (queryStatusFilter != QueryStatusFilter.all) {
bool isActive = query.isActive();
if (queryStatusFilter == QueryStatusFilter.active && !isActive) {
return false;
}
if (queryStatusFilter == QueryStatusFilter.inactive && isActive) {
return false;
}
}
if (filters.stale != null && query.isStale() != filters.stale) {
return false;
}
if (filters.fetching != null && query.isFetching() != filters.fetching) {
return false;
}
if (filters.predicate != null && !filters.predicate!(query)) {
return false;
}
return true;
}
void noop([e]) => null;
bool shallowEqualMap(Map? a, Map? b) {
if ((a != null && b == null) || (b != null && a == null)) {
return false;
}
for (var item in a!.entries) {
var aVal = item.value;
var bVal = b?[item.key];
if (aVal != bVal) return false;
}
return true;
}
/// This function returns `a` if `b` is deeply equal\
/// If not, it will replace any deeply equal children of `b` with those
/// of `a`\
/// This can be used for structural sharing between JSON values for example.
/// `a` & `b` can only be Type of [Iterable] or [Map]
T replaceEqualDeep<T>(T a, T b) {
if (a == b) {
return a;
}
bool isList = (a is Iterable && b is Iterable);
if (isList || (a is Map && b is Map)) {
var aSize = isList ? a.length : (a as Map).keys.length;
var bItems = isList ? b : (b as Map).keys;
var bSize = bItems.length;
var copy;
int equalItems = 0;
for (int i = 0; i < bSize; i++) {
var key = isList ? i : (bItems as Map)[i];
if (isList) {
copy ??= [];
copy[key] = replaceEqualDeep((a as List)[key], (b as List)[key]);
if (copy[key] == a[key]) {
equalItems++;
}
} else {
copy ??= {};
copy[key] = replaceEqualDeep((a as Map)[key], (b as Map)[key]);
if (copy[key] == a[key]) {
equalItems++;
}
}
}
return aSize == bSize && equalItems == aSize ? a : copy as T;
}
return b;
}
Duration timeUntilStale(DateTime updatedAt, [Duration? staleTime]) =>
updatedAt.add(staleTime ?? Duration.zero).difference(DateTime.now());
typedef DataUpdateFunction<TInput, TOutput> = TOutput Function(TInput input);