<commit_before>package com.couchbase.lite.replicator;

import com.couchbase.lite.CouchbaseLiteException;
import com.couchbase.lite.Database;
import com.couchbase.lite.Misc;
import com.couchbase.lite.ReplicationFilter;
import com.couchbase.lite.RevisionList;
import com.couchbase.lite.SavedRevision;
import com.couchbase.lite.Status;
import com.couchbase.lite.auth.Authenticator;
import com.couchbase.lite.auth.AuthenticatorImpl;
import com.couchbase.lite.internal.InterfaceAudience;
import com.couchbase.lite.internal.RevisionInternal;
import com.couchbase.lite.support.BatchProcessor;
import com.couchbase.lite.support.Batcher;
import com.couchbase.lite.support.BlockingQueueListener;
import com.couchbase.lite.support.CustomFuture;
import com.couchbase.lite.support.CustomLinkedBlockingQueue;
import com.couchbase.lite.support.HttpClientFactory;
import com.couchbase.lite.support.RemoteRequestCompletionBlock;
import com.couchbase.lite.support.RemoteRequestRetry;
import com.couchbase.lite.util.CollectionUtils;
import com.couchbase.lite.util.Log;
import com.couchbase.lite.util.TextUtils;
import com.couchbase.lite.util.URIUtils;
import com.couchbase.lite.util.Utils;
import com.couchbase.org.apache.http.entity.mime.MultipartEntity;
import com.github.oxo42.stateless4j.StateMachine;
import com.github.oxo42.stateless4j.delegates.Action1;
import com.github.oxo42.stateless4j.transitions.Transition;

import org.apache.http.Header;
import org.apache.http.HttpResponse;
import org.apache.http.client.HttpResponseException;
import org.apache.http.cookie.Cookie;
import org.apache.http.impl.cookie.BasicClientCookie2;

import java.io.IOException;
import java.net.MalformedURLException;
import java.net.URL;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.TreeMap;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Internal Replication object that does the heavy lifting
 *
 * @exclude
 */
@InterfaceAudience.Private
abstract class ReplicationInternal implements BlockingQueueListener {

    // Change listeners can be called back synchronously or asynchronously.
    protected enum ChangeListenerNotifyStyle {
        SYNC, ASYNC
    }

    public static final String BY_CHANNEL_FILTER_NAME = "sync_gateway/bychannel";

    public static final String CHANNELS_QUERY_PARAM = "channels";

    public static final int EXECUTOR_THREAD_POOL_SIZE = 5;

    private static int lastSessionID = 0;

    public static int MAX_RETRIES = 10;  // total number of attempts = 11 (1 initial + MAX_RETRIES)
    public static int RETRY_DELAY_SECONDS = 60; // #define kRetryDelay 60.0 in CBL_Replicator.m

    protected Replication parentReplication;
    protected Database db;
    protected URL remote;
    protected HttpClientFactory clientFactory;
    protected String lastSequence;
    protected Authenticator authenticator;
    protected String filterName;
    protected Map<String, Object> filterParams;
    protected List<String> documentIDs;
    private String remoteUUID;
    protected Map<String, Object> requestHeaders;
    private String serverType;
    protected Batcher<RevisionInternal> batcher;
    protected static int PROCESSOR_DELAY = 500;
    protected static int INBOX_CAPACITY = 100;
    protected ScheduledExecutorService remoteRequestExecutor;
    private Throwable error; // use private to make sure if error is set through setError()
    private String remoteCheckpointDocID;
    protected Map<String, Object> remoteCheckpoint;
    protected AtomicInteger completedChangesCount;
    protected AtomicInteger changesCount;
    protected CollectionUtils.Functor<RevisionInternal, RevisionInternal> revisionBodyTransformationBlock;
    protected String sessionID;
    protected BlockingQueue<Future> pendingFutures;
    private boolean lastSequenceChanged = false;
    private boolean savingCheckpoint;
    private boolean overdueForCheckpointSave;

    // the code assumes this is a _single threaded_ work executor.
    // if it's not, the behavior will be buggy.  I don't see a way to assert this in the code.
    protected ScheduledExecutorService workExecutor;

