import com.couchbase.lite.replicator.Replication; //导入依赖的package包/类
@Override
public void changed(Replication.ChangeEvent event) {
    Throwable error = null;
    if (mPull != null) {
        if (error == null)
            error = mPull.getLastError();
    }

    if (error == null || error == mReplError)
        error = mPush.getLastError();

    if (error != mReplError) {
        mReplError = error;
        if (mReplError != null)
            showErrorMessage(mReplError.getMessage(), null);
    }
}
 

import com.couchbase.lite.replicator.Replication; //导入依赖的package包/类
private void startSync() {
    URL url;

    try {
        url = new URL(syncUrl);
    } catch (MalformedURLException e) {
        Log.e(TAG, e.getMessage());
        return;
    }

    Replication replication = mDatabase.createPullReplication(url);
    replication.setChannels(Arrays.asList(channels));
    replication.setContinuous(true);
    replication.addChangeListener(changeListener);
    replication.start();
}
 

import com.couchbase.lite.replicator.Replication; //导入依赖的package包/类
private void startSync() {

        URL syncUrl;
        try {
            syncUrl = new URL(SYNC_URL);
        } catch (MalformedURLException e) {
            throw new RuntimeException(e);
        }

        Replication pullReplication = database.createPullReplication(syncUrl);
        pullReplication.setContinuous(true);

        Replication pushReplication = database.createPushReplication(syncUrl);
        pushReplication.setContinuous(true);

        pullReplication.start();
        pushReplication.start();

        pullReplication.addChangeListener(this);
        pushReplication.addChangeListener(this);

    }
 

import com.couchbase.lite.replicator.Replication; //导入依赖的package包/类
private void startSync() {
    URL url = null;
    try {
        url = new URL(SYNC_URL);
    } catch (MalformedURLException e) {
        e.printStackTrace();
    }

    Authenticator authenticator = new BasicAuthenticator("james", "letmein");

    Replication push = database.createPushReplication(url);
    push.setContinuous(true);
    push.setAuthenticator(authenticator);
    push.start();

    Replication pull = database.createPullReplication(url);
    pull.setContinuous(true);
    pull.setAuthenticator(authenticator);
    pull.start();
}
 

import com.couchbase.lite.replicator.Replication; //导入依赖的package包/类
/**
 * @exclude
 */
@InterfaceAudience.Private
protected void forgetDatabase(Database db) {
    synchronized (lockDatabases) {
        Log.v(Log.TAG_DATABASE, "Fogetting forgetDatabase() %s %s", this, db);

        // remove from cached list of dbs
        databases.remove(db.getName());

        // remove from list of replications
        // TODO: should there be something that actually stops the replication(s) first?
        Iterator<Replication> replicationIterator = this.replications.iterator();
        while (replicationIterator.hasNext()) {
            Replication replication = replicationIterator.next();
            if (replication.getLocalDatabase().getName().equals(db.getName())) {
                replicationIterator.remove();
            }
        }

        // Remove registered encryption key if available:
        encryptionKeys.remove(db.getName());

        Log.v(Log.TAG_DATABASE, "Forgot forgetDatabase() %s %s", this, db);
    }
}
 

import com.couchbase.lite.replicator.Replication; //导入依赖的package包/类
@InterfaceAudience.Private
public void addActiveReplication(Replication replication) {
    replication.addChangeListener(new Replication.ChangeListener() {
        @Override
        public void changed(Replication.ChangeEvent event) {
            ReplicationStateTransition transition = event.getTransition();
            if (transition != null && transition.getDestination() == ReplicationState.STOPPED) {
                synchronized (activeReplicators) {
                    activeReplicators.remove(event.getSource());
                    activeReplicators.notifyAll();
                }
            }
        }
    });
    activeReplicators.add(replication);
}
 

