stream-consumer.js
function executeProcessBatchTask(batch, cancellable, context) {
  /**
   * Defines the process batch task (and its sub-tasks) to be used to track the state of the processing phase.
   * @returns {ProcessBatchTaskDef} a new process batch task definition (with its sub-task definitions)
   */
  function defineProcessBatchTask() {
    const taskDef = TaskDef.defineTask(processBatch.name, processBatch, batchAndCancellableSettings);
    taskDef.defineSubTask(executeAllProcessOneTasks.name, executeAllProcessOneTasks, batchAndCancellableSettings);
    taskDef.defineSubTask(executeAllProcessAllTasks.name, executeAllProcessAllTasks, batchAndCancellableSettings);
    taskDef.defineSubTask(discardUnusableRecords.name, discardUnusableRecords, batchAndUnusableRecordsSettings);
    taskDef.defineSubTask(preFinaliseBatch.name, preFinaliseBatch, batchSettings);
    return taskDef;
  }

  /**
   * Creates a new process batch task to be used to process the batch and to strack the state of the processing phase.
   * @param {Batch} batch - the batch to be processed
   * @param {StreamConsumerContext} context - the context to use
   * @returns {ProcessBatchTask} a new process batch task (with sub-tasks)
   */
  function createProcessBatchTask(batch, context) {
    // Define a new process task definition for the batch & updates the batch with it
    batch.taskDefs.processTaskDef = defineProcessBatchTask();

    const task = context.taskFactory.createTask(batch.taskDefs.processTaskDef, processTaskOpts);

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

    return task;
  }

  // Create a new process batch task to handle and track the state of the processing phase
  const processBatchTask = createProcessBatchTask(batch, context);

  // Process the batch
  const p = processBatchTask.execute(batch, cancellable, context);
  return whenDone(processBatchTask, p, cancellable, context);
  // const donePromise = whenDone(processBatchTask, p, cancellable, context);
  // handleUnhandledRejections([p, donePromise], context);
  // return donePromise;
}
function executeProcessBatchTask(batch, cancellable, context) {
  /**
   * Defines the process batch task (and its sub-tasks) to be used to track the state of the processing phase.
   * @returns {ProcessBatchTaskDef} a new process batch task definition (with its sub-task definitions)
   */
  function defineProcessBatchTask() {
    const taskDef = TaskDef.defineTask(processBatch.name, processBatch, batchAndCancellableSettings);
    taskDef.defineSubTask(executeAllProcessOneTasks.name, executeAllProcessOneTasks, batchAndCancellableSettings);
    taskDef.defineSubTask(executeAllProcessAllTasks.name, executeAllProcessAllTasks, batchAndCancellableSettings);
    taskDef.defineSubTask(discardUnusableRecords.name, discardUnusableRecords, batchAndUnusableRecordsSettings);
    taskDef.defineSubTask(preFinaliseBatch.name, preFinaliseBatch, batchSettings);
    return taskDef;
  }

  /**
   * Creates a new process batch task to be used to process the batch and to strack the state of the processing phase.
   * @param {Batch} batch - the batch to be processed
   * @param {StreamConsumerContext} context - the context to use
   * @returns {ProcessBatchTask} a new process batch task (with sub-tasks)
   */
  function createProcessBatchTask(batch, context) {
    // Define a new process task definition for the batch & updates the batch with it
    batch.taskDefs.processTaskDef = defineProcessBatchTask();

    const task = context.taskFactory.createTask(batch.taskDefs.processTaskDef, processTaskOpts);

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

    return task;
  }

  // Create a new process batch task to handle and track the state of the processing phase
  const processBatchTask = createProcessBatchTask(batch, context);

  // Process the batch
  const p = processBatchTask.execute(batch, cancellable, context);
  return whenDone(processBatchTask, p, cancellable, context);
  // const donePromise = whenDone(processBatchTask, p, cancellable, context);
  // handleUnhandledRejections([p, donePromise], context);
  // return donePromise;
}
function executeProcessBatchTask ( batch , cancellable , context ) { /**
   * Defines the process batch task (and its sub-tasks) to be used to track the state of the processing phase.
   * @returns {ProcessBatchTaskDef} a new process batch task definition (with its sub-task definitions)
   */ function defineProcessBatchTask ( ) { const taskDef = TaskDef . defineTask ( processBatch . name , processBatch , batchAndCancellableSettings ) ; taskDef . defineSubTask ( executeAllProcessOneTasks . name , executeAllProcessOneTasks , batchAndCancellableSettings ) ; taskDef . defineSubTask ( executeAllProcessAllTasks . name , executeAllProcessAllTasks , batchAndCancellableSettings ) ; taskDef . defineSubTask ( discardUnusableRecords . name , discardUnusableRecords , batchAndUnusableRecordsSettings ) ; taskDef . defineSubTask ( preFinaliseBatch . name , preFinaliseBatch , batchSettings ) ; return taskDef ; } /**
   * Creates a new process batch task to be used to process the batch and to strack the state of the processing phase.
   * @param {Batch} batch - the batch to be processed
   * @param {StreamConsumerContext} context - the context to use
   * @returns {ProcessBatchTask} a new process batch task (with sub-tasks)
   */ function createProcessBatchTask ( batch , context ) { // Define a new process task definition for the batch & updates the batch with it batch . taskDefs . processTaskDef = defineProcessBatchTask ( ) ; const task = context . taskFactory . createTask ( batch . taskDefs . processTaskDef , processTaskOpts ) ; // Cache it on the batch const processingTasks = batch . getOrSetProcessingTasks ( ) ; processingTasks [ task . name ] = task ; return task ; } // Create a new process batch task to handle and track the state of the processing phase const processBatchTask = createProcessBatchTask ( batch , context ) ; // Process the batch const p = processBatchTask . execute ( batch , cancellable , context ) ; return whenDone ( processBatchTask , p , cancellable , context ) ; // const donePromise = whenDone(processBatchTask, p, cancellable, context); // handleUnhandledRejections([p, donePromise], context); // return donePromise; }
Creates and executes a new process batch task on the given batch.
@param {Batch} batch - the batch to be processed
@param {Cancellable|Object|undefined} [cancellable] - a cancellable object onto which to install cancel functionality
@param {StreamConsumerContext} context - the context to use
@returns {Promise.<Outcomes>} a promise that will resolve with the outcomes of the process batch task's execution
Creates and executes a new process batch task on the given batch .