    protected StateMachine<ReplicationState, ReplicationTrigger> stateMachine;
    protected List<ChangeListener> changeListeners;
    protected Replication.Lifecycle lifecycle;
    protected ChangeListenerNotifyStyle changeListenerNotifyStyle;

    private Future retryFuture = null;  // future obj of retry task
    private int retryCount = 0;         // counter for retry.

    /**
     * Constructor
     */
    ReplicationInternal(Database db, URL remote,
                        HttpClientFactory clientFactory,
                        ScheduledExecutorService workExecutor,
                        Replication.Lifecycle lifecycle,
                        Replication parentReplication) {

        Utils.assertNotNull(lifecycle, "Must pass in a non-null lifecycle");

        this.parentReplication = parentReplication;
        this.db = db;
        this.remote = remote;
        this.clientFactory = clientFactory;
        this.workExecutor = workExecutor;
        this.lifecycle = lifecycle;

        this.requestHeaders = new HashMap<String, Object>();

        changeListeners = new CopyOnWriteArrayList<ChangeListener>();

        // The reason that notifications are ASYNC is to make the public API call
        // Replication.getStatus() work as expected.  Because if this is set to SYNC,
        // it causes the following issue:
        // - Notification sent from state transition from INITIAL -> RUNNING.
        // - Replication change listener called back during transition.
        // - Replication change listener calls replication.status(), and gets INITIAL instead of RUNNING.
        // - Replication change listener never notified when it goes into the RUNNING state.
        // Workarounds to above problem:
        // - By sending notifications ASYNC, by the time that the listener calls replication.status(),
        //   calling replication.getStatus() will return RUNNING.
        // - Alternatively, change listeners could look at the passed transition rather than
        //   depending on calling replication.status(), and changeListenerNotifyStyle could be set to SYNC.
        changeListenerNotifyStyle = ChangeListenerNotifyStyle.ASYNC;

        pendingFutures = new CustomLinkedBlockingQueue<Future>(this);

        initializeStateMachine();

        this.retryCount = 0;
    }

    /**
     * Trigger this replication to start (async)
     */
    public void triggerStart() {
        fireTrigger(ReplicationTrigger.START);
    }

    /**
     * Trigger this replication to stop (async)
     */
    public void triggerStopGraceful() {
        fireTrigger(ReplicationTrigger.STOP_GRACEFUL);
    }

    /**
     * Trigger this replication to go offline (async)
     */
    public void triggerGoOffline() {
        fireTrigger(ReplicationTrigger.GO_OFFLINE);
    }

    /**
     * Trigger this replication to go online (async)
     */
    public void triggerGoOnline() {
        fireTrigger(ReplicationTrigger.GO_ONLINE);
    }

    /**
     * Fire a trigger to the state machine
     */
    protected void fireTrigger(final ReplicationTrigger trigger) {
        Log.d(Log.TAG_SYNC, "[fireTrigger()] => " + trigger);
        // All state machine triggers need to happen on the replicator thread
        synchronized (workExecutor) {
            if (!workExecutor.isShutdown()) {
                workExecutor.submit(new Runnable() {
                    @Override
                    public void run() {
                        try {
                            Log.d(Log.TAG_SYNC, "firing trigger: %s", trigger);
                            stateMachine.fire(trigger);
                        } catch (Exception e) {
                            e.printStackTrace();
                            throw new RuntimeException(e);
                        }
                    }
                });
            }
        }
    }

    /**
     * Trigger this replication to stop immediately -- assumes pending work has
     * been drained, or that caller chooses to ignore any pending work.
     */
    protected void triggerStopImmediate() {
        fireTrigger(ReplicationTrigger.STOP_IMMEDIATE);
    }

