Merge pull request #1388 from couchbase/feature/post_replicate_issues

Fixed #1386 and andriod #939 : POST /_replicate to start multiple replicators and threading issues

Author: Hideki Itakura

src/main/java/com/couchbase/lite/Database.java

@@ -2415,40 +2415,27 @@ private boolean postChangeNotifications() {
 
     // Database+Replication
 
-    protected Replication getActiveReplicator(URL remote, boolean push) {
+    protected Replication findActiveReplicator(Replication replicator) {
         synchronized (activeReplicators) {
-            try {
-                java.net.URI remoteUri = remote.toURI();
-                for (Replication replicator : activeReplicators) {
-                    if (replicator.getRemoteUrl().toURI().equals(remoteUri) &&
-                            replicator.isPull() == !push && replicator.isRunning())
-                        return replicator;
-                }
-            } catch (java.net.URISyntaxException uriException) {
-                // Not possible since it would not be an active replicator.
-                // However, until we refactor everything to use java.net,
-                // I'm not sure we have a choice but to swallow this.
-                Log.e(Log.TAG_DATABASE, "Active replicator found with invalid URI", uriException);
+            String remoteCheckpointDocID = replicator.remoteCheckpointDocID();
+            if (remoteCheckpointDocID == null)
+                return null;
+
+            for (Replication r : activeReplicators) {
+                if (remoteCheckpointDocID.equals(r.remoteCheckpointDocID()) && r.isRunning())
+                    return r;
             }
         }
         return null;
     }
 
-    protected Replication getReplicator(URL remote,
-                                        HttpClientFactory httpClientFactory,
-                                        boolean push,
-                                        boolean continuous) {
-        Replication result = getActiveReplicator(remote, push);
-        if (result != null) {
-            return result;
-        }
-        if (push) {
-            result = new Replication(this, remote, Replication.Direction.PUSH, httpClientFactory);
-        } else {
-            result = new Replication(this, remote, Replication.Direction.PULL, httpClientFactory);
-        }
-        result.setContinuous(continuous);
-        return result;
+    protected Replication createReplicator(URL remote, boolean push, HttpClientFactory factory) {
+        Replication replicator;
+        if (push)
+            replicator = new Replication(this, remote, Replication.Direction.PUSH, factory);
+        else
+            replicator = new Replication(this, remote, Replication.Direction.PULL, factory);
+        return replicator;
     }
 
     // Database+LocalDocs

src/main/java/com/couchbase/lite/Manager.java

@@ -574,14 +574,6 @@ public synchronized Database getDatabase(String name, boolean mustExist) {
      */
     @InterfaceAudience.Private
     public Replication getReplicator(Map<String, Object> properties) throws CouchbaseLiteException {
-
-        // TODO: in the iOS equivalent of this code, there is: {@"doc_ids", _documentIDs}) - write unit test that detects this bug
-        // TODO: ditto for "headers"
-
-        Authorizer authorizer = null;
-        Replication repl = null;
-        URL remote = null;
-
         Map<String, Object> remoteMap;
 
         Map<String, Object> sourceMap = parseSourceOrTarget(properties, "source");
@@ -601,7 +593,7 @@ public Replication getReplicator(Map<String, Object> properties) throws Couchbas
 
         // Map the 'source' and 'target' JSON params to a local database and remote URL:
         if (source == null || target == null) {
-            throw new CouchbaseLiteException("source and target are both null", new Status(Status.BAD_REQUEST));
+            throw new CouchbaseLiteException("Source and target are both null", new Status(Status.BAD_REQUEST));
         }
 
         boolean push = false;
@@ -623,7 +615,7 @@ public Replication getReplicator(Map<String, Object> properties) throws Couchbas
                 db = getExistingDatabase(target);
             }
             if (db == null) {
-                throw new CouchbaseLiteException("database is null", new Status(Status.NOT_FOUND));
+                throw new CouchbaseLiteException("Database is null", new Status(Status.NOT_FOUND));
             }
             remoteMap = sourceMap;
         }
@@ -633,20 +625,21 @@ public Replication getReplicator(Map<String, Object> properties) throws Couchbas
             throw new CouchbaseLiteException("Can't specify both a filter and doc IDs",
                     new Status(Status.BAD_REQUEST));
 
+        URL remote = null;
         try {
             remote = new URL(remoteStr);
         } catch (MalformedURLException e) {
-            throw new CouchbaseLiteException("malformed remote url: " + remoteStr,
+            throw new CouchbaseLiteException("Malformed remote url: " + remoteStr,
                     new Status(Status.BAD_REQUEST));
         }
         if (remote == null) {
-            throw new CouchbaseLiteException("remote URL is null: " + remoteStr,
+            throw new CouchbaseLiteException("Remote URL is null: " + remoteStr,
                     new Status(Status.BAD_REQUEST));
         }
 
+        Authorizer authorizer = null;
         Map<String, Object> authMap = (Map<String, Object>) remoteMap.get("auth");
         if (authMap != null) {
-
             Map<String, Object> persona = (Map<String, Object>) authMap.get("persona");
             if (persona != null) {
                 String email = (String) persona.get("email");
@@ -661,63 +654,51 @@ public Replication getReplicator(Map<String, Object> properties) throws Couchbas
             authorizer.setLocalUUID(db.publicUUID());
         }
 
+        Replication repl = db.createReplicator(remote, push, getDefaultHttpClientFactory());
 
+        repl.setContinuous(continuous);
 
-        if (!cancel) {
-            repl = db.getReplicator(remote, getDefaultHttpClientFactory(), push, continuous);
-            if (repl == null) {
-                throw new CouchbaseLiteException("unable to create replicator with remote: " + remote,
-                        new Status(Status.INTERNAL_SERVER_ERROR));
-            }
-
-            if (authorizer != null) {
-                repl.setAuthenticator(authorizer);
-            }
+        if (authorizer != null) {
+            repl.setAuthenticator(authorizer);
+        }
 
-            Map<String, Object> headers = null;
-            if (remoteMap != null) {
-                headers = (Map) remoteMap.get("headers");
-            }
+        Map<String, Object> headers = null;
+        if (remoteMap != null) {
+            headers = (Map) remoteMap.get("headers");
+        }
 
-            if (headers != null && !headers.isEmpty()) {
-                repl.setHeaders(headers);
-            }
+        if (headers != null && !headers.isEmpty()) {
+            repl.setHeaders(headers);
+        }
 
-            String filterName = (String) properties.get("filter");
-            if (filterName != null) {
-                repl.setFilter(filterName);
-                Map<String, Object> filterParams = (Map<String, Object>) properties.get("query_params");
-                if (filterParams != null) {
-                    repl.setFilterParams(filterParams);
-                }
+        String filterName = (String) properties.get("filter");
+        if (filterName != null) {
+            repl.setFilter(filterName);
+            Map<String, Object> filterParams = (Map<String, Object>) properties.get("query_params");
+            if (filterParams != null) {
+                repl.setFilterParams(filterParams);
             }
+        }
 
-            // docIDs
-            if(properties.get("doc_ids") != null) {
-                if(properties.get("doc_ids") instanceof List){
-                    List<String> docIds = (List<String>)properties.get("doc_ids");
-                    repl.setDocIds(docIds);
-                }
+        // docIDs
+        if(properties.get("doc_ids") != null) {
+            if(properties.get("doc_ids") instanceof List){
+                List<String> docIds = (List<String>)properties.get("doc_ids");
+                repl.setDocIds(docIds);
             }
+        }
 
-            String remoteUUID = (String) properties.get("remoteUUID");
-            if (remoteUUID != null) {
-                repl.setRemoteUUID(remoteUUID);
-            }
+        String remoteUUID = (String) properties.get("remoteUUID");
+        if (remoteUUID != null) {
+            repl.setRemoteUUID(remoteUUID);
+        }
 
-            if (push) {
-                repl.setCreateTarget(createTarget);
-            }
-        } else {
-            // Cancel replication:
-            repl = db.getActiveReplicator(remote, push);
-            if (repl == null) {
-                throw new CouchbaseLiteException("unable to lookup replicator with remote: " + remote,
-                        new Status(Status.NOT_FOUND));
-            }
+        if (push) {
+            repl.setCreateTarget(createTarget);
         }
 
-        return repl;
+        Replication activeReplicator = db.findActiveReplicator(repl);
+        return activeReplicator != null ? activeReplicator : repl;
     }
 
     /**

src/main/java/com/couchbase/lite/replicator/PullerInternal.java

@@ -1047,10 +1047,10 @@ protected void pauseOrResume() {
 
     @Override
     public String toString() {
-        if (str == null) {
+        if (str == null || str.contains("unknown")) {
             String maskedRemote = "unknown";
             if (remote != null)
-                remote.toExternalForm();
+                maskedRemote = remote.toExternalForm();
             maskedRemote = maskedRemote.replaceAll("://.*:.*@", "://---:---@");
             String type = isPull() ? "pull" : "push";
             String replicationIdentifier = Utils.shortenString(remoteCheckpointDocID(), 5);

src/main/java/com/couchbase/lite/replicator/Replication.java

@@ -464,7 +464,8 @@ protected void setLastError(Throwable lastError) {
     /**
      * Following two methods for temporary methods instead of CBL_ReplicatorSettings implementation.
      */
-    protected String remoteCheckpointDocID() {
+    @InterfaceAudience.Private
+    public String remoteCheckpointDocID() {
         return replicationInternal.remoteCheckpointDocID();
     }