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

import com.couchbase.lite.CouchbaseLiteException;
import com.couchbase.lite.Database;
import com.couchbase.lite.Document;
import com.couchbase.lite.Manager;
import com.couchbase.lite.Misc;
import com.couchbase.lite.RevisionList;
import com.couchbase.lite.Status;
import com.couchbase.lite.TransactionalTask;
import com.couchbase.lite.internal.InterfaceAudience;
import com.couchbase.lite.internal.RevisionInternal;
import com.couchbase.lite.storage.SQLException;
import com.couchbase.lite.support.BatchProcessor;
import com.couchbase.lite.support.Batcher;
import com.couchbase.lite.support.CustomFuture;
import com.couchbase.lite.support.HttpClientFactory;
import com.couchbase.lite.support.RemoteRequestCompletionBlock;
import com.couchbase.lite.support.SequenceMap;
import com.couchbase.lite.util.CollectionUtils;
import com.couchbase.lite.util.Log;
import com.couchbase.lite.util.URIUtils;
import com.couchbase.lite.util.Utils;

import org.apache.http.HttpResponse;
import org.apache.http.client.HttpClient;
import org.apache.http.client.HttpResponseException;

import java.net.URL;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * Pull Replication
 *
 * @exclude
 */
@InterfaceAudience.Private
public class PullerInternal extends ReplicationInternal implements ChangeTrackerClient {

    private static final String TAG = Log.TAG_SYNC;

    private static final int MAX_OPEN_HTTP_CONNECTIONS = 16;

    // Maximum number of revs to fetch in a single bulk request
    public static final int MAX_REVS_TO_GET_IN_BULK = 50;

    // Maximum number of revision IDs to pass in an "?atts_since=" query param
    public static final int MAX_NUMBER_OF_ATTS_SINCE = 50;

    // Tune this parameter based on application needs.
    public static int CHANGE_TRACKER_RESTART_DELAY_MS = 10 * 1000;

    public static final int MAX_PENDING_DOCS = 200;

    private static final int INSERTION_BATCHER_DELAY = 250; // 0.25 Seconds

    private ChangeTracker changeTracker;
    protected SequenceMap pendingSequences;
    protected Boolean canBulkGet;  // Does the server support _bulk_get requests?
    protected List<RevisionInternal> revsToPull = Collections.synchronizedList(
            new ArrayList<RevisionInternal>(100));
    protected List<RevisionInternal> bulkRevsToPull = Collections.synchronizedList(
            new ArrayList<RevisionInternal>(100));
    protected List<RevisionInternal> deletedRevsToPull = Collections.synchronizedList(
            new ArrayList<RevisionInternal>(100));
    protected int httpConnectionCount;
    protected Batcher<RevisionInternal> downloadsToInsert;

    private String str = null;

    public PullerInternal(Database db,
                          URL remote,
                          HttpClientFactory clientFactory,
                          ScheduledExecutorService workExecutor,
                          Replication.Lifecycle lifecycle,
                          Replication parentReplication) {
        super(db, remote, clientFactory, workExecutor, lifecycle, parentReplication);
    }

    /**
     * Actual work of starting the replication process.
     */
    protected void beginReplicating() {
        Log.v(TAG, "submit startReplicating()");
        workExecutor.submit(new Runnable() {
            @Override
            public void run() {
                if(isRunning()) {
                    Log.v(TAG, "start startReplicating()");
                    initPendingSequences();
                    initDownloadsToInsert();
                    startChangeTracker();
                }
                // start replicator ...
            }
        });
    }

    private void initDownloadsToInsert() {
        if (downloadsToInsert == null) {
            int capacity = 200;
            downloadsToInsert = new Batcher<RevisionInternal>(workExecutor,
                    capacity, INSERTION_BATCHER_DELAY, new BatchProcessor<RevisionInternal>() {
                @Override
                public void process(List<RevisionInternal> inbox) {
                    insertDownloads(inbox);
                }
            });
        }
    }

    public boolean isPull() {
        return true;
    }

    protected void maybeCreateRemoteDB() {
        // puller never needs to do this
    }

