/**
 * Create and return a c4Replicator targeting the passed Database
 *
 * @param otherDb a local database for the replication target
 * @return the c4Replicator
 * @throws LiteCoreException on failure to create the replicator
 */
@GuardedBy("lock")
@NonNull
protected final C4Replicator getLocalC4ReplicatorLocked(@NonNull Database otherDb) throws LiteCoreException {
    setupFilters();

    final boolean continuous = config.isContinuous();

    c4ReplListener = new ReplicatorListener();

    return config.getDatabase().createLocalReplicator(
        (Replicator) this,
        otherDb.getC4Database(),
        mkmode(isPush(config.getReplicatorType()), continuous),
        mkmode(isPull(config.getReplicatorType()), continuous),
        getFleeceOptions(),
        c4ReplListener,
        c4ReplPushFilter,
        c4ReplPullFilter);
}
 

@GuardedBy("lock")
private void verifyQuiescent() throws CouchbaseLiteException {
    if (hasActiveReplicators()) {
        throw new CouchbaseLiteException(
            "DeleteDBFailedReplications",
            CBLError.Domain.CBLITE,
            CBLError.Code.BUSY);
    }

    if (!activeLiveQueries.isEmpty()) {
        throw new CouchbaseLiteException(
            "DeleteDBFailedQueryListeners",
            CBLError.Domain.CBLITE,
            CBLError.Code.BUSY);
    }
}
 

@GuardedBy("lock")
private void shutdown() {
    // cancel purge
    if (purgeStrategy != null) { purgeStrategy.cancelPurges(); }

    // release instances
    freeC4Observers(true);

    if ((!shellMode) && (c4db != null)) {
        path = c4db.getPath();
        c4db = null;
    }

    // shutdown executor service
    shutdownExecutors(postExecutor, queryExecutor, SHUTDOWN_DELAY_SECS);
}
 

@GuardedBy("lock")
private void updateC4DocumentLocked(@Nullable C4Document c4Doc) {
    if (c4Document == c4Doc) { return; }

    if (c4Doc != null) { revId = null; }

    c4Document = c4Doc;
}
 

@GuardedBy("lock")
private void updateDictionaryLocked(boolean mutable) {
    if (data == null) {
        root = null;
        internalDict = mutable ? new MutableDictionary() : new Dictionary();
        return;
    }

    final Database db = getDatabase();
    if (db == null) { throw new IllegalStateException(""); }

    final MRoot newRoot = new MRoot(new DocContext(db, c4Document), data.toFLValue(), mutable);
    root = newRoot;
    synchronized (db.getLock()) { internalDict = (Dictionary) newRoot.asNative(); }
}
 

/**
 * Create and return a c4Replicator targeting the passed URI
 *
 * @param remoteUri a URI for the replication target
 * @return the c4Replicator
 * @throws LiteCoreException on failure to create the replicator
 */
@GuardedBy("lock")
@NonNull
protected final C4Replicator getRemoteC4ReplicatorLocked(@NonNull URI remoteUri) throws LiteCoreException {
    // Set up the port: core uses 0 for not set
    final int p = remoteUri.getPort();
    final int port = (p <= 0) ? 0 : p;

    // get db name and path
    final Deque<String> splitPath = splitPath(remoteUri.getPath());
    final String dbName = (splitPath.size() <= 0) ? "" : splitPath.removeLast();
    final String path = "/" + StringUtils.join("/", splitPath);

    setupFilters();

    final boolean continuous = config.isContinuous();

    c4ReplListener = new ReplicatorListener();

    return config.getDatabase().createRemoteReplicator(
        (Replicator) this,
        remoteUri.getScheme(),
        remoteUri.getHost(),
        port,
        path,
        dbName,
        mkmode(isPush(config.getReplicatorType()), continuous),
        mkmode(isPull(config.getReplicatorType()), continuous),
        getFleeceOptions(),
        c4ReplListener,
        c4ReplPushFilter,
        c4ReplPullFilter,
        socketFactory,
        C4Socket.NO_FRAMING);
}
 

/**
 * Create and return a c4Replicator.
 * The socket factory is responsible for setting up the target
 *
 * @param framing the framing mode (C4Socket.XXX_FRAMING)
 * @return the c4Replicator
 * @throws LiteCoreException on failure to create the replicator
 */
@GuardedBy("lock")
@NonNull
protected final C4Replicator getMessageC4ReplicatorLocked(int framing)
    throws LiteCoreException {
    setupFilters();

    final boolean continuous = config.isContinuous();

    c4ReplListener = new ReplicatorListener();

    return config.getDatabase().createRemoteReplicator(
        (Replicator) this,
        C4Socket.MESSAGE_SCHEME,
        null,
        0,
        null,
        null,
        mkmode(isPush(config.getReplicatorType()), continuous),
        mkmode(isPull(config.getReplicatorType()), continuous),
        getFleeceOptions(),
        c4ReplListener,
        c4ReplPushFilter,
        c4ReplPullFilter,
        socketFactory,
        framing);
}
 

@GuardedBy("lock")
@NonNull
private ListenerToken addDatabaseChangeListenerLocked(
    @Nullable Executor executor,
    @NonNull DatabaseChangeListener listener) {
    if (dbChangeNotifier == null) {
        dbChangeNotifier = new ChangeNotifier<>();
        registerC4DBObserver();
    }
    return dbChangeNotifier.addChangeListener(executor, listener);
}
 