    /**
     * Start the replication process.
     */
    protected void start() {

        try {
            if (!db.isOpen()) {

                String msg = String.format("Db: %s is not open, abort replication", db);
                parentReplication.setLastError(new Exception(msg));

                fireTrigger(ReplicationTrigger.STOP_IMMEDIATE);

                return;

            }

            db.addReplication(parentReplication);
            db.addActiveReplication(parentReplication);

            initSessionId();

            // init batcher
            initBatcher();

            // init authorizer / authenticator
            initAuthorizer();

            // call goOnline (or trigger state change into online state)
            goOnlineInitialStartup();

            initNetworkReachabilityManager();

            this.retryCount = 0;
        } catch (Exception e) {
            Log.e(Log.TAG_SYNC, "%s: Exception in start()", e, this);
        }

    }

    private void initSessionId() {
        this.sessionID = String.format("repl%03d", ++lastSessionID);
    }

    /**
     * Take the replication offline
     */
    protected void goOffline() {
        // implemented by subclasses
    }

    /**
     * Put the replication back online after being offline
     */
    protected void goOnline() {
        // implemented by subclasses

    }

    public void databaseClosing() {
        triggerStopGraceful();
    }

    /**
     * Close all resources associated with this replicator.
     */
    protected void close() {
        // shutdown ScheduledExecutorService. Without shutdown, cause thread leak
        if (remoteRequestExecutor != null && !remoteRequestExecutor.isShutdown()) {
            // Note: Time to wait is set 60 sec because RemoteRequest's socket timeout is set 60 seconds.
            Utils.shutdownAndAwaitTermination(remoteRequestExecutor,
                    Replication.DEFAULT_MAX_TIMEOUT_FOR_SHUTDOWN,
                    Replication.DEFAULT_MAX_TIMEOUT_FOR_SHUTDOWN);
        }
    }

    protected void initAuthorizer() {
        // TODO: add this back in  .. See Replication constructor

    }

    protected void initBatcher() {

        batcher = new Batcher<RevisionInternal>(workExecutor, INBOX_CAPACITY, PROCESSOR_DELAY, new BatchProcessor<RevisionInternal>() {
            @Override
            public void process(List<RevisionInternal> inbox) {

                try {
                    Log.v(Log.TAG_SYNC, "*** %s: BEGIN processInbox (%d sequences)", this, inbox.size());
                    processInbox(new RevisionList(inbox));
                    Log.v(Log.TAG_SYNC, "*** %s: END processInbox (lastSequence=%s)", this, lastSequence);
                } catch (Exception e) {
                    Log.e(Log.TAG_SYNC, "ERROR: processInbox failed: ", e);
                    throw new RuntimeException(e);
                }
            }
        });


    }

    protected void initNetworkReachabilityManager() {
        db.getManager().getContext().getNetworkReachabilityManager().addNetworkReachabilityListener(parentReplication);
    }

    public abstract boolean shouldCreateTarget();

    public abstract void setCreateTarget(boolean createTarget);

    protected void goOnlineInitialStartup() {

        int executorThreadPoolSize = db.getManager().getExecutorThreadPoolSize() <= 0 ?
                EXECUTOR_THREAD_POOL_SIZE : db.getManager().getExecutorThreadPoolSize();
        Log.v(Log.TAG_SYNC, "executorThreadPoolSize=" + executorThreadPoolSize);
        remoteRequestExecutor = Executors.newScheduledThreadPool(executorThreadPoolSize, new ThreadFactory() {
            private int counter = 0;

            @Override
            public Thread newThread(Runnable r) {
                String threadName = "CBLRequestWorker";
                try {
                    String replicationIdentifier = Utils.shortenString(remoteCheckpointDocID(), 5);
                    threadName = String.format("CBLRequestWorker-%s-%s", replicationIdentifier, counter++);
                } catch (Exception e) {
                    Log.e(Log.TAG_SYNC, "Error creating thread name", e);
                }
                return new Thread(r, threadName);
            }
        });

        checkSession();

    }

    @InterfaceAudience.Private
    protected void checkSession() {
        // REVIEW : This is not in line with the iOS implementation
        if (getAuthenticator() != null && ((AuthenticatorImpl) getAuthenticator()).usesCookieBasedLogin()) {
            checkSessionAtPath("/_session");
        } else {
            fetchRemoteCheckpointDoc();
        }
    }