    protected void startChangeTracker() {

        ChangeTracker.ChangeTrackerMode changeTrackerMode;

        // it always starts out as OneShot, but if its a continuous replication
        // it will switch to longpoll later.
        changeTrackerMode = ChangeTracker.ChangeTrackerMode.OneShot;

        Log.d(TAG, "%s: starting ChangeTracker with since=%s mode=%s",
                this, lastSequence, changeTrackerMode);
        changeTracker = new ChangeTracker(remote, changeTrackerMode, true, lastSequence, this);
        changeTracker.setAuthenticator(getAuthenticator());
        Log.d(TAG, "%s: started ChangeTracker %s", this, changeTracker);
<commit_msg>Fixed #1112 - Multiple ChangeTrackers are started (Scenario 2)<commit_after>package com.couchbase.lite.replicator;

import com.couchbase.lite.CouchbaseLiteException;
import com.couchbase.lite.Database;
import com.couchbase.lite.Document;
import com.couchbase.lite.Manager;
import com.couchbase.lite.Misc;
import com.couchbase.lite.RevisionList;
import com.couchbase.lite.Status;
import com.couchbase.lite.TransactionalTask;
import com.couchbase.lite.internal.InterfaceAudience;
import com.couchbase.lite.internal.RevisionInternal;
import com.couchbase.lite.storage.SQLException;
import com.couchbase.lite.support.BatchProcessor;
import com.couchbase.lite.support.Batcher;
import com.couchbase.lite.support.CustomFuture;
import com.couchbase.lite.support.HttpClientFactory;
import com.couchbase.lite.support.RemoteRequestCompletionBlock;
import com.couchbase.lite.support.SequenceMap;
import com.couchbase.lite.util.CollectionUtils;
import com.couchbase.lite.util.Log;
import com.couchbase.lite.util.URIUtils;
import com.couchbase.lite.util.Utils;

import org.apache.http.HttpResponse;
import org.apache.http.client.HttpClient;
import org.apache.http.client.HttpResponseException;

import java.net.URL;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * Pull Replication
 *
 * @exclude
 */
@InterfaceAudience.Private
public class PullerInternal extends ReplicationInternal implements ChangeTrackerClient {

    private static final String TAG = Log.TAG_SYNC;

    private static final int MAX_OPEN_HTTP_CONNECTIONS = 16;

    // Maximum number of revs to fetch in a single bulk request
    public static final int MAX_REVS_TO_GET_IN_BULK = 50;

    // Maximum number of revision IDs to pass in an "?atts_since=" query param
    public static final int MAX_NUMBER_OF_ATTS_SINCE = 50;

    // Tune this parameter based on application needs.
    public static int CHANGE_TRACKER_RESTART_DELAY_MS = 10 * 1000;

    public static final int MAX_PENDING_DOCS = 200;

    private static final int INSERTION_BATCHER_DELAY = 250; // 0.25 Seconds

    private ChangeTracker changeTracker;
    protected SequenceMap pendingSequences;
    protected Boolean canBulkGet;  // Does the server support _bulk_get requests?
    protected List<RevisionInternal> revsToPull = Collections.synchronizedList(
            new ArrayList<RevisionInternal>(100));
    protected List<RevisionInternal> bulkRevsToPull = Collections.synchronizedList(
            new ArrayList<RevisionInternal>(100));
    protected List<RevisionInternal> deletedRevsToPull = Collections.synchronizedList(
            new ArrayList<RevisionInternal>(100));
    protected int httpConnectionCount;
    protected Batcher<RevisionInternal> downloadsToInsert;

    private String str = null;

    public PullerInternal(Database db,
                          URL remote,
                          HttpClientFactory clientFactory,
                          ScheduledExecutorService workExecutor,
                          Replication.Lifecycle lifecycle,
                          Replication parentReplication) {
        super(db, remote, clientFactory, workExecutor, lifecycle, parentReplication);
    }

    /**
     * Actual work of starting the replication process.
     */
    protected void beginReplicating() {
        Log.v(TAG, "submit startReplicating()");
        workExecutor.submit(new Runnable() {
            @Override
            public void run() {
                if(isRunning()) {
                    Log.v(TAG, "start startReplicating()");
                    initPendingSequences();
                    initDownloadsToInsert();
                    startChangeTracker();
                }
                // start replicator ...
            }
        });
    }

    private void initDownloadsToInsert() {
        if (downloadsToInsert == null) {
            int capacity = 200;
            downloadsToInsert = new Batcher<RevisionInternal>(workExecutor,
                    capacity, INSERTION_BATCHER_DELAY, new BatchProcessor<RevisionInternal>() {
                @Override
                public void process(List<RevisionInternal> inbox) {
                    insertDownloads(inbox);
                }
            });
        }
    }

    public boolean isPull() {
        return true;
    }

    protected void maybeCreateRemoteDB() {
        // puller never needs to do this
    }

    protected void startChangeTracker() {
        // make sure not start new changeTracker if pull replicator is not running or idle
        if (!(stateMachine.isInState(ReplicationState.RUNNING) ||
                stateMachine.isInState(ReplicationState.IDLE)))
            return;

        ChangeTracker.ChangeTrackerMode changeTrackerMode;

        // it always starts out as OneShot, but if its a continuous replication
        // it will switch to longpoll later.
        changeTrackerMode = ChangeTracker.ChangeTrackerMode.OneShot;

        Log.d(TAG, "%s: starting ChangeTracker with since=%s mode=%s",
                this, lastSequence, changeTrackerMode);
        changeTracker = new ChangeTracker(remote, changeTrackerMode, true, lastSequence, this);
        changeTracker.setAuthenticator(getAuthenticator());
        Log.d(TAG, "%s: started ChangeTracker %s", this, changeTracker);

{'fraction_non_alphanumeric': 0.08158039143954637, 'fraction_numerical': 0.006493506493506494, 'mean_word_length': 3.017266715650257, 'pattern_counts': {'":': 0, '<': 23, '<?xml version=': 0, '>': 23, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 12, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3749781', 'n_tokens_mistral': 3096, 'n_tokens_neox': 2865, 'n_words': 699}