@GuardedBy("lock")
private void removeDatabaseChangeListenerLocked(@NonNull ListenerToken token) {
    if (dbChangeNotifier.removeChangeListener(token) == 0) {
        freeC4DBObserver();
        dbChangeNotifier = null;
    }
}
 

@GuardedBy("lock")
@NonNull
private ListenerToken addDocumentChangeListenerLocked(
    @NonNull String docID,
    @Nullable Executor executor,
    @NonNull DocumentChangeListener listener) {
    DocumentChangeNotifier docNotifier = docChangeNotifiers.get(docID);
    if (docNotifier == null) {
        docNotifier = new DocumentChangeNotifier((Database) this, docID);
        docChangeNotifiers.put(docID, docNotifier);
    }
    final ChangeListenerToken token = docNotifier.addChangeListener(executor, listener);
    token.setKey(docID);
    return token;
}
 

@GuardedBy("lock")
private void removeDocumentChangeListenerLocked(@NonNull ChangeListenerToken token) {
    final String docID = (String) token.getKey();
    if (docChangeNotifiers.containsKey(docID)) {
        final DocumentChangeNotifier notifier = docChangeNotifiers.get(docID);
        if (notifier != null && notifier.removeChangeListener(token) == 0) {
            notifier.stop();
            docChangeNotifiers.remove(docID);
        }
    }
}
 

@GuardedBy("lock")
private void freeC4DBObserver() {
    final C4DatabaseObserver observer = c4DbObserver;
    c4DbObserver = null;

    if (observer == null) { return; }
    observer.free();
}
 

@GuardedBy("lock")
private C4Query prepQueryLocked() throws CouchbaseLiteException {
    database = (Database) from.getSource();

    final String json = encodeAsJson();
    Log.v(DOMAIN, "Encoded query: %s", json);
    if (json == null) { throw new CouchbaseLiteException("Failed to generate JSON query."); }

    if (columnNames == null) { columnNames = getColumnNames(); }

    try { return database.getC4Database().createQuery(json); }
    catch (LiteCoreException e) { throw CBLStatus.convertException(e); }
}
 

@GuardedBy("this")
private void executeTask(@NonNull InstrumentedTask newTask) {
    try {
        executor.execute(newTask);
        running++;
    }
    catch (RejectedExecutionException e) {
        dumpExecutorState(newTask, e);
        throw e;
    }
}
 

@GuardedBy("this")
private void executeTask(@Nullable InstrumentedTask prevTask) {
    final InstrumentedTask nextTask = pendingTasks.peek();
    if (nextTask == null) { return; }
    try {
        executor.execute(nextTask);
        needsRestart = false;
    }
    catch (RejectedExecutionException e) {
        needsRestart = true;
        dumpExecutorState(e, prevTask);
    }
}
 

@GuardedBy("lock")
@SuppressWarnings("PMD.NPathComplexity")
private void saveResolvedDocument(
    @Nullable Document resolvedDoc,
    @NonNull Document localDoc,
    @NonNull Document remoteDoc)
    throws CouchbaseLiteException {
    FLSliceResult mergedBody = null;
    int mergedFlags = 0x00;

    if (resolvedDoc == null) {
        if (remoteDoc.isDeleted()) { resolvedDoc = remoteDoc; }
        else if (localDoc.isDeleted()) { resolvedDoc = localDoc; }
    }

    if (resolvedDoc != null) {
        if (resolvedDoc != localDoc) { resolvedDoc.setDatabase((Database) this); }

        final C4Document c4Doc = resolvedDoc.getC4doc();
        if (c4Doc != null) { mergedFlags = c4Doc.getSelectedFlags(); }
    }

    try {
        // Unless the remote revision is being used as-is, we need a new revision:
        if (resolvedDoc != remoteDoc) {
            if ((resolvedDoc != null) && !resolvedDoc.isDeleted()) { mergedBody = resolvedDoc.encode(); }
            else {
                mergedFlags |= C4Constants.RevisionFlags.DELETED;
                final FLEncoder enc = getC4Database().getSharedFleeceEncoder();
                try {
                    enc.writeValue(new HashMap<>()); // Need an empty dictionary body
                    mergedBody = enc.finish2();
                }
                finally { enc.reset(); }
            }
        }

        // Merged body:
        final byte[] mergedBodyBytes = mergedBody == null ? null : mergedBody.getBuf();

        // Ask LiteCore to do the resolution:
        final C4Document rawDoc = Preconditions.assertNotNull(localDoc.getC4doc(), "raw doc is null");
        // The remote branch has to win so that the doc revision history matches the server's.
        rawDoc.resolveConflict(remoteDoc.getRevisionID(), localDoc.getRevisionID(), mergedBodyBytes, mergedFlags);
        rawDoc.save(0);

        Log.i(DOMAIN, "Conflict resolved as doc '%s' rev %s", rawDoc.getDocID(), rawDoc.getRevID());
    }
    catch (LiteCoreException e) {
        throw CBLStatus.convertException(e);
    }
    finally {
        if (mergedBody != null) { mergedBody.free(); }
    }
}
 

@GuardedBy("lock")
private void freeC4Observers(boolean cleanup) {
    final Map<String, DocumentChangeNotifier> notifiers = docChangeNotifiers;

    freeC4DBObserver();

    if (notifiers == null) { return; }

    for (DocumentChangeNotifier notifier : notifiers.values()) { notifier.stop(); }

    if (cleanup) { notifiers.clear(); }
}