stream-consumer.js
function preProcessBatch(batch, context) {
  const task = this;
  // const initiatingTask = task.parent;

  // Look up the actual function to be used to do the load of the previous tracked state (if any) of the current batch
  const preProcessBatchFn = Settings.getPreProcessBatchFunction(context);

  if (!preProcessBatchFn) {
    if (context.traceEnabled) context.trace(`Skipping pre-process of ${batch.describe(true)}, since no preProcessBatch function configured`);
    return Promise.resolve([]);
  }

  // Create a task to be used to execute and track the configured preProcessBatch function
  const fnName = isNotBlank(preProcessBatchFn.name) ? preProcessBatchFn.name : 'preProcessBatch';
  const preProcessBatchTask = task.createSubTask(fnName, preProcessBatchFn, batchSettings, initiateTaskOpts);

  // Preprocess the batch
  const p = preProcessBatchTask.execute(batch, context);
  return whenDone(preProcessBatchTask, p, undefined, context).then(
    () => {
      if (context.traceEnabled) context.trace(`Pre-processed ${batch.describe(true)}`);
      return batch;
    },
    err => {
      context.error(`Failed to pre-process ${batch.describe(true)}`, err);
      throw err;
    }
  );
}
function preProcessBatch(batch, context) {
  const task = this;
  // const initiatingTask = task.parent;

  // Look up the actual function to be used to do the load of the previous tracked state (if any) of the current batch
  const preProcessBatchFn = Settings.getPreProcessBatchFunction(context);

  if (!preProcessBatchFn) {
    if (context.traceEnabled) context.trace(`Skipping pre-process of ${batch.describe(true)}, since no preProcessBatch function configured`);
    return Promise.resolve([]);
  }

  // Create a task to be used to execute and track the configured preProcessBatch function
  const fnName = isNotBlank(preProcessBatchFn.name) ? preProcessBatchFn.name : 'preProcessBatch';
  const preProcessBatchTask = task.createSubTask(fnName, preProcessBatchFn, batchSettings, initiateTaskOpts);

  // Preprocess the batch
  const p = preProcessBatchTask.execute(batch, context);
  return whenDone(preProcessBatchTask, p, undefined, context).then(
    () => {
      if (context.traceEnabled) context.trace(`Pre-processed ${batch.describe(true)}`);
      return batch;
    },
    err => {
      context.error(`Failed to pre-process ${batch.describe(true)}`, err);
      throw err;
    }
  );
}
function preProcessBatch ( batch , context ) { const task = this ; // const initiatingTask = task.parent; // Look up the actual function to be used to do the load of the previous tracked state (if any) of the current batch const preProcessBatchFn = Settings . getPreProcessBatchFunction ( context ) ; if ( ! preProcessBatchFn ) { if ( context . traceEnabled ) context . trace ( ` ${ batch . describe ( true ) } ` ) ; return Promise . resolve ( [ ] ) ; } // Create a task to be used to execute and track the configured preProcessBatch function const fnName = isNotBlank ( preProcessBatchFn . name ) ? preProcessBatchFn . name : 'preProcessBatch' ; const preProcessBatchTask = task . createSubTask ( fnName , preProcessBatchFn , batchSettings , initiateTaskOpts ) ; // Preprocess the batch const p = preProcessBatchTask . execute ( batch , context ) ; return whenDone ( preProcessBatchTask , p , undefined , context ) . then ( ( ) => { if ( context . traceEnabled ) context . trace ( ` ${ batch . describe ( true ) } ` ) ; return batch ; } , err => { context . error ( ` ${ batch . describe ( true ) } ` , err ) ; throw err ; } ) ; }
Provides a hook for an optional pre-process batch function to be invoked after the batch has been successfully
initiated and before the batch is processed.
@param {Batch} batch
@param {StreamConsumerContext} context
@param {PreFinaliseBatch} [context.preProcessBatch] - an optional pre-process batch function to execute
@returns {Promise.<Batch|undefined>}
Provides a hook for an optional pre - process batch function to be invoked after the batch has been successfully initiated and before the batch is processed .