Support for doc_ids filter in Pull filtered replication. (#582)

* Support for doc_ids filter in Pull filtered replication.
Fixes #578

Author: jjrodrig

CHANGES.md

@@ -1,6 +1,7 @@
 # Unreleased
 - [IMPROVED] Forced a TLS1.2 `SSLSocketFactory` where possible on Android API versions < 20 (it is
   already enabled by default on newer API levels).
+- [NEW] Added API for specifying a list of document IDs in the filtered pull replicator
 
 # 2.2.0 (2018-02-14)
 - [NEW] Added API for specifying a mango selector in the filtered pull replicator

cloudant-sync-datastore-core/src/main/java/com/cloudant/sync/internal/mazha/CouchClient.java

@@ -446,7 +446,7 @@ private Map<String, Object> getParametrizedChangeFeedOptions(Object since, Integ
 
     public ChangesResult changes(Object since, Integer limit) {
         Map<String, Object> options = getParametrizedChangeFeedOptions(since, limit);
-        return this.changes(options);
+        return this.changesRequestWithGet(options);
     }
 
     public ChangesResult changes(PullFilter filter, Object since, Integer limit) {
@@ -461,22 +461,41 @@ public ChangesResult changes(PullFilter filter, Object since, Integer limit) {
                 }
             }
         }
-        return this.changes(options);
+        return this.changesRequestWithGet(options);
     }
 
     public ChangesResult changes(String selector, Object since, Integer limit) {
-        Misc.checkNotNullOrEmpty(selector,null);
+        Misc.checkNotNullOrEmpty(selector, null);
+
         Map<String, Object> options = getParametrizedChangeFeedOptions(since, limit);
         options.put("filter", "_selector");
+
+        return changesRequestWithPost(selector, options);
+    }
+
+    public ChangesResult changes(List<String> docIds, Object since, Integer limit) {
+        Misc.checkState((docIds != null && !docIds.isEmpty()), null);
+
+        Map<String, Object> options = getParametrizedChangeFeedOptions(since, limit);
+        options.put("filter", "_doc_ids");
+
+        Map<String, Object> docIdsMap = new HashMap<String, Object>();
+        docIdsMap.put("doc_ids", docIds);
+        String docsIdsDoc = JSONUtils.serializeAsString(docIdsMap);
+
+        return changesRequestWithPost(docsIdsDoc, options);
+    }
+
+    private ChangesResult changesRequestWithGet(final Map<String, Object> options) {
         URI changesFeedUri = uriHelper.changesUri(options);
-        HttpConnection connection = Http.POST(changesFeedUri, "application/json");
-        connection.setRequestBody(selector);
+        HttpConnection connection = Http.GET(changesFeedUri);
         return executeToJsonObjectWithRetry(connection, ChangesResult.class);
     }
 
-    public ChangesResult changes(final Map<String, Object> options) {
+    private ChangesResult changesRequestWithPost( String body, final Map<String, Object> options) {
         URI changesFeedUri = uriHelper.changesUri(options);
-        HttpConnection connection = Http.GET(changesFeedUri);
+        HttpConnection connection = Http.POST(changesFeedUri, "application/json");
+        connection.setRequestBody(body);
         return executeToJsonObjectWithRetry(connection, ChangesResult.class);
     }
 

cloudant-sync-datastore-core/src/main/java/com/cloudant/sync/internal/replication/CouchClientWrapper.java

@@ -150,6 +150,15 @@ public ChangesResult changes(String selector, Object lastSequence, int limit) {
         }
     }
 