    @InterfaceAudience.Private
    protected void checkSessionAtPath(final String sessionPath) {

        Future future = sendAsyncRequest("GET", sessionPath, null, new RemoteRequestCompletionBlock() {

            @Override
            public void onCompletion(HttpResponse httpResponse, Object result, Throwable err) {

                try {
                    if (err != null) {
                        // If not at /db/_session, try CouchDB location /_session
                        if (err instanceof HttpResponseException &&
                                ((HttpResponseException) err).getStatusCode() == 404 &&
                                "/_session".equalsIgnoreCase(sessionPath)) {

                            checkSessionAtPath("_session");
                            return;
                        }
                        Log.e(Log.TAG_SYNC, this + ": Session check failed", err);
                        setError(err);

                    } else {
                        Map<String, Object> response = (Map<String, Object>) result;
                        Log.e(Log.TAG_SYNC, "%s checkSessionAtPath() response: %s", this, response);
                        Map<String, Object> userCtx = (Map<String, Object>) response.get("userCtx");
                        String username = (String) userCtx.get("name");
                        if (username != null && username.length() > 0) {
                            Log.d(Log.TAG_SYNC, "%s Active session, logged in as %s", this, username);
                            fetchRemoteCheckpointDoc();
                        } else {
                            Log.d(Log.TAG_SYNC, "%s No active session, going to login", this);
                            login();
                        }
                    }

                } catch (Exception e) {
                    Log.e(Log.TAG_SYNC, "%s Exception in checkSessionAtPath()", this, e);
                }

            }

        });
        pendingFutures.add(future);
    }

    @InterfaceAudience.Private
    protected void login() {
        Map<String, String> loginParameters = ((AuthenticatorImpl) getAuthenticator()).loginParametersForSite(remote);
        if (loginParameters == null) {
            Log.d(Log.TAG_SYNC, "%s: %s has no login parameters, so skipping login", this, getAuthenticator());
            fetchRemoteCheckpointDoc();
            return;
        }

        final String loginPath = ((AuthenticatorImpl) getAuthenticator()).loginPathForSite(remote);

        Log.d(Log.TAG_SYNC, "%s: Doing login with %s at %s", this, getAuthenticator().getClass(), loginPath);

        Future future = sendAsyncRequest("POST", loginPath, loginParameters, new RemoteRequestCompletionBlock() {

            @Override
            public void onCompletion(HttpResponse httpResponse, Object result, Throwable e) {
                if (e != null) {
                    Log.d(Log.TAG_SYNC, "%s: Login failed for path: %s", this, loginPath);
                    setError(e);

                    // TODO: double check this behavior against iOS implementation, especially
                    // TODO: with regards to behavior of a continuous replication.
                    // Note: was added in order that unit test testReplicatorErrorStatus() finished and passed.
                    // (before adding this, the replication would just end up in limbo and never finish)
                    triggerStopGraceful();

                } else {
                    Log.v(Log.TAG_SYNC, "%s: Successfully logged in!", this);
                    fetchRemoteCheckpointDoc();
                }
            }

        });
        pendingFutures.add(future);

    }