import com.couchbase.lite.replicator.Replication; //导入依赖的package包/类
@Override
public void changed(Replication.ChangeEvent event) {
    final int changeCount = event.getChangeCount();
    if (changeCount > 0) {
        ((Activity) mContext).runOnUiThread(new Runnable() {
            @Override
            public void run() {
                Toast.makeText(mContext, String.format("%d document(s) changed", changeCount), Toast.LENGTH_SHORT).show();
            }
        });
    }
}
 

import com.couchbase.lite.replicator.Replication; //导入依赖的package包/类
/**
 * Constructor
 *
 * @throws java.lang.SecurityException - Runtime exception that can be thrown by File.mkdirs()
 */
@InterfaceAudience.Public
public Manager(Context context, ManagerOptions options) throws IOException {
    Log.i(Database.TAG, "### %s ###", getFullVersionInfo());

    this.context = context;
    this.directoryFile = context.getFilesDir();
    this.options = (options != null) ? options : DEFAULT_OPTIONS;
    this.databases = new HashMap<String, Database>();
    this.encryptionKeys = new HashMap<String, Object>();
    this.replications = new ArrayList<Replication>();

    if (!directoryFile.exists()) {
        directoryFile.mkdirs();
    }
    if (!directoryFile.isDirectory()) {
        throw new IOException(String.format(Locale.ENGLISH, "Unable to create directory for: %s", directoryFile));
    }

    upgradeOldDatabaseFiles(directoryFile);

    // this must be a single threaded executor due to contract w/ Replication object
    // which must run on either:
    // - a shared single threaded executor
    // - its own single threaded executor
    workExecutor = Executors.newSingleThreadScheduledExecutor(new ThreadFactory() {
        @Override
        public Thread newThread(Runnable r) {
            return new Thread(r, "CBLManagerWorkExecutor");
        }
    });
}
 

import com.couchbase.lite.replicator.Replication; //导入依赖的package包/类
/**
 * Returns YES if the given name is a valid database name.
 * (Only the characters in "abcdefghijklmnopqrstuvwxyz0123456789_$()+-/" are allowed.)
 */
@InterfaceAudience.Public
public static boolean isValidDatabaseName(String databaseName) {
    if (databaseName.length() > 0 && databaseName.length() < 240 &&
            containsOnlyLegalCharacters(databaseName) &&
            Character.isLowerCase(databaseName.charAt(0))) {
        return true;
    }
    return databaseName.equals(Replication.REPLICATOR_DATABASE_NAME);
}
 

import com.couchbase.lite.replicator.Replication; //导入依赖的package包/类
/**
 * Get all the replicators associated with this database.
 */
@InterfaceAudience.Public
public List<Replication> getAllReplications() {
    if (!isOpen()) throw new CouchbaseLiteRuntimeException("Database is closed.");
    storeRef.retain();
    try {
        List<Replication> allReplicatorsList = new ArrayList<Replication>();
        synchronized (activeReplicators) {
            allReplicatorsList.addAll(activeReplicators);
        }
        return allReplicatorsList;
    } finally {
        storeRef.release();
    }
}
 

import com.couchbase.lite.replicator.Replication; //导入依赖的package包/类
/**
 * Creates a new Replication that will pull from the source Database at the given url.
 *
 * @param remote the remote URL to pull from
 * @return A new Replication that will pull from the source Database at the given url.
 */
@InterfaceAudience.Public
public Replication createPullReplication(URL remote) {
    if (remote == null) throw new IllegalArgumentException("remote is null");
    if (!isOpen()) throw new CouchbaseLiteRuntimeException("Database is closed.");
    storeRef.retain();
    try {
        return new Replication(this, remote, Replication.Direction.PULL, getHttpClientFactory());
    } finally {
        storeRef.release();
    }
}
 

import com.couchbase.lite.replicator.Replication; //导入依赖的package包/类
/**
 * Creates a new Replication that will push to the target Database at the given url.
 *
 * @param remote the remote URL to push to
 * @return A new Replication that will push to the target Database at the given url.
 */
