lib/sync-server.js
function start(cb) {
  if (arguments.length < 1) throw new Error('start requires 1 argument');

  syncStarted = true;

  if (mongoDbClient === null || redisClient === null) {
    throw new Error('MongoDB Client & Redis Client are not connected. Ensure you have called sync.connect() before calling sync.init()');
  }

  metricsClient = metricsModule.init(syncConfig, redisClient);

  async.series([
    function createQueues(callback) {
      ackQueue = new MongodbQueue('fhsync_ack_queue', metricsClient, syncLock, {mongodb: mongoDbClient, queueMessagesTTL: syncConfig.queueMessagesTTL});
      pendingQueue = new MongodbQueue('fhsync_pending_queue', metricsClient, syncLock, {mongodb: mongoDbClient, queueMessagesTTL: syncConfig.queueMessagesTTL, visibility: syncConfig.pendingWorkerRetryIntervalInSeconds});
      syncQueue = new MongodbQueue('fhsync_queue', metricsClient, syncLock, {mongodb: mongoDbClient, queueMessagesTTL: syncConfig.queueMessagesTTL});

      async.parallel([
        async.apply(ackQueue.create.bind(ackQueue)),
        async.apply(pendingQueue.create.bind(pendingQueue)),
        async.apply(syncQueue.create.bind(syncQueue))
      ], callback);
    },
    function initApis(callback) {
      apiSync = syncApiModule(interceptors, ackQueue, pendingQueue, syncStorage, syncConfig);
      apiSyncRecords = syncRecordsApiModule(syncStorage, pendingQueue, syncConfig);
      return callback();
    },
    function createWorkers(callback) {
      var syncProcessorImpl = syncProcessor(syncStorage, dataHandlers, metricsClient, hashProvider);
      var syncWorkerOpts = {
        name: 'sync_worker',
        interval: syncConfig.syncWorkerInterval,
        backoff: syncConfig.syncWorkerBackoff,
        collectStatsInterval: syncConfig.collectStatsInterval
      };
      for (var i = 0; i < syncConfig.syncWorkerConcurrency; i++) {
        var syncWorker = new Worker(syncQueue, syncProcessorImpl, metricsClient, syncWorkerOpts);
        syncWorkers.push(syncWorker);
      }

      var ackProcessorImpl = ackProcessor(syncStorage);
      var ackWorkerOpts = {
        name: 'ack_worker',
        interval: syncConfig.ackWorkerInterval,
        backoff: syncConfig.ackWorkerBackoff,
        collectStatsInterval: syncConfig.collectStatsInterval
      };
      for (var j = 0; j < syncConfig.ackWorkerConcurrency; j++) {
        var ackWorker = new Worker(ackQueue, ackProcessorImpl, metricsClient, ackWorkerOpts);
        ackWorkers.push(ackWorker);
      }

      var pendingProcessorImpl = pendingProcessor(syncStorage, dataHandlers, hashProvider, metricsClient, syncConfig.pendingWorkerRetryLimit);
      var pendingWorkerOpts = {
        name: 'pending_worker',
        interval: syncConfig.pendingWorkerInterval,
        backoff: syncConfig.pendingWorkerBackoff,
        collectStatsInterval: syncConfig.collectStatsInterval
      };
      for (var k = 0; k < syncConfig.pendingWorkerConcurrency; k++) {
        var pendingWorker = new Worker(pendingQueue, pendingProcessorImpl, metricsClient, pendingWorkerOpts);
        pendingWorkers.push(pendingWorker);
      }

      startAllWorkers(syncWorkers);
      startAllWorkers(ackWorkers);
      startAllWorkers(pendingWorkers);

      return callback();
    },
    function startSyncScheduler(callback) {
      var SyncScheduler = syncSchedulerModule(syncLock, syncStorage, metricsClient).SyncScheduler;
      syncScheduler = new SyncScheduler(syncQueue, {timeBetweenChecks: syncConfig.schedulerInterval, timeBeforeCrashAssumed: syncConfig.schedulerLockMaxTime, syncSchedulerLockName: syncConfig.schedulerLockName});
      syncScheduler.start();
      return callback();
    },
    function startDatasetClientCleaner(callback) {
      var datasetClientCleanerBuilder = datasetClientCleanerModule(syncStorage, syncLock);
      datasetClientCleaner = datasetClientCleanerBuilder({retentionPeriod: syncConfig.datasetClientCleanerRetentionPeriod, checkFrequency: syncConfig.datasetClientCleanerCheckFrequency});
      datasetClientCleaner.start(true, callback);
    }
  ], function(err) {
    if (err) {
      // If there is any problem setting up the necessary sync internals,
      // throw an error to crash the app.
      // This is necessary as it is in an unknown state.
      throw err;
    }
    return cb();
  });
}
function start(cb) {
  if (arguments.length < 1) throw new Error('start requires 1 argument');

  syncStarted = true;

  if (mongoDbClient === null || redisClient === null) {
    throw new Error('MongoDB Client & Redis Client are not connected. Ensure you have called sync.connect() before calling sync.init()');
  }

  metricsClient = metricsModule.init(syncConfig, redisClient);

  async.series([
    function createQueues(callback) {
      ackQueue = new MongodbQueue('fhsync_ack_queue', metricsClient, syncLock, {mongodb: mongoDbClient, queueMessagesTTL: syncConfig.queueMessagesTTL});
      pendingQueue = new MongodbQueue('fhsync_pending_queue', metricsClient, syncLock, {mongodb: mongoDbClient, queueMessagesTTL: syncConfig.queueMessagesTTL, visibility: syncConfig.pendingWorkerRetryIntervalInSeconds});
      syncQueue = new MongodbQueue('fhsync_queue', metricsClient, syncLock, {mongodb: mongoDbClient, queueMessagesTTL: syncConfig.queueMessagesTTL});

      async.parallel([
        async.apply(ackQueue.create.bind(ackQueue)),
        async.apply(pendingQueue.create.bind(pendingQueue)),
        async.apply(syncQueue.create.bind(syncQueue))
      ], callback);
    },
    function initApis(callback) {
      apiSync = syncApiModule(interceptors, ackQueue, pendingQueue, syncStorage, syncConfig);
      apiSyncRecords = syncRecordsApiModule(syncStorage, pendingQueue, syncConfig);
      return callback();
    },
    function createWorkers(callback) {
      var syncProcessorImpl = syncProcessor(syncStorage, dataHandlers, metricsClient, hashProvider);
      var syncWorkerOpts = {
        name: 'sync_worker',
        interval: syncConfig.syncWorkerInterval,
        backoff: syncConfig.syncWorkerBackoff,
        collectStatsInterval: syncConfig.collectStatsInterval
      };
      for (var i = 0; i < syncConfig.syncWorkerConcurrency; i++) {
        var syncWorker = new Worker(syncQueue, syncProcessorImpl, metricsClient, syncWorkerOpts);
        syncWorkers.push(syncWorker);
      }

      var ackProcessorImpl = ackProcessor(syncStorage);
      var ackWorkerOpts = {
        name: 'ack_worker',
        interval: syncConfig.ackWorkerInterval,
        backoff: syncConfig.ackWorkerBackoff,
        collectStatsInterval: syncConfig.collectStatsInterval
      };
      for (var j = 0; j < syncConfig.ackWorkerConcurrency; j++) {
        var ackWorker = new Worker(ackQueue, ackProcessorImpl, metricsClient, ackWorkerOpts);
        ackWorkers.push(ackWorker);
      }

      var pendingProcessorImpl = pendingProcessor(syncStorage, dataHandlers, hashProvider, metricsClient, syncConfig.pendingWorkerRetryLimit);
      var pendingWorkerOpts = {
        name: 'pending_worker',
        interval: syncConfig.pendingWorkerInterval,
        backoff: syncConfig.pendingWorkerBackoff,
        collectStatsInterval: syncConfig.collectStatsInterval
      };
      for (var k = 0; k < syncConfig.pendingWorkerConcurrency; k++) {
        var pendingWorker = new Worker(pendingQueue, pendingProcessorImpl, metricsClient, pendingWorkerOpts);
        pendingWorkers.push(pendingWorker);
      }

      startAllWorkers(syncWorkers);
      startAllWorkers(ackWorkers);
      startAllWorkers(pendingWorkers);

      return callback();
    },
    function startSyncScheduler(callback) {
      var SyncScheduler = syncSchedulerModule(syncLock, syncStorage, metricsClient).SyncScheduler;
      syncScheduler = new SyncScheduler(syncQueue, {timeBetweenChecks: syncConfig.schedulerInterval, timeBeforeCrashAssumed: syncConfig.schedulerLockMaxTime, syncSchedulerLockName: syncConfig.schedulerLockName});
      syncScheduler.start();
      return callback();
    },
    function startDatasetClientCleaner(callback) {
      var datasetClientCleanerBuilder = datasetClientCleanerModule(syncStorage, syncLock);
      datasetClientCleaner = datasetClientCleanerBuilder({retentionPeriod: syncConfig.datasetClientCleanerRetentionPeriod, checkFrequency: syncConfig.datasetClientCleanerCheckFrequency});
      datasetClientCleaner.start(true, callback);
    }
  ], function(err) {
    if (err) {
      // If there is any problem setting up the necessary sync internals,
      // throw an error to crash the app.
      // This is necessary as it is in an unknown state.
      throw err;
    }
    return cb();
  });
}
Starts all sync queues, workers & the sync scheduler.
This should only be called after `connect()`.
If this is not explicitly called before clients send sync requests,
it will be called when a client sends a sync request.
It is OK for this to be called multiple times.

@param {function} cb