    @InterfaceAudience.Private
    protected void setError(Throwable throwable) {

        // TODO - needs to port
        //if (error.code == NSURLErrorCancelled && $equal(error.domain, NSURLErrorDomain))
        //    return;

        if (throwable != this.error) {
            Log.e(Log.TAG_SYNC, "%s: Progress: set error = %s", this, throwable);
            parentReplication.setLastError(throwable);
            this.error = throwable;

            // if permanent error, stop immediately
            if(Utils.isPermanentError(this.error)) {
                triggerStopGraceful();
            }

            // iOS version sends notification from stop() method, but java version does not.
            // following codes are always executed.
            Replication.ChangeEvent changeEvent = new Replication.ChangeEvent(this);
            changeEvent.setError(this.error);
            notifyChangeListeners(changeEvent);
        }
    }
<commit_msg>Stop single-shot replicator for error.<commit_after>package com.couchbase.lite.replicator;

import com.couchbase.lite.CouchbaseLiteException;
import com.couchbase.lite.Database;
import com.couchbase.lite.Misc;
import com.couchbase.lite.ReplicationFilter;
import com.couchbase.lite.RevisionList;
import com.couchbase.lite.SavedRevision;
import com.couchbase.lite.Status;
import com.couchbase.lite.auth.Authenticator;
import com.couchbase.lite.auth.AuthenticatorImpl;
import com.couchbase.lite.internal.InterfaceAudience;
import com.couchbase.lite.internal.RevisionInternal;
import com.couchbase.lite.support.BatchProcessor;
import com.couchbase.lite.support.Batcher;
import com.couchbase.lite.support.BlockingQueueListener;
import com.couchbase.lite.support.CustomFuture;
import com.couchbase.lite.support.CustomLinkedBlockingQueue;
import com.couchbase.lite.support.HttpClientFactory;
import com.couchbase.lite.support.RemoteRequestCompletionBlock;
import com.couchbase.lite.support.RemoteRequestRetry;
import com.couchbase.lite.util.CollectionUtils;
import com.couchbase.lite.util.Log;
import com.couchbase.lite.util.TextUtils;
import com.couchbase.lite.util.URIUtils;
import com.couchbase.lite.util.Utils;
import com.couchbase.org.apache.http.entity.mime.MultipartEntity;
import com.github.oxo42.stateless4j.StateMachine;
import com.github.oxo42.stateless4j.delegates.Action1;
import com.github.oxo42.stateless4j.transitions.Transition;

import org.apache.http.Header;
import org.apache.http.HttpResponse;
import org.apache.http.client.HttpResponseException;
import org.apache.http.cookie.Cookie;
import org.apache.http.impl.cookie.BasicClientCookie2;

import java.io.IOException;
import java.net.MalformedURLException;
import java.net.URL;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.TreeMap;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Internal Replication object that does the heavy lifting
 *
 * @exclude
 */
@InterfaceAudience.Private
abstract class ReplicationInternal implements BlockingQueueListener {

    // Change listeners can be called back synchronously or asynchronously.
    protected enum ChangeListenerNotifyStyle {
        SYNC, ASYNC
    }

    public static final String BY_CHANNEL_FILTER_NAME = "sync_gateway/bychannel";

    public static final String CHANNELS_QUERY_PARAM = "channels";

    public static final int EXECUTOR_THREAD_POOL_SIZE = 5;

    private static int lastSessionID = 0;

    public static int MAX_RETRIES = 10;  // total number of attempts = 11 (1 initial + MAX_RETRIES)
    public static int RETRY_DELAY_SECONDS = 60; // #define kRetryDelay 60.0 in CBL_Replicator.m

    protected Replication parentReplication;
    protected Database db;
    protected URL remote;
    protected HttpClientFactory clientFactory;
    protected String lastSequence;
    protected Authenticator authenticator;
    protected String filterName;
    protected Map<String, Object> filterParams;
    protected List<String> documentIDs;
    private String remoteUUID;
    protected Map<String, Object> requestHeaders;
    private String serverType;
    protected Batcher<RevisionInternal> batcher;
    protected static int PROCESSOR_DELAY = 500;
    protected static int INBOX_CAPACITY = 100;
    protected ScheduledExecutorService remoteRequestExecutor;
    private Throwable error; // use private to make sure if error is set through setError()
    private String remoteCheckpointDocID;
    protected Map<String, Object> remoteCheckpoint;
    protected AtomicInteger completedChangesCount;
    protected AtomicInteger changesCount;
    protected CollectionUtils.Functor<RevisionInternal, RevisionInternal> revisionBodyTransformationBlock;
    protected String sessionID;
    protected BlockingQueue<Future> pendingFutures;
    private boolean lastSequenceChanged = false;
    private boolean savingCheckpoint;
    private boolean overdueForCheckpointSave;

    // the code assumes this is a _single threaded_ work executor.
    // if it's not, the behavior will be buggy.  I don't see a way to assert this in the code.
    protected ScheduledExecutorService workExecutor;

