more tests added for Query

prefetchQuery support added
seperate methods for adding & removing listeners for events of both query & mutations
refetchInterval not working when used Query.fromOptions bug fixed

packages/example/lib/components/basic_mutation.dart

@@ -56,12 +56,14 @@ class _BasicMutationExampleState extends State<BasicMutationExample> {
       crossAxisAlignment: CrossAxisAlignment.start,
       children: [
         Text(
-          "# Basic Mutation Example",
+          "# Basic Mutation Example (with Failure & Retry simulation)",
           style: Theme.of(context).textTheme.headline5,
         ),
         MutationBuilder<Map, Map<String, dynamic>>(
             job: basicMutationJob,
             onMutate: (v) {
+              final data =
+                  QueryBowl.of(context).getQuery(successJob.queryKey)?.data;
               QueryBowl.of(context)
                   .setQueryData<String, void>(successJob.queryKey, (oldData) {
                 if (oldData?.contains("After Mutate (OPTIMISTIC UPDATE)") ==
@@ -70,6 +72,11 @@ class _BasicMutationExampleState extends State<BasicMutationExample> {
                 }
                 return "$oldData - After Mutate (OPTIMISTIC UPDATE)";
               });
+              return data;
+            },
+            onData: (data, variables, context) {
+              print("Passed Variable: $variables");
+              print("Safe Previous Value: $context");
             },
             builder: (context, mutation) {
               return Padding(
@@ -94,7 +101,7 @@ class _BasicMutationExampleState extends State<BasicMutationExample> {
                           "title": title,
                           "body": body,
                           "id": id,
-                        }, onData: (data) {
+                        }, onData: (data, variables, context) {
                           // resetting the form
                           titleController.text = "";
                           bodyController.text = "";

packages/example/lib/components/hooks/basic_hook_mutation.dart

@@ -49,7 +49,7 @@ class BasicHookMutationExample extends HookWidget {
       crossAxisAlignment: CrossAxisAlignment.start,
       children: [
         Text(
-          "# Basic Mutation Hook Example",
+          "# Basic Mutation Hook Example (with failure & retry simulation)",
           style: Theme.of(context).textTheme.headline5,
         ),
         Padding(
@@ -74,7 +74,7 @@ class BasicHookMutationExample extends HookWidget {
                     "title": title,
                     "body": body,
                     "id": id,
-                  }, onData: (data) {
+                  }, onData: (data, variable, context) {
                     // resetting the form
                     titleController.text = "";
                     bodyController.text = "";

packages/fl_query/.flutter-plugins-dependencies

@@ -1 +1 @@
-{"info":"This is a generated file; do not edit or check into version control.","plugins":{"ios":[{"name":"connectivity_plus","path":"/home/krtirtho/.pub-cache/hosted/pub.dartlang.org/connectivity_plus-2.3.5/","native_build":true,"dependencies":[]}],"android":[{"name":"connectivity_plus","path":"/home/krtirtho/.pub-cache/hosted/pub.dartlang.org/connectivity_plus-2.3.5/","native_build":true,"dependencies":[]}],"macos":[{"name":"connectivity_plus_macos","path":"/home/krtirtho/.pub-cache/hosted/pub.dartlang.org/connectivity_plus_macos-1.2.4/","native_build":true,"dependencies":[]}],"linux":[{"name":"connectivity_plus_linux","path":"/home/krtirtho/.pub-cache/hosted/pub.dartlang.org/connectivity_plus_linux-1.3.1/","native_build":false,"dependencies":[]}],"windows":[{"name":"connectivity_plus_windows","path":"/home/krtirtho/.pub-cache/hosted/pub.dartlang.org/connectivity_plus_windows-1.2.2/","native_build":true,"dependencies":[]}],"web":[{"name":"connectivity_plus_web","path":"/home/krtirtho/.pub-cache/hosted/pub.dartlang.org/connectivity_plus_web-1.2.2/","dependencies":[]}]},"dependencyGraph":[{"name":"connectivity_plus","dependencies":["connectivity_plus_linux","connectivity_plus_macos","connectivity_plus_web","connectivity_plus_windows"]},{"name":"connectivity_plus_linux","dependencies":[]},{"name":"connectivity_plus_macos","dependencies":[]},{"name":"connectivity_plus_web","dependencies":[]},{"name":"connectivity_plus_windows","dependencies":[]}],"date_created":"2022-07-20 12:20:16.617434","version":"3.0.1"}
+{"info":"This is a generated file; do not edit or check into version control.","plugins":{"ios":[{"name":"connectivity_plus","path":"/home/krtirtho/.pub-cache/hosted/pub.dartlang.org/connectivity_plus-2.3.5/","native_build":true,"dependencies":[]}],"android":[{"name":"connectivity_plus","path":"/home/krtirtho/.pub-cache/hosted/pub.dartlang.org/connectivity_plus-2.3.5/","native_build":true,"dependencies":[]}],"macos":[{"name":"connectivity_plus_macos","path":"/home/krtirtho/.pub-cache/hosted/pub.dartlang.org/connectivity_plus_macos-1.2.4/","native_build":true,"dependencies":[]}],"linux":[{"name":"connectivity_plus_linux","path":"/home/krtirtho/.pub-cache/hosted/pub.dartlang.org/connectivity_plus_linux-1.3.1/","native_build":false,"dependencies":[]}],"windows":[{"name":"connectivity_plus_windows","path":"/home/krtirtho/.pub-cache/hosted/pub.dartlang.org/connectivity_plus_windows-1.2.2/","native_build":true,"dependencies":[]}],"web":[{"name":"connectivity_plus_web","path":"/home/krtirtho/.pub-cache/hosted/pub.dartlang.org/connectivity_plus_web-1.2.2/","dependencies":[]}]},"dependencyGraph":[{"name":"connectivity_plus","dependencies":["connectivity_plus_linux","connectivity_plus_macos","connectivity_plus_web","connectivity_plus_windows"]},{"name":"connectivity_plus_linux","dependencies":[]},{"name":"connectivity_plus_macos","dependencies":[]},{"name":"connectivity_plus_web","dependencies":[]},{"name":"connectivity_plus_windows","dependencies":[]}],"date_created":"2022-07-28 10:55:33.933539","version":"3.0.1"}

packages/fl_query/lib/src/models/query_job.dart

@@ -1,3 +1,4 @@
+import 'package:connectivity_plus/connectivity_plus.dart';
 import 'package:fl_query/src/query.dart';
 
 class QueryJob<T extends Object, Outside> {
@@ -20,6 +21,7 @@ class QueryJob<T extends Object, Outside> {
   Duration? cacheTime;
 
   Duration? refetchInterval;
+  Connectivity? connectivity;
 
   QueryJob({
     required String queryKey,
@@ -34,6 +36,7 @@ class QueryJob<T extends Object, Outside> {
     this.refetchOnMount,
     this.refetchOnReconnect,
     this.refetchOnExternalDataChange,
+    this.connectivity,
   }) : _queryKey = queryKey;
 
   String get queryKey => _queryKey;
@@ -56,6 +59,7 @@ class QueryJob<T extends Object, Outside> {
     bool? refetchOnMount,
     bool? refetchOnReconnect,
     bool? refetchOnExternalDataChange,
+    Connectivity? connectivity,
   }) {
     return (String queryKey) {
       if (preQueryKey != null) queryKey = "$preQueryKey#$queryKey";
@@ -72,6 +76,7 @@ class QueryJob<T extends Object, Outside> {
         refetchOnMount: refetchOnMount,
         refetchOnReconnect: refetchOnReconnect,
         refetchOnExternalDataChange: refetchOnExternalDataChange,
+        connectivity: connectivity,
       );
     };
   }

packages/fl_query/lib/src/mutation.dart

@@ -1,5 +1,6 @@
 import 'dart:async';
 
+import 'package:fl_query/fl_query.dart';
 import 'package:fl_query/src/base_operation.dart';
 import 'package:fl_query/src/models/mutation_job.dart';
 import 'package:flutter/widgets.dart';
@@ -11,7 +12,13 @@ enum MutationStatus {
   idle,
 }
 
-typedef MutationListener<T> = FutureOr<void> Function(T);
+typedef MutationListenerReturnable<T, R> = FutureOr<R> Function(T);
+
+typedef MutationListener<T, V> = FutureOr<void> Function(
+  T payload,
+  V variables,
+  dynamic context,
+);
 
 typedef MutationTaskFunction<T, V> = FutureOr<T> Function(
     String queryKey, V variables);
@@ -23,12 +30,18 @@ class Mutation<T extends Object, V> extends BaseOperation<T> {
 
   MutationStatus status;
 
+  dynamic _sideEffectContext;
+
   @protected
-  final Set<MutationListener<T>> onDataListeners = {};
+  final Set<MutationListener<T, V>> _onDataListeners = {};
   @protected
-  final Set<MutationListener<dynamic>> onErrorListeners = {};
+  final Set<MutationListener<dynamic, V>> _onErrorListeners = {};
   @protected
-  final Set<MutationListener<V>> onMutateListeners = {};
+  final Set<MutationListenerReturnable<V, dynamic>> _onMutateListeners = {};
+
+  // using late as _variables will only be used after a [mutate] or
+  // [mutateAsync] is executed
+  late V _variables;
 
   Mutation({
     required this.mutationKey,
@@ -37,21 +50,21 @@ class Mutation<T extends Object, V> extends BaseOperation<T> {
     required super.retryDelay,
     required super.queryBowl,
     required Duration cacheTime,
-    MutationListener<T>? onData,
-    MutationListener<dynamic>? onError,
-    MutationListener<V>? onMutate,
+    MutationListener<T, V>? onData,
+    MutationListener<dynamic, V>? onError,
+    MutationListenerReturnable<V, dynamic>? onMutate,
   })  : status = MutationStatus.idle,
         super(cacheTime: cacheTime) {
-    if (onData != null) onDataListeners.add(onData);
-    if (onError != null) onErrorListeners.add(onError);
-    if (onMutate != null) onMutateListeners.add(onMutate);
+    if (onData != null) _onDataListeners.add(onData);
+    if (onError != null) _onErrorListeners.add(onError);
+    if (onMutate != null) _onMutateListeners.add(onMutate);
   }
 
   Mutation.fromOptions(
     MutationJob<T, V> options, {
-    MutationListener<T>? onData,
-    MutationListener<dynamic>? onError,
-    MutationListener<V>? onMutate,
+    MutationListener<T, V>? onData,
+    MutationListener<dynamic, V>? onError,
+    MutationListenerReturnable<V, dynamic>? onMutate,
     required super.queryBowl,
   })  : mutationKey = options.mutationKey,
         task = options.task,
@@ -61,8 +74,8 @@ class Mutation<T extends Object, V> extends BaseOperation<T> {
           retryDelay: options.retryDelay ?? const Duration(milliseconds: 200),
           cacheTime: options.cacheTime ?? const Duration(minutes: 5),
         ) {
-    if (onData != null) onDataListeners.add(onData);
-    if (onError != null) onErrorListeners.add(onError);
+    if (onData != null) _onDataListeners.add(onData);
+    if (onError != null) _onErrorListeners.add(onError);
   }
 
   // all methods
@@ -74,22 +87,22 @@ class Mutation<T extends Object, V> extends BaseOperation<T> {
       status = MutationStatus.loading;
       notifyListeners();
       retryAttempts = 0;
-      for (final onMutate in onMutateListeners) {
-        onMutate(variables);
+      for (final onMutate in _onMutateListeners) {
+        _sideEffectContext = await onMutate(variables);
       }
       data = await task(mutationKey, variables);
       updatedAt = DateTime.now();
       status = MutationStatus.success;
-      for (final onData in onDataListeners) {
-        onData(data!);
+      for (final onData in _onDataListeners) {
+        onData(data!, _variables, _sideEffectContext);
       }
       notifyListeners();
     } catch (e) {
       if (retries == 0) {
         status = MutationStatus.error;
         error = e;
-        for (final onError in onErrorListeners) {
-          onError(error);
+        for (final onError in _onErrorListeners) {
+          onError(error, variables, _sideEffectContext);
         }
         notifyListeners();
       } else {
@@ -97,13 +110,13 @@ class Mutation<T extends Object, V> extends BaseOperation<T> {
         while (retryAttempts <= retries) {
           await Future.delayed(retryDelay);
           try {
-            for (final onMutate in onMutateListeners) {
-              onMutate(variables);
+            for (final onMutate in _onMutateListeners) {
+              _sideEffectContext = onMutate(variables);
             }
             data = await task(mutationKey, variables);
             status = MutationStatus.success;
-            for (final onData in onDataListeners) {
-              onData(data!);
+            for (final onData in _onDataListeners) {
+              onData(data!, variables, _sideEffectContext);
             }
             notifyListeners();
             break;
@@ -111,8 +124,8 @@ class Mutation<T extends Object, V> extends BaseOperation<T> {
             if (retryAttempts == retries) {
               status = MutationStatus.error;
               error = e;
-              for (final onError in onErrorListeners) {
-                onError(error);
+              for (final onError in _onErrorListeners) {
+                onError(error, variables, _sideEffectContext);
               }
               notifyListeners();
             }
@@ -123,20 +136,46 @@ class Mutation<T extends Object, V> extends BaseOperation<T> {
     }
   }
 
+  void addDataListener(MutationListener<T, V> listener) {
+    _onDataListeners.add(listener);
+  }
+
+  void addErrorListener(MutationListener<dynamic, V> listener) {
+    _onErrorListeners.add(listener);
+  }
+
+  void addMutateListener(MutationListenerReturnable<V, dynamic> listener) {
+    _onMutateListeners.add(listener);
+  }
+
+  void removeDataListener(MutationListener<T, V> listener) {
+    _onDataListeners.remove(listener);
+  }
+
+  void removeErrorListener(MutationListener<dynamic, V> listener) {
+    _onErrorListeners.remove(listener);
+  }
+
+  void removeMutateListener(MutationListenerReturnable<V, dynamic> listener) {
+    _onMutateListeners.remove(listener);
+  }
+
   void mutate(
     V variables, {
-    MutationListener<T>? onData,
-    MutationListener<dynamic>? onError,
+    MutationListener<T, V>? onData,
+    MutationListener<dynamic, V>? onError,
   }) {
-    if (onData != null) onDataListeners.add(onData);
-    if (onError != null) onErrorListeners.add(onError);
+    _variables = variables;
+    if (onData != null) _onDataListeners.add(onData);
+    if (onError != null) _onErrorListeners.add(onError);
     _execute(variables).then((_) {
-      onDataListeners.remove(onData);
-      onErrorListeners.remove(onError);
+      _onDataListeners.remove(onData);
+      _onErrorListeners.remove(onError);
     });
   }
 
   Future<T?> mutateAsync(V variables) async {
+    _variables = variables;
     return await _execute(variables).then((_) => data);
   }
 
@@ -155,10 +194,11 @@ class Mutation<T extends Object, V> extends BaseOperation<T> {
     data = null;
     retryAttempts = 0;
     updatedAt = DateTime.now();
-    onDataListeners.clear();
-    onErrorListeners.clear();
+    _onDataListeners.clear();
+    _onErrorListeners.clear();
     status = MutationStatus.idle;
-    onMutateListeners.clear();
+    _onMutateListeners.clear();
+    _sideEffectContext = null;
   }
 
   A? cast<A>() => this is A ? this as A : null;

packages/fl_query/lib/src/mutation_builder.dart

@@ -10,15 +10,15 @@ class MutationBuilder<T extends Object, V> extends StatefulWidget {
 
   /// Called when the query returns new data, on query
   /// refetch or query gets expired
-  final MutationListener<T>? onData;
+  final MutationListener<T, V>? onData;
 
   /// Called when the query returns error
-  final MutationListener<dynamic>? onError;
+  final MutationListener<dynamic, V>? onError;
 
   /// called right before the mutation is about to run
   ///
   /// perfect scenario for doing optimistic updates
-  final MutationListener<V>? onMutate;
+  final MutationListenerReturnable<V, dynamic>? onMutate;
 
   const MutationBuilder({
     required this.job,
@@ -66,19 +66,17 @@ class _MutationBuilderState<T extends Object, V>
       init();
     } else {
       if (oldWidget.onData != widget.onData && oldWidget.onData != null) {
-        mutation?.onDataListeners.remove(oldWidget.onData);
-        if (widget.onData != null)
-          mutation?.onDataListeners.add(widget.onData!);
+        mutation?.removeDataListener(oldWidget.onData!);
+        if (widget.onData != null) mutation?.addDataListener(widget.onData!);
       }
       if (oldWidget.onError != widget.onError && oldWidget.onError != null) {
-        mutation?.onErrorListeners.remove(oldWidget.onError);
-        if (widget.onError != null)
-          mutation?.onErrorListeners.add(widget.onError!);
+        mutation?.removeErrorListener(oldWidget.onError!);
+        if (widget.onError != null) mutation?.addErrorListener(widget.onError!);
       }
       if (oldWidget.onMutate != widget.onMutate && oldWidget.onMutate != null) {
-        mutation?.onMutateListeners.remove(oldWidget.onMutate);
+        mutation?.removeMutateListener(oldWidget.onMutate!);
         if (widget.onMutate != null)
-          mutation?.onMutateListeners.add(widget.onMutate!);
+          mutation?.addMutateListener(widget.onMutate!);
       }
     }
     super.didUpdateWidget(oldWidget);
@@ -92,11 +90,9 @@ class _MutationBuilderState<T extends Object, V>
 
   void _mutationDispose() {
     mutation?.unmount(uKey);
-    if (widget.onData != null) mutation?.onDataListeners.remove(widget.onData);
-    if (widget.onError != null)
-      mutation?.onErrorListeners.remove(widget.onError);
-    if (widget.onMutate != null)
-      mutation?.onMutateListeners.remove(widget.onMutate);
+    if (widget.onData != null) mutation?.addDataListener(widget.onData!);
+    if (widget.onError != null) mutation?.addErrorListener(widget.onError!);
+    if (widget.onMutate != null) mutation?.addMutateListener(widget.onMutate!);
   }
 
   @override

packages/fl_query/lib/src/query.dart

@@ -56,10 +56,8 @@ class Query<T extends Object, Outside> extends BaseOperation<T> {
 
   QueryStatus status;
 
-  @protected
-  final Set<QueryListener<T>> onDataListeners = Set<QueryListener<T>>();
-  @protected
-  final Set<QueryListener<dynamic>> onErrorListeners =
+  final Set<QueryListener<T>> _onDataListeners = Set<QueryListener<T>>();
+  final Set<QueryListener<dynamic>> _onErrorListeners =
       Set<QueryListener<dynamic>>();
 
   // externalData will always be passed to the task Callback
@@ -72,6 +70,8 @@ class Query<T extends Object, Outside> extends BaseOperation<T> {
 
   Timer? _refetchIntervalTimer;
 
+  Connectivity _connectivity;
+
   Query({
     required this.queryKey,
     required this.task,
@@ -85,6 +85,7 @@ class Query<T extends Object, Outside> extends BaseOperation<T> {
     this.refetchOnReconnect,
     this.refetchInterval,
     this.enabled = true,
+    Connectivity? connectivity,
     T? initialData,
     QueryListener<T>? onData,
     QueryListener<dynamic>? onError,
@@ -92,9 +93,10 @@ class Query<T extends Object, Outside> extends BaseOperation<T> {
         _initialData = initialData,
         _externalData = externalData,
         status = QueryStatus.idle,
+        _connectivity = connectivity ?? Connectivity(),
         super(data: initialData) {
-    if (onData != null) onDataListeners.add(onData);
-    if (onError != null) onErrorListeners.add(onError);
+    if (onData != null) _onDataListeners.add(onData);
+    if (onError != null) _onErrorListeners.add(onError);
 
     if (refetchInterval != null && refetchInterval != Duration.zero) {
       _refetchIntervalTimer = _createRefetchTimer();
@@ -117,14 +119,18 @@ class Query<T extends Object, Outside> extends BaseOperation<T> {
         refetchOnMount = options.refetchOnMount,
         refetchOnReconnect = options.refetchOnReconnect,
         status = QueryStatus.idle,
+        _connectivity = options.connectivity ?? Connectivity(),
         super(
           cacheTime: options.cacheTime ?? const Duration(minutes: 5),
           retries: options.retries ?? 3,
           retryDelay: options.retryDelay ?? const Duration(milliseconds: 200),
           data: options.initialData,
         ) {
-    if (onData != null) onDataListeners.add(onData);
-    if (onError != null) onErrorListeners.add(onError);
+    if (onData != null) _onDataListeners.add(onData);
+    if (onError != null) _onErrorListeners.add(onError);
+    if (refetchInterval != null && refetchInterval != Duration.zero) {
+      _refetchIntervalTimer = _createRefetchTimer();
+    }
   }
 
   // all getters & setters
@@ -138,9 +144,7 @@ class Query<T extends Object, Outside> extends BaseOperation<T> {
       (_) async {
         // only refetch if its connected to the internet or refetch will
         // always result in error while there's no internet
-        if (isStale &&
-            isConnectedToInternet(await Connectivity().checkConnectivity()))
-          await refetch();
+        if (isStale && await isInternetConnected()) await refetch();
       },
     );
   }
@@ -157,7 +161,7 @@ class Query<T extends Object, Outside> extends BaseOperation<T> {
       _prevUsedExternalData = _externalData;
       updatedAt = DateTime.now();
       status = QueryStatus.success;
-      for (final onData in onDataListeners) {
+      for (final onData in _onDataListeners) {
         onData(data!);
       }
       notifyListeners();
@@ -165,7 +169,7 @@ class Query<T extends Object, Outside> extends BaseOperation<T> {
       if (retries == 0) {
         status = QueryStatus.error;
         error = e;
-        for (final onError in onErrorListeners) {
+        for (final onError in _onErrorListeners) {
           onError(error);
         }
         notifyListeners();
@@ -180,7 +184,7 @@ class Query<T extends Object, Outside> extends BaseOperation<T> {
             );
             _prevUsedExternalData = _externalData;
             status = QueryStatus.success;
-            for (final onData in onDataListeners) {
+            for (final onData in _onDataListeners) {
               await onData(data!);
             }
             notifyListeners();
@@ -189,7 +193,7 @@ class Query<T extends Object, Outside> extends BaseOperation<T> {
             if (retryAttempts == retries) {
               status = QueryStatus.error;
               error = e;
-              for (final onError in onErrorListeners) {
+              for (final onError in _onErrorListeners) {
                 await onError(error);
               }
               notifyListeners();
@@ -202,6 +206,22 @@ class Query<T extends Object, Outside> extends BaseOperation<T> {
     }
   }
 
+  void addDataListener(QueryListener<T> listener) {
+    _onDataListeners.add(listener);
+  }
+
+  void addErrorListener(QueryListener<dynamic> listener) {
+    _onErrorListeners.add(listener);
+  }
+
+  void removeDataListener(QueryListener<T> listener) {
+    _onDataListeners.remove(listener);
+  }
+
+  void removeErrorListener(QueryListener<dynamic> listener) {
+    _onErrorListeners.remove(listener);
+  }
+
   Future<T?> fetch() async {
     if (!enabled) return null;
 
@@ -258,8 +278,8 @@ class Query<T extends Object, Outside> extends BaseOperation<T> {
     fetched = false;
     status = QueryStatus.idle;
     retryAttempts = 0;
-    onDataListeners.clear();
-    onErrorListeners.clear();
+    _onDataListeners.clear();
+    _onErrorListeners.clear();
     mounts.clear();
   }
 
@@ -290,13 +310,27 @@ class Query<T extends Object, Outside> extends BaseOperation<T> {
     notifyListeners();
   }
 
+  Future<bool> isInternetConnected() async {
+    return isConnectedToInternet(await _connectivity.checkConnectivity());
+  }
+
+  /// invalidates the query
+  void invalidate() {
+    /// subtracting [staleTime] from [updatedAt] as staleTime=Duration.zero
+    /// indicates the query must never become stale but subtracting the
+    /// [staleTime] will always revert the updatedAt time to the default
+    /// time whenever isStale is called
+    updatedAt = updatedAt.subtract(_staleTime);
+    notifyListeners();
+  }
+
   bool get isStale {
     /// when [_staleTime] is [Duration.zero], the query will always be
     /// stale & will never refetch in the background. But can be inactive
     /// if [mounts.length] become zero
     if (_staleTime == Duration.zero) return false;
 
-    // when current DateTime is after [update_at + stale_time] it means
+    // when [DateTime.now()] is after [update_at + stale_time] it means
     // the data has become stale
     return DateTime.now().isAfter(updatedAt.add(_staleTime));
   }
@@ -315,10 +349,8 @@ class Query<T extends Object, Outside> extends BaseOperation<T> {
   void mount(ValueKey<String> uKey) {
     super.mount(uKey);
     if (refetchOnMount == true && isStale) {
-      Connectivity().checkConnectivity().then((status) async {
-        if (isConnectedToInternet(status)) {
-          await refetch();
-        }
+      this.isInternetConnected().then((isConnected) async {
+        if (isConnected) await refetch();
       });
     }
   }

packages/fl_query/lib/src/query_bowl.dart

@@ -282,6 +282,39 @@ class QueryBowl extends InheritedWidget {
         _addMutation = addMutation,
         super(child: child, key: key);
 
+  Query<T, Outside> _createQueryWithDefaults<T extends Object, Outside>(
+    QueryJob<T, Outside> options,
+    Outside externalData,
+  ) {
+    final query = Query<T, Outside>.fromOptions(
+      options,
+      externalData: externalData,
+      queryBowl: this,
+    );
+    query.updateDefaultOptions(
+      cacheTime: cacheTime,
+      staleTime: staleTime,
+      refetchInterval: refetchInterval,
+      refetchOnMount: refetchOnMount,
+      refetchOnReconnect: refetchOnReconnect,
+    );
+    return query;
+  }
+
+  Future<T?> prefetchQuery<T extends Object, Outside>(
+    QueryJob<T, Outside> options, {
+    required Outside externalData,
+  }) async {
+    final prevQuery =
+        _queries.firstWhereOrNull((q) => q.queryKey == options.queryKey);
+    if (prevQuery != null && prevQuery is Query<T, Outside>)
+      return prevQuery.data;
+
+    final query = _createQueryWithDefaults<T, Outside>(options, externalData);
+    _addQuery<T, Outside>(query);
+    return await query.fetch();
+  }
+
   /// !⚠️**Warning** only for internal library usage
   @protected
   Future<T?> fetchQuery<T extends Object, Outside>(
@@ -303,8 +336,8 @@ class QueryBowl extends InheritedWidget {
             externalData,
           );
       prevQuery.mount(key);
-      if (onData != null) prevQuery.onDataListeners.add(onData);
-      if (onError != null) prevQuery.onErrorListeners.add(onError);
+      if (onData != null) prevQuery.addDataListener(onData);
+      if (onError != null) prevQuery.addErrorListener(onError);
       if (!prevQuery.hasData || hasExternalDataChanged) {
         if (hasExternalDataChanged) prevQuery.setExternalData(externalData);
 
@@ -314,19 +347,7 @@ class QueryBowl extends InheritedWidget {
       return prevQuery.data;
     }
 
-    /// populating with default configurations
-    options.refetchInterval ??= refetchInterval;
-    options.staleTime ??= staleTime;
-    options.cacheTime ??= cacheTime;
-    options.refetchOnMount ??= refetchOnMount;
-    options.refetchOnReconnect ??= refetchOnReconnect;
-    final query = Query<T, Outside>.fromOptions(
-      options,
-      externalData: externalData,
-      queryBowl: this,
-      onData: onData,
-      onError: onError,
-    );
+    final query = _createQueryWithDefaults<T, Outside>(options, externalData);
     query.mount(key);
     _addQuery<T, Outside>(query);
     return await query.fetch();
@@ -355,25 +376,14 @@ class QueryBowl extends InheritedWidget {
         prevQuery.setExternalData(externalData);
       }
       prevQuery.mount(key);
-      if (onData != null) prevQuery.onDataListeners.add(onData);
-      if (onError != null) prevQuery.onErrorListeners.add(onError);
+      if (onData != null) prevQuery.addDataListener(onData);
+      if (onError != null) prevQuery.addErrorListener(onError);
       // mounting the widget that is using the query in the prevQuery
       return prevQuery;
     }
-    final query = Query<T, Outside>.fromOptions(
-      queryJob,
-      externalData: externalData,
-      queryBowl: this,
-    );
-    if (onData != null) query.onDataListeners.add(onData);
-    if (onError != null) query.onErrorListeners.add(onError);
-    query.updateDefaultOptions(
-      cacheTime: cacheTime,
-      staleTime: staleTime,
-      refetchInterval: refetchInterval,
-      refetchOnMount: refetchOnMount,
-      refetchOnReconnect: refetchOnReconnect,
-    );
+    final query = _createQueryWithDefaults<T, Outside>(queryJob, externalData);
+    if (onData != null) query.addDataListener(onData);
+    if (onError != null) query.addErrorListener(onError);
     query.mount(key);
     _addQuery<T, Outside>(query);
     return query;
@@ -383,17 +393,17 @@ class QueryBowl extends InheritedWidget {
   @protected
   Mutation<T, V> addMutation<T extends Object, V>(
     MutationJob<T, V> mutationJob, {
-    final MutationListener<T>? onData,
-    final MutationListener<dynamic>? onError,
-    final MutationListener<V>? onMutate,
+    final MutationListener<T, V>? onData,
+    final MutationListener<dynamic, V>? onError,
+    final MutationListenerReturnable<V, dynamic>? onMutate,
     required ValueKey<String> key,
   }) {
     final prevMutation = _mutations.firstWhereOrNull(
         (prevMutation) => prevMutation.mutationKey == mutationJob.mutationKey);
     if (prevMutation != null && prevMutation is Mutation<T, V>) {
-      if (onData != null) prevMutation.onDataListeners.add(onData);
-      if (onError != null) prevMutation.onErrorListeners.add(onError);
-      if (onMutate != null) prevMutation.onMutateListeners.add(onMutate);
+      if (onData != null) prevMutation.addDataListener(onData);
+      if (onError != null) prevMutation.addErrorListener(onError);
+      if (onMutate != null) prevMutation.addMutateListener(onMutate);
       prevMutation.mount(key);
       return prevMutation;
     } else {
@@ -401,9 +411,9 @@ class QueryBowl extends InheritedWidget {
         mutationJob,
         queryBowl: this,
       );
-      if (onData != null) mutation.onDataListeners.add(onData);
-      if (onError != null) mutation.onErrorListeners.add(onError);
-      if (onMutate != null) mutation.onMutateListeners.add(onMutate);
+      if (onData != null) mutation.addDataListener(onData);
+      if (onError != null) mutation.addErrorListener(onError);
+      if (onMutate != null) mutation.addMutateListener(onMutate);
       mutation.updateDefaultOptions(cacheTime: cacheTime);
       mutation.mount(key);
       _addMutation(mutation);
@@ -461,29 +471,37 @@ class QueryBowl extends InheritedWidget {
   }
 
   /// resets all the queries matching the passed List of queryKeys
+  ///
+  /// If an empty list of [queryKeys] is passed then all of the queries
+  /// will be reset
   void resetQueries(List<String> queryKeys) {
     for (final query in _queries) {
-      if (!queryKeys.contains(query.queryKey)) continue;
+      if (queryKeys.isNotEmpty && !queryKeys.contains(query.queryKey)) continue;
       query.reset();
     }
   }
 
   /// makes all the queries matching the passed List of queryKeys stale
+  ///
+  /// If an empty list of [queryKeys] is passed then all of the queries
+  /// will be invalidated
   void invalidateQueries(List<String> queryKeys) {
     for (final query in _queries) {
-      if (!queryKeys.contains(query.queryKey)) continue;
-      // TODO: Implement Invaldiate Queries
+      if (queryKeys.isNotEmpty && queryKeys.contains(query.queryKey)) continue;
+      query.invalidate();
     }
   }
 
   /// refetches all the queries matching the passed List of queryKeys
+  ///
+  /// If an empty list of [queryKeys] is passed then all of the queries
+  /// will be refetched
   Future<void> refetchQueries(List<String> queryKeys) async {
     for (final query in _queries) {
-      if (queryKeys.contains(query.queryKey)) {
+      if (queryKeys.isNotEmpty && queryKeys.contains(query.queryKey)) continue;
       await query.refetch();
     }
   }
-  }
 
   static QueryBowl of(BuildContext context) =>
       context.dependOnInheritedWidgetOfExactType<QueryBowl>()!;

packages/fl_query/lib/src/query_builder.dart

@@ -90,17 +90,16 @@ class _QueryBuilderState<T extends Object, Outside>
             .getQuery(widget.job.queryKey)
             ?.setExternalData(widget.externalData);
       }
-      if (hasOnDataChanged) query?.onDataListeners.remove(oldWidget.onData);
-      if (hasOnErrorChanged) query?.onErrorListeners.remove(oldWidget.onError);
+      if (hasOnDataChanged) query?.removeDataListener(oldWidget.onData!);
+      if (hasOnErrorChanged) query?.removeErrorListener(oldWidget.onError!);
     } else {
       if (hasOnDataChanged) {
-        query?.onDataListeners.remove(oldWidget.onData);
-        if (widget.onData != null) query?.onDataListeners.add(widget.onData!);
+        query?.removeDataListener(oldWidget.onData!);
+        if (widget.onData != null) query?.addDataListener(widget.onData!);
       }
       if (hasOnErrorChanged) {
-        query?.onErrorListeners.remove(oldWidget.onError);
-        if (widget.onError != null)
-          query?.onErrorListeners.add(widget.onError!);
+        query?.removeErrorListener(oldWidget.onError!);
+        if (widget.onError != null) query?.addErrorListener(widget.onError!);
       }
     }
     super.didUpdateWidget(oldWidget);
@@ -108,8 +107,8 @@ class _QueryBuilderState<T extends Object, Outside>
 
   _queryDispose() {
     query?.unmount(uKey);
-    if (widget.onData != null) query?.onDataListeners.remove(widget.onData);
-    if (widget.onError != null) query?.onErrorListeners.remove(widget.onError);
+    if (widget.onData != null) query?.removeDataListener(widget.onData!);
+    if (widget.onError != null) query?.removeErrorListener(widget.onError!);
   }
 
   @override

packages/fl_query/test/query_test.dart

@@ -1,15 +1,18 @@
 import 'dart:math';
 
+import 'package:connectivity_plus/connectivity_plus.dart';
 import 'package:fl_query/src/models/query_job.dart';
 import 'package:fl_query/src/query.dart';
 import 'package:fl_query/src/query_bowl.dart';
+import 'package:fl_query/src/utils.dart';
+import 'package:flutter/widgets.dart';
 import 'package:flutter_test/flutter_test.dart';
 import 'package:mockito/annotations.dart';
 import 'package:mockito/mockito.dart';
 import 'query_test.mocks.dart';
 
 @GenerateMocks(
-  [QueryBowl],
+  [QueryBowl, Connectivity],
   customMocks: [
     MockSpec<QueryJob<Object, void>>(
       as: #MockQueryJobVoidObject,
@@ -129,7 +132,7 @@ void main() {
       "onData listeners are called When new data is fetched and set",
       () async {
         int count = 0;
-        query.onDataListeners.add((_) {
+        query.addDataListener((_) {
           count++;
         });
         await query.fetch();
@@ -138,20 +141,111 @@ void main() {
     );
     test(
       "onError listeners are called When any error occurs",
-      () {},
+      () async {
+        reset(queryJob);
+        when(queryJob.queryKey).thenReturn("test");
+        when(queryJob.task).thenAnswer(
+          (_) => (_, __) => Future.error("Error"),
         );
-    test("query should become stale after defined amount time", () {});
+        query = Query.fromOptions(
+          queryJob,
+          externalData: null,
+          queryBowl: queryBowl,
+        );
+        int count = 0;
+        query.addErrorListener((_) {
+          count++;
+        });
+        await query.fetch();
+        expect(count, 1);
+      },
+    );
+    test("query should become stale after defined amount time", () async {
+      when(queryJob.staleTime).thenReturn(Duration(milliseconds: 500));
+      query = Query.fromOptions(
+        queryJob,
+        externalData: null,
+        queryBowl: queryBowl,
+      );
+      expect(query.isStale, isFalse);
+      await Future.delayed(Duration(milliseconds: 300));
+      expect(query.isStale, isFalse);
+      await Future.delayed(Duration(milliseconds: 600));
+      expect(query.isStale, isTrue);
+    });
     test(
       "query should refetch in interval When refetchInterval is specified",
-      () {},
+      () async {
+        reset(queryJob);
+        when(queryJob.queryKey).thenReturn("test");
+        when(queryJob.task).thenAnswer(
+          (_) => (_, __) => Future.value(Random().nextInt(100)),
+        );
+        when(queryJob.staleTime).thenReturn(
+          Duration(milliseconds: 1),
+        );
+        when(queryJob.refetchInterval).thenReturn(
+          Duration(milliseconds: 200),
+        );
+        query = Query.fromOptions(
+          queryJob,
+          externalData: null,
+          queryBowl: queryBowl,
+        );
+
+        final data = await query.fetch();
+        await Future.delayed(Duration(milliseconds: 400));
+        expect(data, isNot(equals(await query.fetch())));
+      },
     );
     test(
       "query should revalidate When a new caller gets mounted",
-      () {},
+      () async {
+        reset(queryJob);
+        when(queryJob.queryKey).thenReturn("test");
+        when(queryJob.task).thenAnswer(
+          (_) => (_, __) => Future.value(Random().nextInt(150)),
+        );
+        when(queryJob.refetchOnMount).thenReturn(true);
+        query = Query.fromOptions(
+          queryJob,
+          externalData: null,
+          queryBowl: queryBowl,
+        );
+        final data = await query.fetch();
+        query.invalidate();
+        query.mount(ValueKey<String>(uuid.v4()));
+        await Future.delayed(Duration(milliseconds: 100));
+        expect(data, isNot(equals(await query.fetch())));
+        expect(query.refetchCount, 1);
+      },
     );
     test(
       "query should not revalidate When there's no Internet Connectivity and a new caller gets mounted",
-      () {},
+      () async {
+        final connectivityMock = MockConnectivity();
+        when(connectivityMock.checkConnectivity()).thenAnswer(
+          (_) async => ConnectivityResult.none,
+        );
+        reset(queryJob);
+        when(queryJob.queryKey).thenReturn("test");
+        when(queryJob.task).thenAnswer(
+          (_) => (_, __) => Future.value(Random().nextInt(150)),
+        );
+        when(queryJob.connectivity).thenReturn(connectivityMock);
+        when(queryJob.refetchOnMount).thenReturn(true);
+        query = Query.fromOptions(
+          queryJob,
+          externalData: null,
+          queryBowl: queryBowl,
+        );
+        final data = await query.fetch();
+        query.invalidate();
+        query.mount(ValueKey<String>(uuid.v4()));
+        await Future.delayed(Duration(milliseconds: 100));
+        expect(data, equals(await query.fetch()));
+        expect(query.refetchCount, 0);
+      },
     );
   });
 }

packages/fl_query/test/query_test.mocks.dart

@@ -4,13 +4,13 @@
 
 import 'dart:async' as _i7;
 
+import 'package:connectivity_plus/connectivity_plus.dart' as _i10;
 import 'package:fl_query/src/models/mutation_job.dart' as _i9;
 import 'package:fl_query/src/models/query_job.dart' as _i8;
 import 'package:fl_query/src/mutation.dart' as _i4;
 import 'package:fl_query/src/query.dart' as _i3;
 import 'package:fl_query/src/query_bowl.dart' as _i6;
 import 'package:flutter/foundation.dart' as _i5;
-import 'package:flutter/rendering.dart' as _i10;
 import 'package:flutter/widgets.dart' as _i2;
 import 'package:mockito/mockito.dart' as _i1;
 
@@ -136,9 +136,9 @@ class MockQueryBowl extends _i1.Mock implements _i6.QueryBowl {
   @override
   _i4.Mutation<T, V> addMutation<T extends Object, V>(
           _i9.MutationJob<T, V>? mutationJob,
-          {_i4.MutationListener<T>? onData,
-          _i4.MutationListener<dynamic>? onError,
-          _i4.MutationListener<V>? onMutate,
+          {_i4.MutationListener<T, V>? onData,
+          _i4.MutationListener<dynamic, V>? onError,
+          _i4.MutationListenerReturnable<V, dynamic>? onMutate,
           _i2.ValueKey<String>? key}) =>
       (super.noSuchMethod(
           Invocation.method(#addMutation, [
@@ -190,7 +190,7 @@ class MockQueryBowl extends _i1.Mock implements _i6.QueryBowl {
           .noSuchMethod(Invocation.method(#toStringShort, []), returnValue: '')
       as String);
   @override
-  void debugFillProperties(_i10.DiagnosticPropertiesBuilder? properties) =>
+  void debugFillProperties(_i5.DiagnosticPropertiesBuilder? properties) =>
       super.noSuchMethod(Invocation.method(#debugFillProperties, [properties]),
           returnValueForMissingStub: null);
   @override
@@ -229,6 +229,27 @@ class MockQueryBowl extends _i1.Mock implements _i6.QueryBowl {
       super.toString();
 }
 
+/// A class which mocks [Connectivity].
+///
+/// See the documentation for Mockito's code generation for more information.
+class MockConnectivity extends _i1.Mock implements _i10.Connectivity {
+  MockConnectivity() {
+    _i1.throwOnMissingStub(this);
+  }
+
+  @override
+  _i7.Stream<_i10.ConnectivityResult> get onConnectivityChanged =>
+      (super.noSuchMethod(Invocation.getter(#onConnectivityChanged),
+              returnValue: Stream<_i10.ConnectivityResult>.empty())
+          as _i7.Stream<_i10.ConnectivityResult>);
+  @override
+  _i7.Future<_i10.ConnectivityResult> checkConnectivity() =>
+      (super.noSuchMethod(Invocation.method(#checkConnectivity, []),
+              returnValue: Future<_i10.ConnectivityResult>.value(
+                  _i10.ConnectivityResult.bluetooth))
+          as _i7.Future<_i10.ConnectivityResult>);
+}
+
 /// A class which mocks [QueryJob].
 ///
 /// See the documentation for Mockito's code generation for more information.
@@ -271,6 +292,10 @@ class MockQueryJobVoidObject extends _i1.Mock
       super.noSuchMethod(Invocation.setter(#refetchInterval, _refetchInterval),
           returnValueForMissingStub: null);
   @override
+  set connectivity(_i10.Connectivity? _connectivity) =>
+      super.noSuchMethod(Invocation.setter(#connectivity, _connectivity),
+          returnValueForMissingStub: null);
+  @override
   String get queryKey =>
       (super.noSuchMethod(Invocation.getter(#queryKey), returnValue: '')
           as String);

packages/fl_query_hooks/.flutter-plugins-dependencies

@@ -1 +1 @@
-{"info":"This is a generated file; do not edit or check into version control.","plugins":{"ios":[{"name":"connectivity_plus","path":"/home/krtirtho/.pub-cache/hosted/pub.dartlang.org/connectivity_plus-2.3.6/","native_build":true,"dependencies":[]}],"android":[{"name":"connectivity_plus","path":"/home/krtirtho/.pub-cache/hosted/pub.dartlang.org/connectivity_plus-2.3.6/","native_build":true,"dependencies":[]}],"macos":[{"name":"connectivity_plus_macos","path":"/home/krtirtho/.pub-cache/hosted/pub.dartlang.org/connectivity_plus_macos-1.2.4/","native_build":true,"dependencies":[]}],"linux":[{"name":"connectivity_plus_linux","path":"/home/krtirtho/.pub-cache/hosted/pub.dartlang.org/connectivity_plus_linux-1.3.1/","native_build":false,"dependencies":[]}],"windows":[{"name":"connectivity_plus_windows","path":"/home/krtirtho/.pub-cache/hosted/pub.dartlang.org/connectivity_plus_windows-1.2.2/","native_build":true,"dependencies":[]}],"web":[{"name":"connectivity_plus_web","path":"/home/krtirtho/.pub-cache/hosted/pub.dartlang.org/connectivity_plus_web-1.2.3/","dependencies":[]}]},"dependencyGraph":[{"name":"connectivity_plus","dependencies":["connectivity_plus_linux","connectivity_plus_macos","connectivity_plus_web","connectivity_plus_windows"]},{"name":"connectivity_plus_linux","dependencies":[]},{"name":"connectivity_plus_macos","dependencies":[]},{"name":"connectivity_plus_web","dependencies":[]},{"name":"connectivity_plus_windows","dependencies":[]}],"date_created":"2022-07-19 23:08:04.911997","version":"3.0.1"}
+{"info":"This is a generated file; do not edit or check into version control.","plugins":{"ios":[{"name":"connectivity_plus","path":"/home/krtirtho/.pub-cache/hosted/pub.dartlang.org/connectivity_plus-2.3.6/","native_build":true,"dependencies":[]}],"android":[{"name":"connectivity_plus","path":"/home/krtirtho/.pub-cache/hosted/pub.dartlang.org/connectivity_plus-2.3.6/","native_build":true,"dependencies":[]}],"macos":[{"name":"connectivity_plus_macos","path":"/home/krtirtho/.pub-cache/hosted/pub.dartlang.org/connectivity_plus_macos-1.2.4/","native_build":true,"dependencies":[]}],"linux":[{"name":"connectivity_plus_linux","path":"/home/krtirtho/.pub-cache/hosted/pub.dartlang.org/connectivity_plus_linux-1.3.1/","native_build":false,"dependencies":[]}],"windows":[{"name":"connectivity_plus_windows","path":"/home/krtirtho/.pub-cache/hosted/pub.dartlang.org/connectivity_plus_windows-1.2.2/","native_build":true,"dependencies":[]}],"web":[{"name":"connectivity_plus_web","path":"/home/krtirtho/.pub-cache/hosted/pub.dartlang.org/connectivity_plus_web-1.2.3/","dependencies":[]}]},"dependencyGraph":[{"name":"connectivity_plus","dependencies":["connectivity_plus_linux","connectivity_plus_macos","connectivity_plus_web","connectivity_plus_windows"]},{"name":"connectivity_plus_linux","dependencies":[]},{"name":"connectivity_plus_macos","dependencies":[]},{"name":"connectivity_plus_web","dependencies":[]},{"name":"connectivity_plus_windows","dependencies":[]}],"date_created":"2022-07-28 10:55:26.979398","version":"3.0.1"}

packages/fl_query_hooks/lib/src/use_mutation.dart

@@ -8,15 +8,15 @@ Mutation<T, V> useMutation<T extends Object, V>({
 
   /// Called when the query returns new data, on query
   /// refetch or query gets expired
-  MutationListener<T>? onData,
+  MutationListener<T, V>? onData,
 
   /// Called when the query returns error
-  MutationListener<dynamic>? onError,
+  MutationListener<dynamic, V>? onError,
 
   /// called right before the mutation is about to run
   ///
   /// perfect scenario for doing optimistic updates
-  MutationListener<V>? onMutate,
+  MutationListenerReturnable<V, dynamic>? onMutate,
   List<Object?>? keys,
 }) {
   final context = useContext();
@@ -37,9 +37,9 @@ Mutation<T, V> useMutation<T extends Object, V>({
 
   final disposeMutation = useCallback(() {
     mutation.value.unmount(uKey);
-    if (onData != null) mutation.value.onDataListeners.remove(onData);
-    if (onError != null) mutation.value.onErrorListeners.remove(onError);
-    if (onMutate != null) mutation.value.onMutateListeners.remove(onMutate);
+    if (onData != null) mutation.value.removeDataListener(onData);
+    if (onError != null) mutation.value.removeErrorListener(onError);
+    if (onMutate != null) mutation.value.removeMutateListener(onMutate);
   }, [mutation.value, onData, onError, onMutate]);
 
   final oldJob = usePrevious(job);
@@ -58,16 +58,16 @@ Mutation<T, V> useMutation<T extends Object, V>({
       init();
     } else {
       if (oldOnData != onData && oldOnData != null) {
-        mutation.value.onDataListeners.remove(oldOnData);
-        if (onData != null) mutation.value.onDataListeners.add(onData);
+        mutation.value.removeDataListener(oldOnData);
+        if (onData != null) mutation.value.addDataListener(onData);
       }
       if (oldOnError != onError && oldOnError != null) {
-        mutation.value.onErrorListeners.remove(oldOnError);
-        if (onError != null) mutation.value.onErrorListeners.add(onError);
+        mutation.value.removeErrorListener(oldOnError);
+        if (onError != null) mutation.value.addErrorListener(onError);
       }
       if (oldOnMutate != onMutate && oldOnMutate != null) {
-        mutation.value.onMutateListeners.remove(oldOnMutate);
-        if (onMutate != null) mutation.value.onMutateListeners.add(onMutate);
+        mutation.value.removeMutateListener(oldOnMutate);
+        if (onMutate != null) mutation.value.addMutateListener(onMutate);
       }
     }
     return null;

packages/fl_query_hooks/lib/src/use_query.dart

@@ -52,8 +52,8 @@ Query<T, Outside> useQuery<T extends Object, Outside>({
 
   final disposeQuery = useCallback(() {
     query.value.unmount(uKey);
-    if (onData != null) query.value.onDataListeners.remove(onData);
-    if (onError != null) query.value.onErrorListeners.remove(onError);
+    if (onData != null) query.value.removeDataListener(onData);
+    if (onError != null) query.value.removeErrorListener(onError);
   }, [query.value, onData, onError, uKey]);
 
   useEffect(() {
@@ -85,16 +85,16 @@ Query<T, Outside> useQuery<T extends Object, Outside>({
             ?.setExternalData(externalData);
       }
 
-      if (hasOnDataChanged) query.value.onDataListeners.remove(oldOnData);
-      if (hasOnErrorChanged) query.value.onErrorListeners.remove(oldOnError);
+      if (hasOnDataChanged) query.value.removeDataListener(oldOnData);
+      if (hasOnErrorChanged) query.value.removeErrorListener(oldOnError);
     } else {
       if (hasOnDataChanged) {
-        query.value.onDataListeners.remove(oldOnData);
-        if (onData != null) query.value.onDataListeners.add(onData);
+        query.value.removeDataListener(oldOnData);
+        if (onData != null) query.value.addDataListener(onData);
       }
       if (hasOnErrorChanged) {
-        query.value.onErrorListeners.remove(oldOnError);
-        if (onError != null) query.value.onErrorListeners.add(onError);
+        query.value.removeErrorListener(oldOnError);
+        if (onError != null) query.value.addErrorListener(onError);
       }
     }
     return null;