@InterfaceAudience.Public
public Replication createPushReplication(URL remote) {
    if (remote == null) throw new IllegalArgumentException("remote is null");
    if (!isOpen()) throw new CouchbaseLiteRuntimeException("Database is closed.");
    storeRef.retain();
    try {
        return new Replication(this, remote, Replication.Direction.PUSH, getHttpClientFactory());
    } finally {
        storeRef.release();
    }
}
 

import com.couchbase.lite.replicator.Replication; //导入依赖的package包/类
protected Replication findActiveReplicator(Replication replicator) {
    synchronized (activeReplicators) {
        String remoteCheckpointDocID = replicator.remoteCheckpointDocID();
        if (remoteCheckpointDocID == null)
            return null;

        for (Replication r : activeReplicators) {
            if (remoteCheckpointDocID.equals(r.remoteCheckpointDocID()) && r.isRunning())
                return r;
        }
    }
    return null;
}
 

import com.couchbase.lite.replicator.Replication; //导入依赖的package包/类
public void testPull() throws IOException {

        CountDownLatch doneSignal = new CountDownLatch(1);

        HttpClient httpClient = new CBLiteHttpClient(manager);
        CouchDbInstance dbInstance = new StdCouchDbInstance(httpClient);

        // create a local database
        CouchDbConnector dbConnector = dbInstance.createConnector(DEFAULT_TEST_DB, true);

        // push this database to the test replication server
        ReplicationCommand pushCommand = new ReplicationCommand.Builder()
            .source(getReplicationURL().toExternalForm())
            .target(DEFAULT_TEST_DB)
            .continuous(false)
            .build();

        ReplicationStatus status = dbInstance.replicate(pushCommand);
        Replication repl = database.getReplicator(status.getSessionId());

        ReplicationChangeListener replicationObserver = new ReplicationChangeListener(doneSignal);
        repl.addChangeListener(replicationObserver);

        try {
            boolean success = doneSignal.await(30, TimeUnit.SECONDS);
            assertTrue(success);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        Assert.assertNotNull(status.getSessionId());

    }
 

import com.couchbase.lite.replicator.Replication; //导入依赖的package包/类
public void testPullWithObserver() throws IOException {

        CountDownLatch doneSignal = new CountDownLatch(1);
        HttpClient httpClient = new CBLiteHttpClient(manager);
        CouchDbInstance dbInstance = new StdCouchDbInstance(httpClient);

        // push this database to the test replication server
        ReplicationCommand pushCommand = new ReplicationCommand.Builder()
            .source(getReplicationURL().toExternalForm())
            .target(DEFAULT_TEST_DB)
            .continuous(false)
            .build();

        ReplicationStatus status = dbInstance.replicate(pushCommand);
        Replication repl = database.getReplicator(status.getSessionId());
        ReplicationChangeListener replicationObserver = new ReplicationChangeListener(doneSignal);
        repl.addChangeListener(replicationObserver);


        Assert.assertNotNull(status.getSessionId());
        Assert.assertEquals(repl.getSessionID(), status.getSessionId());

        try {
            boolean success = doneSignal.await(30, TimeUnit.SECONDS);
            assertTrue(success);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        Assert.assertTrue(replicationObserver.isReplicationFinished());

        	

    }
 

import com.couchbase.lite.replicator.Replication; //导入依赖的package包/类
@Override
public void changed(Replication.ChangeEvent changeEvent) {
  if (changeEvent.getError() != null) {
    Throwable lastError = changeEvent.getError();

    Dialog.display(lastError.getMessage());

    return;
  }

  if (changeEvent.getTransition() == null) return;

  ReplicationState dest = changeEvent.getTransition().getDestination();

  replicationActive = ((dest == ReplicationState.STOPPING || dest == ReplicationState.STOPPED) ? false : true);

  stateListeners.forEach(listener -> listener.onChange(replicationActive));
}
 

import com.couchbase.lite.replicator.Replication; //导入依赖的package包/类
@InterfaceAudience.Public
public synchronized boolean close() {
    storeRef.await();

    // NOTE: synchronized Manager.lockDatabases to prevent Manager to give the deleting Database
    //       instance. See also `Manager.getDatabase(String, boolean)`.
    synchronized (manager.lockDatabases) {
        try {
            closing.set(true);

            if (!open.get()) {
                // Ensure that the database is forgotten:
                manager.forgetDatabase(this);
                return false;
            }

            synchronized (databaseListeners) {
                for (DatabaseListener listener : databaseListeners)
                    listener.databaseClosing();
            }

            synchronized (lockViews) {
                if (views != null) {
                    for (View view : views.values())
                        view.close();
                }
                views = null;
            }

            // Make all replicators stop and wait:
            boolean stopping = false;
            synchronized (activeReplicators) {
                for (Replication replicator : activeReplicators) {
                    if (replicator.getStatus() == Replication.ReplicationStatus.REPLICATION_STOPPED)
                        continue;
                    replicator.stop();
                    stopping = true;
                }

                // maximum wait time per replicator is 60 sec.
                // total maximum wait time for all replicators is between 60sec and 119 sec.
                long timeout = Replication.DEFAULT_MAX_TIMEOUT_FOR_SHUTDOWN * 1000;
                long startTime = System.currentTimeMillis();
                while (activeReplicators.size() > 0 && stopping &&
                        (System.currentTimeMillis() - startTime) < timeout) {
                    try {
                        activeReplicators.wait(timeout);
                    } catch (InterruptedException e) {
                    }
                }
                // clear active replicators:
                activeReplicators.clear();
            }

            // cancel purge timer
            cancelPurgeTimer();

            // Close Store:
            if (store != null)
                store.close();

            // Clear document cache:
            clearDocumentCache();

            // Forget database:
            manager.forgetDatabase(this);

            open.set(false);
            return true;
        } finally {
            closing.set(false);
        }
    }
}
 

import com.couchbase.lite.replicator.Replication; //导入依赖的package包/类
/**
 * TODO: To be compatible with CBL iOS, this method should move to Replicator.activeTaskInfo().
 * TODO: Reference: - (NSDictionary*) activeTaskInfo in CBL_Replicator.m
 */
private static Map<String, Object> getActivity(Replication replicator) {
    // For schema, see http://wiki.apache.org/couchdb/HttpGetActiveTasks
    Map<String, Object> activity = new HashMap<String, Object>();

    String source = replicator.getRemoteUrl().toExternalForm();
    String target = replicator.getLocalDatabase().getName();

    if (!replicator.isPull()) {
        String tmp = source;
        source = target;
        target = tmp;
    }
    int processed = replicator.getCompletedChangesCount();
    int total = replicator.getChangesCount();
    String status = String.format(Locale.ENGLISH, "Processed %d / %d changes", processed, total);
    if (!replicator.getStatus().equals(ReplicationStatus.REPLICATION_ACTIVE)) {
        //These values match the values for IOS.
        if (replicator.getStatus().equals(ReplicationStatus.REPLICATION_IDLE)) {
            status = "Idle";     // nonstandard
        } else if (replicator.getStatus().equals(ReplicationStatus.REPLICATION_STOPPED)) {
            status = "Stopped";
        } else if (replicator.getStatus().equals(ReplicationStatus.REPLICATION_OFFLINE)) {
            status = "Offline";  // nonstandard
        }
    }
    int progress = (total > 0) ? (int) 100 * processed / total : 0;

    activity.put("type", "Replication");
    activity.put("task", replicator.getSessionID());
    activity.put("source", source);
    activity.put("target", target);
    activity.put("status", status);
    activity.put("progress", progress);
    activity.put("continuous", replicator.isContinuous());

    //NOTE: Need to support "x_active_requests"

    if (replicator.getLastError() != null) {
        String msg = String.format(Locale.ENGLISH, "Replicator error: %s.  Repl: %s.  Source: %s, Target: %s",
                replicator.getLastError(), replicator, source, target);
        Log.w(TAG, msg);
        Throwable error = replicator.getLastError();
        int statusCode = 400;
        if (error instanceof RemoteRequestResponseException) {
            statusCode = ((RemoteRequestResponseException) error).getCode();
        }
        Object[] errorObjects = new Object[]{statusCode, replicator.getLastError().toString()};
        activity.put("error", errorObjects);
    } else {
        // NOTE: Following two parameters: CBL iOS does not support. We might remove them in the future.
        activity.put("change_count", total);
        activity.put("completed_change_count", processed);
    }

    return activity;
}
 

import com.couchbase.lite.replicator.Replication; //导入依赖的package包/类
public void testPush() throws IOException {

        CountDownLatch doneSignal = new CountDownLatch(1);
        HttpClient httpClient = new CBLiteHttpClient(manager);
        CouchDbInstance dbInstance = new StdCouchDbInstance(httpClient);

        // create a local database
        CouchDbConnector dbConnector = dbInstance.createConnector(DEFAULT_TEST_DB, true);

        // create 3 objects
        TestObject test1 = new TestObject(1, false, "ektorp-1");
        TestObject test2 = new TestObject(2, false, "ektorp-2");
        TestObject test3 = new TestObject(3, false, "ektorp-3");

        // save these objects in the database
        dbConnector.create(test1);
        dbConnector.create(test2);
        dbConnector.create(test3);

        // push this database to the test replication server
        ReplicationCommand pushCommand = new ReplicationCommand.Builder()
            .source(DEFAULT_TEST_DB)
            .target(getReplicationURL().toExternalForm())
            .continuous(false)
            .createTarget(true)
            .build();

        ReplicationStatus status = dbInstance.replicate(pushCommand);
        Replication repl = database.getReplicator(status.getSessionId());
        ReplicationChangeListener replicationObserver = new ReplicationChangeListener(doneSignal);
        repl.addChangeListener(replicationObserver);

        try {
            boolean success = doneSignal.await(30, TimeUnit.SECONDS);
            assertTrue(success);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        Assert.assertNotNull(status.getSessionId());

    }
 

import com.couchbase.lite.replicator.Replication; //导入依赖的package包/类
public void testFilteredPush() throws IOException {

        CountDownLatch doneSignal = new CountDownLatch(1);

        // install the filter
        database.setFilter("evenFoo", new ReplicationFilter() {

            @Override
            public boolean filter(SavedRevision revision, Map<String, Object> params) {
                Integer foo = (Integer) revision.getProperties().get("foo");
                if (foo != null && foo.intValue() % 2 == 0) {
                    return true;
                }
                return false;
            }
        });

        HttpClient httpClient = new CBLiteHttpClient(manager);
        CouchDbInstance dbInstance = new StdCouchDbInstance(httpClient);

        // create a local database
        CouchDbConnector dbConnector = dbInstance.createConnector(DEFAULT_TEST_DB, true);

        // create 3 objects
        TestObject test1 = new TestObject(1, false, "ektorp-1");
        TestObject test2 = new TestObject(2, false, "ektorp-2");
        TestObject test3 = new TestObject(3, false, "ektorp-3");

        // save these objects in the database
        dbConnector.create(test1);
        dbConnector.create(test2);
        dbConnector.create(test3);

        // push this database to the test replication server
        ReplicationCommand pushCommand = new ReplicationCommand.Builder()
            .source(DEFAULT_TEST_DB)
            .target(getReplicationURL().toExternalForm())
            .continuous(false)
            .filter("evenFoo")
            .build();

        ReplicationStatus status = dbInstance.replicate(pushCommand);
        Replication repl = database.getReplicator(status.getSessionId());

        ReplicationChangeListener replicationObserver = new ReplicationChangeListener(doneSignal);
        repl.addChangeListener(replicationObserver);

        try {
            boolean success = doneSignal.await(30, TimeUnit.SECONDS);
            assertTrue(success);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        Assert.assertNotNull(status.getSessionId());

    }