    protected StateMachine<ReplicationState, ReplicationTrigger> stateMachine;
    protected List<ChangeListener> changeListeners;
    protected Replication.Lifecycle lifecycle;
    protected ChangeListenerNotifyStyle changeListenerNotifyStyle;

    private Future retryFuture = null;  // future obj of retry task
    private int retryCount = 0;         // counter for retry.

    /**
     * Constructor
     */
    ReplicationInternal(Database db, URL remote,
                        HttpClientFactory clientFactory,
                        ScheduledExecutorService workExecutor,
                        Replication.Lifecycle lifecycle,
                        Replication parentReplication) {

        Utils.assertNotNull(lifecycle, "Must pass in a non-null lifecycle");

        this.parentReplication = parentReplication;
        this.db = db;
        this.remote = remote;
        this.clientFactory = clientFactory;
        this.workExecutor = workExecutor;
        this.lifecycle = lifecycle;

        this.requestHeaders = new HashMap<String, Object>();

        changeListeners = new CopyOnWriteArrayList<ChangeListener>();

        // The reason that notifications are ASYNC is to make the public API call
        // Replication.getStatus() work as expected.  Because if this is set to SYNC,
        // it causes the following issue:
        // - Notification sent from state transition from INITIAL -> RUNNING.
        // - Replication change listener called back during transition.
        // - Replication change listener calls replication.status(), and gets INITIAL instead of RUNNING.
        // - Replication change listener never notified when it goes into the RUNNING state.
        // Workarounds to above problem:
        // - By sending notifications ASYNC, by the time that the listener calls replication.status(),
        //   calling replication.getStatus() will return RUNNING.
        // - Alternatively, change listeners could look at the passed transition rather than
        //   depending on calling replication.status(), and changeListenerNotifyStyle could be set to SYNC.
        changeListenerNotifyStyle = ChangeListenerNotifyStyle.ASYNC;

        pendingFutures = new CustomLinkedBlockingQueue<Future>(this);

        initializeStateMachine();

        this.retryCount = 0;
    }

    /**
     * Trigger this replication to start (async)
     */
    public void triggerStart() {
        fireTrigger(ReplicationTrigger.START);
    }

    /**
     * Trigger this replication to stop (async)
     */
    public void triggerStopGraceful() {
        fireTrigger(ReplicationTrigger.STOP_GRACEFUL);
    }

    /**
     * Trigger this replication to go offline (async)
     */
    public void triggerGoOffline() {
        fireTrigger(ReplicationTrigger.GO_OFFLINE);
    }

    /**
     * Trigger this replication to go online (async)
     */
    public void triggerGoOnline() {
        fireTrigger(ReplicationTrigger.GO_ONLINE);
    }

    /**
     * Fire a trigger to the state machine
     */
    protected void fireTrigger(final ReplicationTrigger trigger) {
        Log.d(Log.TAG_SYNC, "[fireTrigger()] => " + trigger);
        // All state machine triggers need to happen on the replicator thread
        synchronized (workExecutor) {
            if (!workExecutor.isShutdown()) {
                workExecutor.submit(new Runnable() {
                    @Override
                    public void run() {
                        try {
                            Log.d(Log.TAG_SYNC, "firing trigger: %s", trigger);
                            stateMachine.fire(trigger);
                        } catch (Exception e) {
                            e.printStackTrace();
                            throw new RuntimeException(e);
                        }
                    }
                });
            }
        }
    }

    /**
     * Trigger this replication to stop immediately -- assumes pending work has
     * been drained, or that caller chooses to ignore any pending work.
     */
    protected void triggerStopImmediate() {
        fireTrigger(ReplicationTrigger.STOP_IMMEDIATE);
    }

    /**
     * Start the replication process.
     */
    protected void start() {

        try {
            if (!db.isOpen()) {

                String msg = String.format("Db: %s is not open, abort replication", db);
                parentReplication.setLastError(new Exception(msg));

                fireTrigger(ReplicationTrigger.STOP_IMMEDIATE);

                return;

            }

            db.addReplication(parentReplication);
            db.addActiveReplication(parentReplication);

            initSessionId();

            // init batcher
            initBatcher();

            // init authorizer / authenticator
            initAuthorizer();

            // call goOnline (or trigger state change into online state)
            goOnlineInitialStartup();

            initNetworkReachabilityManager();

            this.retryCount = 0;
        } catch (Exception e) {
            Log.e(Log.TAG_SYNC, "%s: Exception in start()", e, this);
        }

    }

