function syncWithBackend(payload, tries, callback) {
  var datasetClientId = payload.id;
  var datasetId = payload.datasetId;
  var startTime = payload.startTime;

  if (!datasetClientId || !datasetId) {
    recordProcessTime(startTime, false);
    debugError("no datasetId value found in sync request payload %j" ,payload);
    return callback();
  }

  if (tries > 1) {
    //the request is already run once, but for some reason is not acked, we just make sure it's completed and ack it
    markDatasetClientAsCompleted(datasetClientId, startTime, callback);
    return;
  }

  var queryParams = payload.queryParams || {};
  var metaData = payload.metaData || {};
  //we need to add this so that if this sync processor crashed before reaching the end, the scheduler will still be able to push a sync request
  var expectedTimeout = datasets.getDatasetConfig(datasetId).backendListTimeout * 1000 || 5*60*1000;

  async.waterfall([
    function setSyncStart(cb) {
      var syncLoopStartTime = Date.now();
      syncStorage.updateDatasetClient(datasetClientId, {syncLoopStart: syncLoopStartTime, syncLoopEnd: syncLoopStartTime + expectedTimeout}, function(err, datasetClient){
        return cb(err, datasetClient);
      });
    },
    function listData(datasetClient, cb) {
      listDataWithTimeout({datasetClientId: datasetClientId, datasetId: datasetId, queryParams: queryParams, metaData: metaData}, expectedTimeout, function(err, res) {
        return cb(err, datasetClient, res);
      });
    },
    function saveRecords(datasetClient, recordsWithHash, cb) {
      var toHash = _.pluck(recordsWithHash, 'hash');
      var globalHash = hashProvider.globalHash(datasetId, toHash);
      var globalHashWithCollisionCount = [globalHash, datasetClient.collisionCount].join('_');
      syncStorage.updateDatasetClientWithRecords(datasetClientId, {globalHash: globalHashWithCollisionCount}, recordsWithHash, cb);
    }
  ], function(err){
    if (err) {
      debugError("[%s] Error when sync data with backend. error = %s",datasetId,err);
    }

    markDatasetClientAsCompleted(datasetClientId, startTime, callback);
  });
}
function syncWithBackend(payload, tries, callback) {
  var datasetClientId = payload.id;
  var datasetId = payload.datasetId;
  var startTime = payload.startTime;

  if (!datasetClientId || !datasetId) {
    recordProcessTime(startTime, false);
    debugError("no datasetId value found in sync request payload %j" ,payload);
    return callback();
  }

  if (tries > 1) {
    
    markDatasetClientAsCompleted(datasetClientId, startTime, callback);
    return;
  }

  var queryParams = payload.queryParams || {};
  var metaData = payload.metaData || {};
  
  var expectedTimeout = datasets.getDatasetConfig(datasetId).backendListTimeout * 1000 || 5*60*1000;

  async.waterfall([
    function setSyncStart(cb) {
      var syncLoopStartTime = Date.now();
      syncStorage.updateDatasetClient(datasetClientId, {syncLoopStart: syncLoopStartTime, syncLoopEnd: syncLoopStartTime + expectedTimeout}, function(err, datasetClient){
        return cb(err, datasetClient);
      });
    },
    function listData(datasetClient, cb) {
      listDataWithTimeout({datasetClientId: datasetClientId, datasetId: datasetId, queryParams: queryParams, metaData: metaData}, expectedTimeout, function(err, res) {
        return cb(err, datasetClient, res);
      });
    },
    function saveRecords(datasetClient, recordsWithHash, cb) {
      var toHash = _.pluck(recordsWithHash, 'hash');
      var globalHash = hashProvider.globalHash(datasetId, toHash);
      var globalHashWithCollisionCount = [globalHash, datasetClient.collisionCount].join('_');
      syncStorage.updateDatasetClientWithRecords(datasetClientId, {globalHash: globalHashWithCollisionCount}, recordsWithHash, cb);
    }
  ], function(err){
    if (err) {
      debugError("[%s] Error when sync data with backend. error = %s",datasetId,err);
    }

    markDatasetClientAsCompleted(datasetClientId, startTime, callback);
  });
}
function syncWithBackend ( payload , tries , callback ) { var datasetClientId = payload . id ; var datasetId = payload . datasetId ; var startTime = payload . startTime ; if ( ! datasetClientId || ! datasetId ) { recordProcessTime ( startTime , false ) ; debugError ( "no datasetId value found in sync request payload %j" , payload ) ; return callback ( ) ; } if ( tries > 1 ) { //the request is already run once, but for some reason is not acked, we just make sure it's completed and ack it markDatasetClientAsCompleted ( datasetClientId , startTime , callback ) ; return ; } var queryParams = payload . queryParams || { } ; var metaData = payload . metaData || { } ; //we need to add this so that if this sync processor crashed before reaching the end, the scheduler will still be able to push a sync request var expectedTimeout = datasets . getDatasetConfig ( datasetId ) . backendListTimeout * 1000 || 5 * 60 * 1000 ; async . waterfall ( [ function setSyncStart ( cb ) { var syncLoopStartTime = Date . now ( ) ; syncStorage . updateDatasetClient ( datasetClientId , { syncLoopStart : syncLoopStartTime , syncLoopEnd : syncLoopStartTime + expectedTimeout } , function ( err , datasetClient ) { return cb ( err , datasetClient ) ; } ) ; } , function listData ( datasetClient , cb ) { listDataWithTimeout ( { datasetClientId : datasetClientId , datasetId : datasetId , queryParams : queryParams , metaData : metaData } , expectedTimeout , function ( err , res ) { return cb ( err , datasetClient , res ) ; } ) ; } , function saveRecords ( datasetClient , recordsWithHash , cb ) { var toHash = _ . pluck ( recordsWithHash , 'hash' ) ; var globalHash = hashProvider . globalHash ( datasetId , toHash ) ; var globalHashWithCollisionCount = [ globalHash , datasetClient . collisionCount ] . join ( '_' ) ; syncStorage . updateDatasetClientWithRecords ( datasetClientId , { globalHash : globalHashWithCollisionCount } , recordsWithHash , cb ) ; } ] , function ( err ) { if ( err ) { debugError ( "[%s] Error when sync data with backend. error = %s" , datasetId , err ) ; } markDatasetClientAsCompleted ( datasetClientId , startTime , callback ) ; } ) ; }
Perform the sync operation
@param {Object} payload the payload of the sync request
@param {Object} payload.id the id of the dataset client
@param {String} payload.datasetId the id of the dataset that needs to be synced
@param {Object} payload.queryParams the query params that is associated with the dataset
@param {Object} payload.metaData the meta data that is associated with the dataset
@param {Timestamp} payload.startTime when the sync request is created
@param {Number} tries the number of tries of the request
@param {Function} callback
Perform the sync operation