stream-consumer.js
function executeFinaliseBatchTask(batch, processOutcomes, cancellable, context) {
  /**
   * Defines the finalise batch task (and its sub-tasks) to be used to track the state of the finalising phase.
   * @returns {FinaliseBatchTaskDef} a new finalise batch task definition (with its sub-task definitions)
   */
  function defineFinaliseBatchTask() {
    const taskDef = TaskDef.defineTask(finaliseBatch.name, finaliseBatch, batchAndProcessOutcomesSettings);
    taskDef.defineSubTask(discardAnyRejectedMessages.name, discardAnyRejectedMessages, batchAndCancellableSettings);
    taskDef.defineSubTask(saveBatchState.name, saveBatchState, batchSettings);
    taskDef.defineSubTask(postFinaliseBatch.name, postFinaliseBatch, batchSettings);
    return taskDef;
  }

  /**
   * Creates a new finalise batch task to be used to finalise the batch and to track the state of the finalising phase.
   * @param {Batch} batch - the batch to be processed
   * @param {StreamConsumerContext} context - the context to use
   * @returns {FinaliseBatchTask} a new finalise batch task (with sub-tasks)
   */
  function createFinaliseBatchTask(batch, context) {
    // Define a new finalise task definition for the batch & updates the batch with it
    batch.taskDefs.finaliseTaskDef = defineFinaliseBatchTask();

    const task = context.taskFactory.createTask(batch.taskDefs.finaliseTaskDef, finaliseTaskOpts);

    // Cache it on the batch
    const finalisingTasks = batch.getOrSetFinalisingTasks();
    finalisingTasks[task.name] = task;

    return task;
  }

  // Create a new finalise batch task to handle and track the state of the finalising phase
  const finaliseBatchTask = createFinaliseBatchTask(batch, context);

  // Regardless of whether processing completes or times out, finalise the batch as best as possible
  const p = finaliseBatchTask.execute(batch, processOutcomes, cancellable, context);

  return whenDone(finaliseBatchTask, p, cancellable, context).then(() => batch);
}
function executeFinaliseBatchTask(batch, processOutcomes, cancellable, context) {
  /**
   * Defines the finalise batch task (and its sub-tasks) to be used to track the state of the finalising phase.
   * @returns {FinaliseBatchTaskDef} a new finalise batch task definition (with its sub-task definitions)
   */
  function defineFinaliseBatchTask() {
    const taskDef = TaskDef.defineTask(finaliseBatch.name, finaliseBatch, batchAndProcessOutcomesSettings);
    taskDef.defineSubTask(discardAnyRejectedMessages.name, discardAnyRejectedMessages, batchAndCancellableSettings);
    taskDef.defineSubTask(saveBatchState.name, saveBatchState, batchSettings);
    taskDef.defineSubTask(postFinaliseBatch.name, postFinaliseBatch, batchSettings);
    return taskDef;
  }

  /**
   * Creates a new finalise batch task to be used to finalise the batch and to track the state of the finalising phase.
   * @param {Batch} batch - the batch to be processed
   * @param {StreamConsumerContext} context - the context to use
   * @returns {FinaliseBatchTask} a new finalise batch task (with sub-tasks)
   */
  function createFinaliseBatchTask(batch, context) {
    // Define a new finalise task definition for the batch & updates the batch with it
    batch.taskDefs.finaliseTaskDef = defineFinaliseBatchTask();

    const task = context.taskFactory.createTask(batch.taskDefs.finaliseTaskDef, finaliseTaskOpts);

    // Cache it on the batch
    const finalisingTasks = batch.getOrSetFinalisingTasks();
    finalisingTasks[task.name] = task;

    return task;
  }

  // Create a new finalise batch task to handle and track the state of the finalising phase
  const finaliseBatchTask = createFinaliseBatchTask(batch, context);

  // Regardless of whether processing completes or times out, finalise the batch as best as possible
  const p = finaliseBatchTask.execute(batch, processOutcomes, cancellable, context);

  return whenDone(finaliseBatchTask, p, cancellable, context).then(() => batch);
}
function executeFinaliseBatchTask ( batch , processOutcomes , cancellable , context ) { /**
   * Defines the finalise batch task (and its sub-tasks) to be used to track the state of the finalising phase.
   * @returns {FinaliseBatchTaskDef} a new finalise batch task definition (with its sub-task definitions)
   */ function defineFinaliseBatchTask ( ) { const taskDef = TaskDef . defineTask ( finaliseBatch . name , finaliseBatch , batchAndProcessOutcomesSettings ) ; taskDef . defineSubTask ( discardAnyRejectedMessages . name , discardAnyRejectedMessages , batchAndCancellableSettings ) ; taskDef . defineSubTask ( saveBatchState . name , saveBatchState , batchSettings ) ; taskDef . defineSubTask ( postFinaliseBatch . name , postFinaliseBatch , batchSettings ) ; return taskDef ; } /**
   * Creates a new finalise batch task to be used to finalise the batch and to track the state of the finalising phase.
   * @param {Batch} batch - the batch to be processed
   * @param {StreamConsumerContext} context - the context to use
   * @returns {FinaliseBatchTask} a new finalise batch task (with sub-tasks)
   */ function createFinaliseBatchTask ( batch , context ) { // Define a new finalise task definition for the batch & updates the batch with it batch . taskDefs . finaliseTaskDef = defineFinaliseBatchTask ( ) ; const task = context . taskFactory . createTask ( batch . taskDefs . finaliseTaskDef , finaliseTaskOpts ) ; // Cache it on the batch const finalisingTasks = batch . getOrSetFinalisingTasks ( ) ; finalisingTasks [ task . name ] = task ; return task ; } // Create a new finalise batch task to handle and track the state of the finalising phase const finaliseBatchTask = createFinaliseBatchTask ( batch , context ) ; // Regardless of whether processing completes or times out, finalise the batch as best as possible const p = finaliseBatchTask . execute ( batch , processOutcomes , cancellable , context ) ; return whenDone ( finaliseBatchTask , p , cancellable , context ) . then ( ( ) => batch ) ; }
Creates and executes a new finalise batch task on the given batch.
@param {Batch} batch - the batch to be finalised
@param {Outcomes} processOutcomes - all of the outcomes of the preceding process stage
@param {Cancellable|Object|undefined} [cancellable] - a cancellable object onto which to install cancel functionality
@param {StreamConsumerContext} context - the context to use
@returns {Promise.<Batch>} a promise that will resolve with the batch when the finalise batch task's execution completes
Creates and executes a new finalise batch task on the given batch .