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();
  });
}
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
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 .