feat(datastore) selective sync (#959)

Author: richardmcclellan

aws-datastore/src/androidTest/java/com/amplifyframework/datastore/AppSyncClientInstrumentationTest.java

@@ -206,15 +206,16 @@ public void testAllOperations() throws AmplifyException {
         // Run sync on Blogs
         // TODO: This is currently a pretty worthless test - mainly for setting a debug point and manually inspecting
         // When you call sync with a null lastSync it gives only one entry per object (the latest state)
-        Iterable<ModelWithMetadata<Blog>> blogSyncResult = sync(api.buildSyncRequest(blogSchema, null, 1000));
+        Iterable<ModelWithMetadata<Blog>> blogSyncResult =
+                sync(api.buildSyncRequest(blogSchema, null, 1000, QueryPredicates.all()));
         assertTrue(blogSyncResult.iterator().hasNext());
 
         // Run sync on Posts
         // TODO: This is currently a pretty worthless test - mainly for setting a debug point and manually inspecting
         // When you call sync with a lastSyncTime it gives you one entry per version of that object which was created
         // since that time.
         Iterable<ModelWithMetadata<Post>> postSyncResult =
-                sync(api.buildSyncRequest(postSchema, startTimeSeconds, 1000));
+                sync(api.buildSyncRequest(postSchema, startTimeSeconds, 1000, QueryPredicates.all()));
         assertTrue(postSyncResult.iterator().hasNext());
     }
 

aws-datastore/src/main/java/com/amplifyframework/datastore/DataStoreConfiguration.java

@@ -21,10 +21,14 @@
 import androidx.annotation.VisibleForTesting;
 import androidx.core.util.ObjectsCompat;
 
+import com.amplifyframework.core.model.Model;
+
 import org.json.JSONException;
 import org.json.JSONObject;
 
+import java.util.HashMap;
 import java.util.Iterator;
+import java.util.Map;
 import java.util.Objects;
 import java.util.concurrent.TimeUnit;
 
@@ -44,21 +48,16 @@ public final class DataStoreConfiguration {
     private final DataStoreConflictHandler conflictHandler;
     private final Integer syncMaxRecords;
     private final Integer syncPageSize;
-    private Long syncIntervalInMinutes;
-
-    private DataStoreConfiguration(
-            DataStoreErrorHandler errorHandler,
-            DataStoreConflictHandler conflictHandler,
-            Long syncIntervalInMinutes,
-            Integer syncMaxRecords,
-            Integer syncPageSize) {
-        this.errorHandler = errorHandler;
-        this.conflictHandler = conflictHandler;
-        this.syncMaxRecords = syncMaxRecords;
-        this.syncPageSize = syncPageSize;
-        if (syncIntervalInMinutes != null) {
-            this.syncIntervalInMinutes = syncIntervalInMinutes;
-        }
+    private final Map<String, DataStoreSyncExpression> syncExpressions;
+    private final Long syncIntervalInMinutes;
+
+    private DataStoreConfiguration(Builder builder) {
+        this.errorHandler = builder.errorHandler;
+        this.conflictHandler = builder.conflictHandler;
+        this.syncMaxRecords = builder.syncMaxRecords;
+        this.syncPageSize = builder.syncPageSize;
+        this.syncIntervalInMinutes = builder.syncIntervalInMinutes;
+        this.syncExpressions = builder.syncExpressions;
     }
 
     /**
@@ -175,6 +174,16 @@ public Integer getSyncPageSize() {
         return this.syncPageSize;
     }
 
+    /**
+     * Returns the Map of all {@link DataStoreSyncExpression}s used to filter data received from AppSync, either during
+     * a sync or over the real-time subscription.
+     * @return the Map of all {@link DataStoreSyncExpression}s.
+     */
+    @NonNull
+    public Map<String, DataStoreSyncExpression> getSyncExpressions() {
+        return this.syncExpressions;
+    }
+
     @Override
     public boolean equals(@Nullable Object thatObject) {
         if (this == thatObject) {
@@ -196,7 +205,13 @@ public boolean equals(@Nullable Object thatObject) {
         if (!ObjectsCompat.equals(getSyncPageSize(), that.getSyncPageSize())) {
             return false;
         }
-        return ObjectsCompat.equals(getSyncIntervalInMinutes(), that.getSyncIntervalInMinutes());
+        if (!ObjectsCompat.equals(getSyncIntervalInMinutes(), that.getSyncIntervalInMinutes())) {
+            return false;
+        }
+        if (!ObjectsCompat.equals(getSyncExpressions(), that.getSyncExpressions())) {
+            return false;
+        }
+        return true;
     }
 
     @Override
@@ -206,6 +221,7 @@ public int hashCode() {
         result = 31 * result + (getSyncMaxRecords() != null ? getSyncMaxRecords().hashCode() : 0);
         result = 31 * result + (getSyncPageSize() != null ? getSyncPageSize().hashCode() : 0);
         result = 31 * result + (getSyncIntervalInMinutes() != null ? getSyncIntervalInMinutes().hashCode() : 0);
+        result = 31 * result + (getSyncExpressions() != null ? getSyncExpressions().hashCode() : 0);
         return result;
     }
 
@@ -217,6 +233,7 @@ public String toString() {
             ", syncMaxRecords=" + syncMaxRecords +
             ", syncPageSize=" + syncPageSize +
             ", syncIntervalInMinutes=" + syncIntervalInMinutes +
+            ", syncExpressions=" + syncExpressions +
             '}';
     }
 
@@ -230,13 +247,15 @@ public static final class Builder {
         private Long syncIntervalInMinutes;
         private Integer syncMaxRecords;
         private Integer syncPageSize;
+        private Map<String, DataStoreSyncExpression> syncExpressions;
         private boolean ensureDefaults;
         private JSONObject pluginJson;
         private DataStoreConfiguration userProvidedConfiguration;
 
         private Builder() {
             this.errorHandler = DefaultDataStoreErrorHandler.instance();
             this.conflictHandler = DataStoreConflictHandler.alwaysApplyRemote();
+            this.syncExpressions = new HashMap<>();
             this.ensureDefaults = false;
         }
 
@@ -305,6 +324,23 @@ public Builder syncPageSize(@IntRange(from = 0) Integer syncPageSize) {
             return Builder.this;
         }
 
+        /**
+         * Sets a sync expression for a particular model to filter which data is synced locally.  The expression
+         * is evaluated each time DataStore is started.  The QueryPredicate is applied on both sync and subscriptions.
+         * @param modelClass the model class for which the filter applies
+         * @param syncExpression DataStoreSyncExpression that should be used to filter the data that is synced.
+         * @return Current builder
+         */
+        @NonNull
+        public Builder syncExpression(@NonNull Class<? extends Model> modelClass,
+                                      @NonNull DataStoreSyncExpression syncExpression) {
+            this.syncExpressions.put(
+                    Objects.requireNonNull(modelClass).getSimpleName(),
+                    Objects.requireNonNull(syncExpression)
+            );
+            return Builder.this;
+        }
+
         private void populateSettingsFromJson() throws DataStoreException {
             if (pluginJson == null) {
                 return;
@@ -356,6 +392,7 @@ private void applyUserProvidedConfiguration() {
                 syncIntervalInMinutes);
             syncMaxRecords = getValueOrDefault(userProvidedConfiguration.getSyncMaxRecords(), syncMaxRecords);
             syncPageSize = getValueOrDefault(userProvidedConfiguration.getSyncPageSize(), syncPageSize);
+            syncExpressions = userProvidedConfiguration.getSyncExpressions();
         }
 
         private static <T> T getValueOrDefault(T value, T defaultValue) {
@@ -383,13 +420,7 @@ public DataStoreConfiguration build() throws DataStoreException {
                 syncMaxRecords = getValueOrDefault(syncMaxRecords, DEFAULT_SYNC_MAX_RECORDS);
                 syncPageSize = getValueOrDefault(syncPageSize, DEFAULT_SYNC_PAGE_SIZE);
             }
-            return new DataStoreConfiguration(
-                errorHandler,
-                conflictHandler,
-                syncIntervalInMinutes,
-                syncMaxRecords,
-                syncPageSize
-            );
+            return new DataStoreConfiguration(this);
         }
     }
 

aws-datastore/src/main/java/com/amplifyframework/datastore/DataStoreSyncExpression.java

@@ -0,0 +1,36 @@
+/*
+ * Copyright 2020 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License").
+ * You may not use this file except in compliance with the License.
+ * A copy of the License is located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ * or in the "license" file accompanying this file. This file is distributed
+ * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ * express or implied. See the License for the specific language governing
+ * permissions and limitations under the License.
+ */
+
+package com.amplifyframework.datastore;
+
+import com.amplifyframework.core.Action;
+import com.amplifyframework.core.Consumer;
+import com.amplifyframework.core.model.Model;
+import com.amplifyframework.core.model.query.predicate.QueryPredicate;
+
+/**
+ * Used to specify a QueryPredicate for a {@link Model} that will filter what is synced to the client.  A DataStore
+ * customer should provide their implementation to the {@link DataStoreConfiguration} while
+ * constructing the DataStore plugin using {@link AWSDataStorePlugin#AWSDataStorePlugin(DataStoreConfiguration)}.
+ */
+public interface DataStoreSyncExpression {
+    /**
+     * This will be called each time DataStore is started.  This allows the QueryPredicate to be modified at runtime,
+     * by calling {@link DataStoreCategoryBehavior#stop(Action, Consumer)} followed by
+     * {@link DataStoreCategoryBehavior#start(Action, Consumer)}.
+     * @return QueryPredicate to filter what is synced.
+     */
+    QueryPredicate resolvePredicate();
+}

aws-datastore/src/main/java/com/amplifyframework/datastore/appsync/AppSync.java

@@ -54,14 +54,16 @@ public interface AppSync {
      * @param modelSchema The schema of the Model we are listening on
      * @param lastSync The time you last synced - all changes since this time are retrieved.
      * @param syncPageSize limit for number of records to return per page.
+     * @param queryPredicate QueryPredicate to filter the records returned.
      * @return A {@link GraphQLRequest} for making a sync query
      * @throws DataStoreException on error building GraphQLRequest due to inability to obtain model schema.
      */
     @NonNull
     <T extends Model> GraphQLRequest<PaginatedResult<ModelWithMetadata<T>>> buildSyncRequest(
             @NonNull ModelSchema modelSchema,
             @Nullable Long lastSync,
-            @Nullable Integer syncPageSize
+            @Nullable Integer syncPageSize,
+            @NonNull QueryPredicate queryPredicate
     ) throws DataStoreException;
 
     /**

aws-datastore/src/main/java/com/amplifyframework/datastore/appsync/AppSyncClient.java

@@ -74,8 +74,9 @@ public static AppSyncClient via(@NonNull GraphQLBehavior api) {
     public <T extends Model> GraphQLRequest<PaginatedResult<ModelWithMetadata<T>>> buildSyncRequest(
             @NonNull ModelSchema modelSchema,
             @Nullable Long lastSync,
-            @Nullable Integer syncPageSize) throws DataStoreException {
-        return AppSyncRequestFactory.buildSyncRequest(modelSchema, lastSync, syncPageSize);
+            @Nullable Integer syncPageSize,
+            @NonNull QueryPredicate queryPredicate) throws DataStoreException {
+        return AppSyncRequestFactory.buildSyncRequest(modelSchema, lastSync, syncPageSize, queryPredicate);
     }
 
     @NonNull

aws-datastore/src/main/java/com/amplifyframework/datastore/appsync/AppSyncRequestFactory.java

@@ -80,7 +80,8 @@ private AppSyncRequestFactory() {}
     static <T> AppSyncGraphQLRequest<T> buildSyncRequest(
             @NonNull final ModelSchema modelSchema,
             @Nullable final Long lastSync,
-            @Nullable final Integer limit)
+            @Nullable final Integer limit,
+            @NonNull final QueryPredicate predicate)
             throws DataStoreException {
         try {
             AppSyncGraphQLRequest.Builder builder = AppSyncGraphQLRequest.builder()
@@ -99,7 +100,10 @@ static <T> AppSyncGraphQLRequest<T> buildSyncRequest(
             if (limit != null) {
                 builder.variable("limit", "Int", limit);
             }
-
+            if (!QueryPredicates.all().equals(predicate)) {
+                String filterType = "Model" + Casing.capitalizeFirst(modelSchema.getName()) + "FilterInput";
+                builder.variable("filter", filterType, parsePredicate(predicate));
+            }
             return builder.build();
         } catch (AmplifyException amplifyException) {
             throw new DataStoreException("Failed to get fields for model.",
@@ -170,7 +174,6 @@ static <M extends Model> AppSyncGraphQLRequest<ModelWithMetadata<M>> buildCreati
             throw new DataStoreException("Failed to get fields for model.",
                     amplifyException, "Validate your model file.");
         }
-
     }
 
     static Map<String, Object> parsePredicate(QueryPredicate queryPredicate) throws DataStoreException {

aws-datastore/src/main/java/com/amplifyframework/datastore/syncengine/Orchestrator.java

@@ -60,6 +60,7 @@ public final class Orchestrator {
     private final SubscriptionProcessor subscriptionProcessor;
     private final SyncProcessor syncProcessor;
     private final MutationProcessor mutationProcessor;
+    private final QueryPredicateProvider queryPredicateProvider;
     private final StorageObserver storageObserver;
     private final Supplier<Mode> targetMode;
     private final AtomicReference<Mode> currentMode;
@@ -103,6 +104,7 @@ public Orchestrator(
         Merger merger = new Merger(mutationOutbox, versionRepository, localStorageAdapter);
         SyncTimeRegistry syncTimeRegistry = new SyncTimeRegistry(localStorageAdapter);
         ConflictResolver conflictResolver = new ConflictResolver(dataStoreConfigurationProvider, appSync);
+        this.queryPredicateProvider = new QueryPredicateProvider(dataStoreConfigurationProvider);
 
         this.mutationProcessor = MutationProcessor.builder()
             .merger(merger)
@@ -119,8 +121,14 @@ public Orchestrator(
             .appSync(appSync)
             .merger(merger)
             .dataStoreConfigurationProvider(dataStoreConfigurationProvider)
+            .queryPredicateProvider(queryPredicateProvider)
             .build();
-        this.subscriptionProcessor = new SubscriptionProcessor(appSync, modelProvider, merger);
+        this.subscriptionProcessor = SubscriptionProcessor.builder()
+                .appSync(appSync)
+                .modelProvider(modelProvider)
+                .merger(merger)
+                .queryPredicateProvider(queryPredicateProvider)
+                .build();
         this.storageObserver = new StorageObserver(localStorageAdapter, mutationOutbox);
         this.currentMode = new AtomicReference<>(Mode.STOPPED);
         this.targetMode = targetMode;
@@ -332,6 +340,11 @@ private Completable startApiSync() {
         return Completable.create(emitter -> {
             LOG.info("Starting API synchronization mode.");
 
+            // Resolve any client provided DataStoreSyncExpressions, before starting sync and subscriptions, once each
+            // time DataStore starts.  The QueryPredicateProvider caches the resolved QueryPredicates, which are then
+            // used to filter data received from AppSync.
+            queryPredicateProvider.resolvePredicates();
+
             subscriptionProcessor.startSubscriptions();
 
             LOG.debug("About to hydrate...");

aws-datastore/src/main/java/com/amplifyframework/datastore/syncengine/QueryPredicateProvider.java

@@ -0,0 +1,80 @@
+/*
+ * Copyright 2020 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License").
+ * You may not use this file except in compliance with the License.
+ * A copy of the License is located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ * or in the "license" file accompanying this file. This file is distributed
+ * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ * express or implied. See the License for the specific language governing
+ * permissions and limitations under the License.
+ */
+
+package com.amplifyframework.datastore.syncengine;
+
+import androidx.annotation.NonNull;
+import androidx.core.util.Pair;
+
+import com.amplifyframework.core.model.Model;
+import com.amplifyframework.core.model.query.predicate.QueryPredicate;
+import com.amplifyframework.core.model.query.predicate.QueryPredicates;
+import com.amplifyframework.datastore.DataStoreConfiguration;
+import com.amplifyframework.datastore.DataStoreConfigurationProvider;
+import com.amplifyframework.datastore.DataStoreException;
+import com.amplifyframework.datastore.DataStoreSyncExpression;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+
+import io.reactivex.rxjava3.core.Observable;
+
+/**
+ * Maintains a Map of QueryPredicates for each {@link Model} for the current DataStore session.
+ */
+final class QueryPredicateProvider {
+    private final DataStoreConfigurationProvider dataStoreConfigurationProvider;
+    private final Map<String, QueryPredicate> predicateMap = new HashMap<>();
+
+    /**
+     * Constructs a QueryPredicateProvider.
+     * @param dataStoreConfigurationProvider a DataStoreConfigurationProvider.
+     */
+    QueryPredicateProvider(DataStoreConfigurationProvider dataStoreConfigurationProvider) {
+        this.dataStoreConfigurationProvider = dataStoreConfigurationProvider;
+    }
+
+    /**
+     * Evaluates any client provided {@link DataStoreSyncExpression}s and caches the resolved {@link QueryPredicate}s
+     * for use in the current DataStore session. These are used to filter data received from AppSync, either during a
+     * sync or over the real-time subscription.  This is called once each time DataStore is started.
+     *
+     * @throws DataStoreException on error obtaining the {@link DataStoreConfiguration}.
+     */
+    public void resolvePredicates() throws DataStoreException {
+        Map<String, DataStoreSyncExpression> expressions =
+                dataStoreConfigurationProvider.getConfiguration().getSyncExpressions();
+        predicateMap.clear();
+        predicateMap.putAll(Observable.fromIterable(expressions.entrySet())
+                .map(entry -> Pair.create(entry.getKey(), entry.getValue().resolvePredicate()))
+                .toMap(pair -> pair.first, pair -> pair.second)
+                .blockingGet());
+    }
+
+    /**
+     * Returns the {@link QueryPredicate} for the given modelName.
+     * @param modelName name of the {@link Model}.
+     * @return the {@link QueryPredicate} for the given modelName.
+     */
+    @NonNull
+    public QueryPredicate getPredicate(@NonNull String modelName) {
+        QueryPredicate predicate = predicateMap.get(Objects.requireNonNull(modelName));
+        if (predicate == null) {
+            predicate = QueryPredicates.all();
+        }
+        return predicate;
+    }
+}