feat(datastore): Add QueryPredicate to Observe (#1332)

Author: fjnoyp

packages/amplify_datastore/example/integration_test/observe_test.dart

@@ -75,5 +75,51 @@ void main() {
       await Amplify.DataStore.save(updatedBlog);
       await Amplify.DataStore.delete(updatedBlog);
     });
+
+    testWidgets('observe with query predicates returns all matches',
+        (WidgetTester tester) async {
+      Blog blog = Blog(name: 'blog');
+      Blog updatedBlog = blog.copyWith(name: 'updated blog');
+
+      var eventItemStream = Amplify.DataStore.observe(Blog.classType,
+              where: Blog.NAME.ne("not a blog name"))
+          .map((event) => event.item);
+      expectLater(
+        eventItemStream,
+        emitsInOrder(
+          [
+            blog,
+            updatedBlog,
+            updatedBlog,
+          ],
+        ),
+      );
+
+      await Amplify.DataStore.save(blog);
+      await Amplify.DataStore.save(updatedBlog);
+      await Amplify.DataStore.delete(updatedBlog);
+    });
+
+    testWidgets('observe with query predicates filters out non matches',
+        (WidgetTester tester) async {
+      Blog blog = Blog(name: 'matching blog');
+      Blog updatedBlog = blog.copyWith(name: 'updated blog');
+      Blog otherBlog = Blog(name: 'matching blog 2');
+
+      var eventItemStream = Amplify.DataStore.observe(Blog.classType,
+              where: Blog.NAME.contains("matching"))
+          .map((event) => event.item);
+      expectLater(
+        eventItemStream,
+        emitsInOrder(
+          [blog, otherBlog],
+        ),
+      );
+
+      await Amplify.DataStore.save(blog);
+      await Amplify.DataStore.save(updatedBlog);
+      await Amplify.DataStore.delete(updatedBlog);
+      await Amplify.DataStore.save(otherBlog);
+    });
   });
 }

packages/amplify_datastore/lib/amplify_datastore.dart

@@ -122,9 +122,9 @@ class AmplifyDataStore extends DataStorePluginInterface {
   }
 
   @override
-  Stream<SubscriptionEvent<T>> observe<T extends Model>(
-      ModelType<T> modelType) {
-    return _instance.observe(modelType);
+  Stream<SubscriptionEvent<T>> observe<T extends Model>(ModelType<T> modelType,
+      {QueryPredicate? where}) {
+    return _instance.observe(modelType, where: where);
   }
 
   @override

packages/amplify_datastore/lib/method_channel_datastore.dart

@@ -178,8 +178,8 @@ class AmplifyDataStoreMethodChannel extends AmplifyDataStore {
   }
 
   @override
-  Stream<SubscriptionEvent<T>> observe<T extends Model>(
-      ModelType<T> modelType) async* {
+  Stream<SubscriptionEvent<T>> observe<T extends Model>(ModelType<T> modelType,
+      {QueryPredicate? where}) async* {
     await _setUpObserveIfNeeded();
 
     // Step #1. Open the event channel if it's not already open. Note
@@ -200,6 +200,7 @@ class AmplifyDataStoreMethodChannel extends AmplifyDataStore {
     // Step #3. Deserialize events and return new broadcast stream
     yield* filteredStream
         .map((event) => SubscriptionEvent.fromMap(event, modelType))
+        .where((event) => where == null || where.evaluate(event.item))
         .asBroadcastStream()
         .cast<SubscriptionEvent<T>>();
   }

packages/amplify_datastore_plugin_interface/lib/amplify_datastore_plugin_interface.dart

@@ -121,8 +121,8 @@ abstract class DataStorePluginInterface extends AmplifyPluginInterface {
     throw UnimplementedError('save() has not been implemented');
   }
 
-  Stream<SubscriptionEvent<T>> observe<T extends Model>(
-      ModelType<T> modelType) {
+  Stream<SubscriptionEvent<T>> observe<T extends Model>(ModelType<T> modelType,
+      {QueryPredicate? where}) {
     throw UnimplementedError('observe() has not been implemented.');
   }
 

packages/amplify_flutter/lib/src/categories/amplify_datastore_category.dart

@@ -92,10 +92,10 @@ class DataStoreCategory {
   }
 
   /// Observe changes on the specified [modelType].
-  Stream<SubscriptionEvent<T>> observe<T extends Model>(
-      ModelType<T> modelType) {
+  Stream<SubscriptionEvent<T>> observe<T extends Model>(ModelType<T> modelType,
+      {QueryPredicate? where}) {
     return plugins.length == 1
-        ? plugins[0].observe(modelType)
+        ? plugins[0].observe(modelType, where: where)
         : throw _pluginNotAddedException('DataStore');
   }