    private void initSessionId() {
        this.sessionID = String.format("repl%03d", ++lastSessionID);
    }

    /**
     * Take the replication offline
     */
    protected void goOffline() {
        // implemented by subclasses
    }

    /**
     * Put the replication back online after being offline
     */
    protected void goOnline() {
        // implemented by subclasses

    }

    public void databaseClosing() {
        triggerStopGraceful();
    }

    /**
     * Close all resources associated with this replicator.
     */
    protected void close() {
        // shutdown ScheduledExecutorService. Without shutdown, cause thread leak
        if (remoteRequestExecutor != null && !remoteRequestExecutor.isShutdown()) {
            // Note: Time to wait is set 60 sec because RemoteRequest's socket timeout is set 60 seconds.
            Utils.shutdownAndAwaitTermination(remoteRequestExecutor,
                    Replication.DEFAULT_MAX_TIMEOUT_FOR_SHUTDOWN,
                    Replication.DEFAULT_MAX_TIMEOUT_FOR_SHUTDOWN);
        }
    }

    protected void initAuthorizer() {
        // TODO: add this back in  .. See Replication constructor

    }

    protected void initBatcher() {

        batcher = new Batcher<RevisionInternal>(workExecutor, INBOX_CAPACITY, PROCESSOR_DELAY, new BatchProcessor<RevisionInternal>() {
            @Override
            public void process(List<RevisionInternal> inbox) {

                try {
                    Log.v(Log.TAG_SYNC, "*** %s: BEGIN processInbox (%d sequences)", this, inbox.size());
                    processInbox(new RevisionList(inbox));
                    Log.v(Log.TAG_SYNC, "*** %s: END processInbox (lastSequence=%s)", this, lastSequence);
                } catch (Exception e) {
                    Log.e(Log.TAG_SYNC, "ERROR: processInbox failed: ", e);
                    throw new RuntimeException(e);
                }
            }
        });


    }

    protected void initNetworkReachabilityManager() {
        db.getManager().getContext().getNetworkReachabilityManager().addNetworkReachabilityListener(parentReplication);
    }

    public abstract boolean shouldCreateTarget();

    public abstract void setCreateTarget(boolean createTarget);

    protected void goOnlineInitialStartup() {

        int executorThreadPoolSize = db.getManager().getExecutorThreadPoolSize() <= 0 ?
                EXECUTOR_THREAD_POOL_SIZE : db.getManager().getExecutorThreadPoolSize();
        Log.v(Log.TAG_SYNC, "executorThreadPoolSize=" + executorThreadPoolSize);
        remoteRequestExecutor = Executors.newScheduledThreadPool(executorThreadPoolSize, new ThreadFactory() {
            private int counter = 0;

            @Override
            public Thread newThread(Runnable r) {
                String threadName = "CBLRequestWorker";
                try {
                    String replicationIdentifier = Utils.shortenString(remoteCheckpointDocID(), 5);
                    threadName = String.format("CBLRequestWorker-%s-%s", replicationIdentifier, counter++);
                } catch (Exception e) {
                    Log.e(Log.TAG_SYNC, "Error creating thread name", e);
                }
                return new Thread(r, threadName);
            }
        });

        checkSession();

    }

    @InterfaceAudience.Private
    protected void checkSession() {
        // REVIEW : This is not in line with the iOS implementation
        if (getAuthenticator() != null && ((AuthenticatorImpl) getAuthenticator()).usesCookieBasedLogin()) {
            checkSessionAtPath("/_session");
        } else {
            fetchRemoteCheckpointDoc();
        }
    }

