stream-consumer.js
function executeInitiateBatchTask(batch, cancellable, context) {
  /**
   * Defines the initiate batch task (and its sub-tasks) to be used to track the state of the initiating phase.
   * @returns {InitiateBatchTaskDef} a new initiate batch task definition (with its sub-task definitions)
   */
  function defineInitiateBatchTask() {
    const taskDef = TaskDef.defineTask(initiateBatch.name, initiateBatch, batchAndCancellableSettings);
    taskDef.defineSubTask(extractAndSequenceMessages.name, extractAndSequenceMessages, batchSettings);
    taskDef.defineSubTask(loadBatchState.name, loadBatchState, batchSettings);
    taskDef.defineSubTask(reviveTasks.name, reviveTasks, batchSettings);
    taskDef.defineSubTask(preProcessBatch.name, preProcessBatch, batchSettings);
    return taskDef;
  }

  /**
   * Creates a new initiate batch task to be used to initiate the batch and to track the state of the initiating
   * (pre-processing) phase.
   * @param {Batch} batch - the batch to be initiated
   * @param {StreamConsumerContext} context - the context to use
   * @returns {InitiateBatchTask} a new initiate batch task (with sub-tasks)
   */
  function createInitiateBatchTask(batch, context) {
    // Define a new initiate batch task definition for the batch & update the batch with it
    batch.taskDefs.initiateTaskDef = defineInitiateBatchTask();

    // Create a new initiate batch task (and all of its sub-tasks) from the task definition
    const task = context.taskFactory.createTask(batch.taskDefs.initiateTaskDef, initiateTaskOpts);

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

    return task;
  }

  // Create a new initiate batch task to handle and track the state of the initiating phase
  const initiateBatchTask = createInitiateBatchTask(batch, context);

  // Initiate the batch
  const p = initiateBatchTask.execute(batch, cancellable, context);

  // Terminate the initiate batch task with its first Failure outcome (if any); otherwise return its successful values
  return whenDone(initiateBatchTask, p, cancellable, context);
  // const donePromise = whenDone(initiateBatchTask, p, cancellable, context);
  // handleUnhandledRejections([p, donePromise], context);
  // return donePromise;
}
function executeInitiateBatchTask(batch, cancellable, context) {
  /**
   * Defines the initiate batch task (and its sub-tasks) to be used to track the state of the initiating phase.
   * @returns {InitiateBatchTaskDef} a new initiate batch task definition (with its sub-task definitions)
   */
  function defineInitiateBatchTask() {
    const taskDef = TaskDef.defineTask(initiateBatch.name, initiateBatch, batchAndCancellableSettings);
    taskDef.defineSubTask(extractAndSequenceMessages.name, extractAndSequenceMessages, batchSettings);
    taskDef.defineSubTask(loadBatchState.name, loadBatchState, batchSettings);
    taskDef.defineSubTask(reviveTasks.name, reviveTasks, batchSettings);
    taskDef.defineSubTask(preProcessBatch.name, preProcessBatch, batchSettings);
    return taskDef;
  }

  /**
   * Creates a new initiate batch task to be used to initiate the batch and to track the state of the initiating
   * (pre-processing) phase.
   * @param {Batch} batch - the batch to be initiated
   * @param {StreamConsumerContext} context - the context to use
   * @returns {InitiateBatchTask} a new initiate batch task (with sub-tasks)
   */
  function createInitiateBatchTask(batch, context) {
    // Define a new initiate batch task definition for the batch & update the batch with it
    batch.taskDefs.initiateTaskDef = defineInitiateBatchTask();

    // Create a new initiate batch task (and all of its sub-tasks) from the task definition
    const task = context.taskFactory.createTask(batch.taskDefs.initiateTaskDef, initiateTaskOpts);

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

    return task;
  }

  // Create a new initiate batch task to handle and track the state of the initiating phase
  const initiateBatchTask = createInitiateBatchTask(batch, context);

  // Initiate the batch
  const p = initiateBatchTask.execute(batch, cancellable, context);

  // Terminate the initiate batch task with its first Failure outcome (if any); otherwise return its successful values
  return whenDone(initiateBatchTask, p, cancellable, context);
  // const donePromise = whenDone(initiateBatchTask, p, cancellable, context);
  // handleUnhandledRejections([p, donePromise], context);
  // return donePromise;
}
function executeInitiateBatchTask ( batch , cancellable , context ) { /**
   * Defines the initiate batch task (and its sub-tasks) to be used to track the state of the initiating phase.
   * @returns {InitiateBatchTaskDef} a new initiate batch task definition (with its sub-task definitions)
   */ function defineInitiateBatchTask ( ) { const taskDef = TaskDef . defineTask ( initiateBatch . name , initiateBatch , batchAndCancellableSettings ) ; taskDef . defineSubTask ( extractAndSequenceMessages . name , extractAndSequenceMessages , batchSettings ) ; taskDef . defineSubTask ( loadBatchState . name , loadBatchState , batchSettings ) ; taskDef . defineSubTask ( reviveTasks . name , reviveTasks , batchSettings ) ; taskDef . defineSubTask ( preProcessBatch . name , preProcessBatch , batchSettings ) ; return taskDef ; } /**
   * Creates a new initiate batch task to be used to initiate the batch and to track the state of the initiating
   * (pre-processing) phase.
   * @param {Batch} batch - the batch to be initiated
   * @param {StreamConsumerContext} context - the context to use
   * @returns {InitiateBatchTask} a new initiate batch task (with sub-tasks)
   */ function createInitiateBatchTask ( batch , context ) { // Define a new initiate batch task definition for the batch & update the batch with it batch . taskDefs . initiateTaskDef = defineInitiateBatchTask ( ) ; // Create a new initiate batch task (and all of its sub-tasks) from the task definition const task = context . taskFactory . createTask ( batch . taskDefs . initiateTaskDef , initiateTaskOpts ) ; // Cache it on the batch const initiatingTasks = batch . getOrSetInitiatingTasks ( ) ; initiatingTasks [ task . name ] = task ; return task ; } // Create a new initiate batch task to handle and track the state of the initiating phase const initiateBatchTask = createInitiateBatchTask ( batch , context ) ; // Initiate the batch const p = initiateBatchTask . execute ( batch , cancellable , context ) ; // Terminate the initiate batch task with its first Failure outcome (if any); otherwise return its successful values return whenDone ( initiateBatchTask , p , cancellable , context ) ; // const donePromise = whenDone(initiateBatchTask, p, cancellable, context); // handleUnhandledRejections([p, donePromise], context); // return donePromise; }
Creates and executes a new initiate batch task on the given batch.
@param {Batch} batch - the batch to be initiated
@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 updated batch when the initiate batch task's execution completes
Creates and executes a new initiate batch task on the given batch .