+    @Override
+    public ChangesResult changes(List<String> docIds, Object lastSequence, int limit) {
+        if (docIds == null || docIds.isEmpty()) {
+            return couchClient.changes(lastSequence, limit);
+        } else {
+            return couchClient.changes(docIds, lastSequence, limit);
+        }
+    }
+
     @Override
     public Iterable<DocumentRevsList> bulkGetRevisions(List<BulkGetRequest> requests,
                                                        boolean pullAttachmentsInline) {

cloudant-sync-datastore-core/src/main/java/com/cloudant/sync/internal/replication/CouchDB.java

@@ -50,6 +50,7 @@ interface CouchDB {
     ChangesResult changes(Object lastSequence, int limit);
     ChangesResult changes(PullFilter filter, Object lastSequence, int limit);
     ChangesResult changes(String selector, Object lastSequence, int limit);
+    ChangesResult changes(List<String> docIds, Object lastSequence, int limit);
     List<DocumentRevs> getRevisions(String documentId,
                                            Collection<String> revisionIds,
                                            Collection<String> attsSince,

cloudant-sync-datastore-core/src/main/java/com/cloudant/sync/internal/replication/PullStrategy.java

@@ -42,6 +42,7 @@
 import java.net.URI;
 import java.nio.charset.Charset;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
@@ -81,6 +82,7 @@ private static class State {
 
     PullFilter filter;
     String selector;
+    List<String> docIds;
 
     DatastoreWrapper targetDb;
 
@@ -101,10 +103,15 @@ public PullStrategy(URI source,
                         Database target,
                         PullFilter filter,
                         String selector,
+                        List<String> docIds,
                         List<HttpConnectionRequestInterceptor> requestInterceptors,
                         List<HttpConnectionResponseInterceptor> responseInterceptors) {
         this.filter = filter;
         this.selector = selector;
+        this.docIds = docIds;
+        if (docIds != null && !docIds.isEmpty()) {
+            Collections.sort(docIds);
+        }
         this.sourceDb = new CouchClientWrapper(new CouchClient(source, requestInterceptors,
                 responseInterceptors));
         this.targetDb = new DatastoreWrapper((DatabaseImpl) target);
@@ -114,6 +121,10 @@ public PullStrategy(URI source,
                     filter.getName());
         } else if (selector != null) {
             replicatorName = String.format("%s <-- %s (%s)", target.getPath(), source, selector);
+        } else if (docIds != null && !docIds.isEmpty()) {
+            String concatenatedIds = Misc.join(",",docIds);
+            replicatorName = String.format("%s <-- %s (%s)", target.getPath(), source,
+                    concatenatedIds);
         } else {
             replicatorName = String.format("%s <-- %s ", target.getPath(), source);
         }
@@ -444,6 +455,8 @@ public String getReplicationId() throws DocumentStoreException {
             dict.put("filter", this.filter.toQueryString());
         } else if (selector != null) {
             dict.put("selector", this.selector);
+        } else if (docIds != null && !docIds.isEmpty()) {
+            dict.put("docIds", Misc.join(",",docIds));
         }
         // get raw SHA-1 of dictionary
         try {
@@ -467,6 +480,11 @@ private ChangesResultWrapper nextBatch() throws DocumentStoreException {
                     this.selector,
                     lastCheckpoint,
                     this.changeLimitPerBatch);
+        } else if (this.docIds != null && !this.docIds.isEmpty()) {
+            changeFeeds = this.sourceDb.changes(
+                    this.docIds,
+                    lastCheckpoint,
+                    this.changeLimitPerBatch);
         } else {
             changeFeeds = this.sourceDb.changes(
                     this.filter,

cloudant-sync-datastore-core/src/main/java/com/cloudant/sync/replication/PullFilter.java

@@ -74,8 +74,7 @@ public PullFilter(String filterName) {
     }
 
     /**
-     * Constructs a filter object for a function that requires no
-     * parameters.
+     * Constructs a filter object for a function that requires parameters.
      *
      *
      *
@@ -86,7 +85,7 @@ public PullFilter(String filterName) {
      *                   and the name of the filter function, separated by a slash. For example,
      *                   {@code filterDoc/filterFunctionName}
      *
-     * @param parameters Any parameters required for the function. Can be {@code null}. The contents
+     * @param parameters Any parameters required for the function. Must not be {@code null}. The contents
      *                   of {@code properties} are expanded to {@code key=value} pairs when
      *                   constructing the {@code _changes} feed call for the remote database.
      *                   Integer values should be added as String objects.

cloudant-sync-datastore-core/src/main/java/com/cloudant/sync/replication/ReplicatorBuilder.java

@@ -259,6 +259,8 @@ public static class Pull extends ReplicatorBuilder<URI, DocumentStore, Pull> {
 
         private String pullPullSelector = null;
 
+        private List<String> pullDocIds = null;
+
         private int changeLimitPerBatch = 1000;
 
         private int insertBatchSize = 100;
@@ -272,6 +274,10 @@ public Replicator build() {
                     "Source and target cannot be null");
             Misc.checkState(this.pullPullFilter == null || this.pullPullSelector == null,
                     "Filter and selector cannot be defined at the same time");
+            if (this.pullPullFilter != null || this.pullPullSelector != null) {
+                Misc.checkState(this.pullDocIds == null || this.pullDocIds.isEmpty(),
+                        "Doc Ids cannot be provided at the same time than selecto or filter");
+            }
 
             // add cookie interceptor and remove creds from URI if required
             super.source = super.addAuthInterceptorIfRequired(super.source);
@@ -280,6 +286,7 @@ public Replicator build() {
                     super.target.database(),
                     pullPullFilter,
                     pullPullSelector,
+                    pullDocIds,
                     super.requestInterceptors,
                     super.responseInterceptors);
 
@@ -321,6 +328,22 @@ public Pull selector(String pullPullSelector) {
             return this;
         }
 
+        /**
+         * <p>Sets the list of doc IDs to use as filtering criteria when a pull replication calls the
+         * source database's {@code _changes} feed.
+         * </p>
+         * Note: Doc IDs filtering are supported only when replicating against a CouchDB 2.x compliant database
+         *
+         * @see
+         * <a target="_blank" href="https://console.bluemix.net/docs/services/Cloudant/api/database.html#get-changes">See doc_ids filtering</a>
+         *
+         * @param docIds - List of document Ids
+         * @return This instance of {@link ReplicatorBuilder}
+         */
+        public Pull docIds(List<String> docIds) {
+            this.pullDocIds = docIds;
+            return this;
+        }
         /**
          * Sets the number of changes to fetch from the _changes feed per batch
          *

cloudant-sync-datastore-core/src/test/java/com/cloudant/sync/internal/replication/PullReplicatorTest.java

@@ -29,6 +29,7 @@
 
 import java.net.URI;
 import java.net.URISyntaxException;
+import java.util.Arrays;
 import java.util.HashMap;
 import java.util.Map;
 
@@ -124,6 +125,19 @@ public void testPullReplicationCreatedSuccessfullyWithSelector() throws Exceptio
         Assert.assertNotNull(replicator);
     }
 
+    @Test
+    public void testPullReplicationCreatedSuccessfullyWithDocIds() throws Exception {
+
+        Replicator replicator = ReplicatorBuilder.pull()
+                .from(this.source)
+                .to(this.documentStore)
+                .docIds(Arrays.asList("id1","id2"))
+                .build();
+
+        Assert.assertNotNull(replicator);
+    }
+
+
     @Test(expected = IllegalStateException.class)
     public void testPullReplicationSelectorAndFilterIncompatible() throws Exception {
 
@@ -136,6 +150,29 @@ public void testPullReplicationSelectorAndFilterIncompatible() throws Exception
 
     }
 
+    @Test(expected = IllegalStateException.class)
+    public void testPullReplicationSelectorAndDocIdIncompatible() throws Exception {
+
+        Replicator replicator = ReplicatorBuilder.pull()
+                .from(this.source)
+                .to(this.documentStore)
+                .selector("{\"selector\":{\"class\":\"a_class\"}}")
+                .docIds(Arrays.asList("id1"))
+                .build();
+
+    }
+
+    @Test(expected = IllegalStateException.class)
+    public void testPullReplicationFilterAndDocIdIncompatible() throws Exception {
+
+        Replicator replicator = ReplicatorBuilder.pull()
+                .from(this.source)
+                .to(this.documentStore)
+                .filter(new PullFilter("a_filter"))
+                .docIds(Arrays.asList("id1"))
+                .build();
+    }
+
     @Test
     public void testPullReplicationCreatedSuccessfullyWithFilter() throws Exception {
 

cloudant-sync-datastore-core/src/test/java/com/cloudant/sync/internal/replication/PullStrategyDocIdTest.java

@@ -0,0 +1,77 @@
+/*
+ * Copyright © 2018 IBM Corp. 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. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the
+ * License 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.cloudant.sync.internal.replication;
+
+import com.cloudant.sync.internal.mazha.AnimalDb;
+import com.cloudant.sync.internal.mazha.ClientTestUtils;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.Arrays;
+import java.util.List;
+
+public class PullStrategyDocIdTest extends ReplicationTestBase {
+
+    // we use this utility method rather than ReplicationTestBase.pull() because some
+    // methods want to make assertions on the BasicPullStrategy after running the replication
+    private void pull(PullStrategy replicator, int expectedDocs) throws Exception {
+        TestStrategyListener listener = new TestStrategyListener();
+        replicator.getEventBus().register(listener);
+        replicator.run();
+        listener.assertReplicationCompletedOrThrow();
+        Assert.assertEquals(expectedDocs, listener.documentsReplicated);
+    }
+
+    @Test
+    public void pull_filterDocIdsFromAnimalDb_twoDocShouldBePulled() throws Exception {
+        org.junit.Assume.assumeTrue(ClientTestUtils.isCouchDBV2(remoteDb.couchClient.getRootUri()));
+        List<String> docIds = Arrays.asList("snipe","kookaburra");
+        PullStrategy replicator = super.getPullStrategy(docIds);
+
+        Assert.assertEquals(0, datastore.getDocumentCount());
+
+        AnimalDb.populate(remoteDb.couchClient);
+        this.pull(replicator, 2);
+
+        Assert.assertEquals(2, datastore.getDocumentCount());
+
+        for (String bird : docIds) {
+            Assert.assertTrue(datastore.contains(bird));
+        }
+    }
+
+    @Test
+    public void
+    pull_filterSelectorMammalFromAnimalDbUsingParameterizedFilter_eightDocShouldBePulled()
+            throws Exception {
+        org.junit.Assume.assumeTrue(ClientTestUtils.isCouchDBV2(remoteDb.couchClient.getRootUri()));
+        List<String> docIds = Arrays.asList("aardvark", "badger", "elephant", "giraffe", "lemur", "llama",
+                "panda", "zebra");
+        PullStrategy replicator = super.getPullStrategy(docIds);
+
+        Assert.assertEquals(0, datastore.getDocumentCount());
+
+        AnimalDb.populate(remoteDb.couchClient);
+        this.pull(replicator, 8);
+
+        Assert.assertEquals(8, datastore.getDocumentCount());
+
+        for (String mammal : docIds) {
+            Assert.assertTrue(datastore.contains(mammal));
+        }
+    }
+
+}

cloudant-sync-datastore-core/src/test/java/com/cloudant/sync/internal/replication/ReplicationTestBase.java

@@ -202,6 +202,22 @@ protected ReplicatorBuilder.Pull getPullBuilder(String selector) {
         return pull;
     }
 
+    protected ReplicatorBuilder.Pull getPullBuilder(List<String> docIds) {
+        ReplicatorBuilder.Pull pull = ReplicatorBuilder.pull().
+                from(this.couchConfig.getRootUri()).
+                to(this.documentStore)
+                .docIds(docIds)
+                .addRequestInterceptors(couchConfig.getRequestInterceptors(false))
+                .addResponseInterceptors(couchConfig.getResponseInterceptors(false));
+        if (couchConfig.getUsername() != null && couchConfig.getPassword() != null) {
+
+            pull.username(couchConfig.getUsername())
+                    .password(couchConfig.getPassword());
+        }
+
+        return pull;
+    }
+
     protected PushStrategy getPushStrategy() {
         return (PushStrategy)((ReplicatorImpl)this.getPushBuilder().build()).strategy;
     }
@@ -218,6 +234,10 @@ protected PullStrategy getPullStrategy(String selector) {
         return (PullStrategy)((ReplicatorImpl)this.getPullBuilder(selector).build()).strategy;
     }
 
+    protected PullStrategy getPullStrategy(List<String> docIds) {
+        return (PullStrategy)((ReplicatorImpl)this.getPullBuilder(docIds).build()).strategy;
+    }
+
     protected PushResult push() throws Exception {
         TestStrategyListener listener = new TestStrategyListener();
         PushStrategy replicator = this.getPushStrategy();