    @InterfaceAudience.Private
    protected void checkSessionAtPath(final String sessionPath) {

        Future future = sendAsyncRequest("GET", sessionPath, null, new RemoteRequestCompletionBlock() {

            @Override
            public void onCompletion(HttpResponse httpResponse, Object result, Throwable err) {

                try {
                    if (err != null) {
                        // If not at /db/_session, try CouchDB location /_session
                        if (err instanceof HttpResponseException &&
                                ((HttpResponseException) err).getStatusCode() == 404 &&
                                "/_session".equalsIgnoreCase(sessionPath)) {

                            checkSessionAtPath("_session");
                            return;
                        }
                        Log.e(Log.TAG_SYNC, this + ": Session check failed", err);
                        setError(err);

                    } else {
                        Map<String, Object> response = (Map<String, Object>) result;
                        Log.e(Log.TAG_SYNC, "%s checkSessionAtPath() response: %s", this, response);
                        Map<String, Object> userCtx = (Map<String, Object>) response.get("userCtx");
                        String username = (String) userCtx.get("name");
                        if (username != null && username.length() > 0) {
                            Log.d(Log.TAG_SYNC, "%s Active session, logged in as %s", this, username);
                            fetchRemoteCheckpointDoc();
                        } else {
                            Log.d(Log.TAG_SYNC, "%s No active session, going to login", this);
                            login();
                        }
                    }

                } catch (Exception e) {
                    Log.e(Log.TAG_SYNC, "%s Exception in checkSessionAtPath()", this, e);
                }

            }

        });
        pendingFutures.add(future);
    }

    @InterfaceAudience.Private
    protected void login() {
        Map<String, String> loginParameters = ((AuthenticatorImpl) getAuthenticator()).loginParametersForSite(remote);
        if (loginParameters == null) {
            Log.d(Log.TAG_SYNC, "%s: %s has no login parameters, so skipping login", this, getAuthenticator());
            fetchRemoteCheckpointDoc();
            return;
        }

        final String loginPath = ((AuthenticatorImpl) getAuthenticator()).loginPathForSite(remote);

        Log.d(Log.TAG_SYNC, "%s: Doing login with %s at %s", this, getAuthenticator().getClass(), loginPath);

        Future future = sendAsyncRequest("POST", loginPath, loginParameters, new RemoteRequestCompletionBlock() {

            @Override
            public void onCompletion(HttpResponse httpResponse, Object result, Throwable e) {
                if (e != null) {
                    Log.d(Log.TAG_SYNC, "%s: Login failed for path: %s", this, loginPath);
                    setError(e);

                    // TODO: double check this behavior against iOS implementation, especially
                    // TODO: with regards to behavior of a continuous replication.
                    // Note: was added in order that unit test testReplicatorErrorStatus() finished and passed.
                    // (before adding this, the replication would just end up in limbo and never finish)
                    triggerStopGraceful();

                } else {
                    Log.v(Log.TAG_SYNC, "%s: Successfully logged in!", this);
                    fetchRemoteCheckpointDoc();
                }
            }

        });
        pendingFutures.add(future);

    }

    @InterfaceAudience.Private
    protected void setError(Throwable throwable) {

        // TODO - needs to port
        //if (error.code == NSURLErrorCancelled && $equal(error.domain, NSURLErrorDomain))
        //    return;

        if (throwable != this.error) {
            Log.e(Log.TAG_SYNC, "%s: Progress: set error = %s", this, throwable);
            parentReplication.setLastError(throwable);
            this.error = throwable;

            // if permanent error, stop immediately
            if(Utils.isPermanentError(this.error) || !isContinuous()) {
                triggerStopGraceful();
            }

            // iOS version sends notification from stop() method, but java version does not.
            // following codes are always executed.
            Replication.ChangeEvent changeEvent = new Replication.ChangeEvent(this);
            changeEvent.setError(this.error);
            notifyChangeListeners(changeEvent);
        }
    }

{'fraction_non_alphanumeric': 0.08695046525881762, 'fraction_numerical': 0.0025073828495013094, 'mean_word_length': 2.331940963519911, 'pattern_counts': {'":': 2, '<': 45, '<?xml version=': 0, '>': 49, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2770170', 'n_tokens_mistral': 9383, 'n_tokens_neox': 8909, 'n_words': 2377}