Function tryReplicate

lib/persisted-model.js:1205–1363

(sourceModel, targetModel, since, options, callback)

  };

  function tryReplicate(sourceModel, targetModel, since, options, callback) {
    const Change = sourceModel.getChangeModel();
    const TargetChange = targetModel.getChangeModel();
    const changeTrackingEnabled = Change && TargetChange;
    let replicationChunkSize = REPLICATION_CHUNK_SIZE;

    if (sourceModel.settings && sourceModel.settings.replicationChunkSize) {
      replicationChunkSize = sourceModel.settings.replicationChunkSize;
    }

    assert(
      changeTrackingEnabled,
      'You must enable change tracking before replicating',
    );

    let diff, updates, newSourceCp, newTargetCp;

    const tasks = [
      checkpoints,
      getSourceChanges,
      getDiffFromTarget,
      createSourceUpdates,
      bulkUpdate,
    ];

    async.waterfall(tasks, done);

    function getSourceChanges(cb) {
      utils.downloadInChunks(
        options.filter,
        replicationChunkSize,
        function(filter, pagingCallback) {
          sourceModel.changes(since.source, filter, pagingCallback);
        },
        debug.enabled ? log : cb,
      );

      function log(err, result) {
        if (err) return cb(err);
        debug('\tusing source changes');
        result.forEach(function(it) { debug('\t\t%j', it); });
        cb(err, result);
      }
    }

    function getDiffFromTarget(sourceChanges, cb) {
      utils.uploadInChunks(
        sourceChanges,
        replicationChunkSize,
        function(smallArray, chunkCallback) {
          return targetModel.diff(since.target, smallArray, chunkCallback);
        },
        debug.enabled ? log : cb,
      );

      function log(err, result) {
        if (err) return cb(err);
        if (result.conflicts && result.conflicts.length) {