stream-consumer.js
function executeAllProcessAllTasks(batch, cancellable, context) {
  const messages = batch.messages;
  const m = messages.length;
  const ms = toCountString(m, 'message');

  const t = batch.taskDefs.processAllTaskDefs.length;
  const ts = toCountString(t, 'process all task');

  if (m <= 0) {
    if (context.debugEnabled) context.debug(`Skipping execution of ${ts} for ${batch.describe(true)}, since ${ms}`);
    return Promise.resolve([]);
  }
  if (t <= 0) {
    context.warn(`Skipping execution of ${ts} for ${batch.describe(true)}), since ${ts}`);
    return Promise.resolve([]);
  }

  // First get all of the batch's process all tasks
  const tasksByName = batch.getProcessAllTasks();
  const tasks = taskUtils.getTasks(tasksByName);

  const incompleteTasks = tasks.filter(task => !task.isFullyFinalised());

  const i = incompleteTasks.length;

  if (i <= 0) {
    const is = toCountString(i, 'incomplete task');
    if (context.debugEnabled) context.debug(`Skipping execution of ${ts} on ${batch.describe(true)}, since ${is}`);
    return Promise.resolve([]);
  } else {
    if (context.traceEnabled) context.trace(`About to execute ${i} incomplete of ${ts} on ${batch.describe(true)}`);
  }

  // Execute all of the incomplete processAll tasks on the batch and collect their promises
  const promises = incompleteTasks.map(task => {
    // Collect all of the incomplete messages from the batch that are not fully finalised yet for the given task
    const incompleteMessages = messages.filter(msg => {
      const msgTask = batch.getProcessAllTask(msg, task.name);
      return !msgTask || !msgTask.isFullyFinalised();
    });

    const p = task.execute(batch, incompleteMessages, context);
    return whenDone(task, p, cancellable, context);
  });

  return Promises.every(promises, cancellable, context);
}
function executeAllProcessAllTasks(batch, cancellable, context) {
  const messages = batch.messages;
  const m = messages.length;
  const ms = toCountString(m, 'message');

  const t = batch.taskDefs.processAllTaskDefs.length;
  const ts = toCountString(t, 'process all task');

  if (m <= 0) {
    if (context.debugEnabled) context.debug(`Skipping execution of ${ts} for ${batch.describe(true)}, since ${ms}`);
    return Promise.resolve([]);
  }
  if (t <= 0) {
    context.warn(`Skipping execution of ${ts} for ${batch.describe(true)}), since ${ts}`);
    return Promise.resolve([]);
  }

  // First get all of the batch's process all tasks
  const tasksByName = batch.getProcessAllTasks();
  const tasks = taskUtils.getTasks(tasksByName);

  const incompleteTasks = tasks.filter(task => !task.isFullyFinalised());

  const i = incompleteTasks.length;

  if (i <= 0) {
    const is = toCountString(i, 'incomplete task');
    if (context.debugEnabled) context.debug(`Skipping execution of ${ts} on ${batch.describe(true)}, since ${is}`);
    return Promise.resolve([]);
  } else {
    if (context.traceEnabled) context.trace(`About to execute ${i} incomplete of ${ts} on ${batch.describe(true)}`);
  }

  // Execute all of the incomplete processAll tasks on the batch and collect their promises
  const promises = incompleteTasks.map(task => {
    // Collect all of the incomplete messages from the batch that are not fully finalised yet for the given task
    const incompleteMessages = messages.filter(msg => {
      const msgTask = batch.getProcessAllTask(msg, task.name);
      return !msgTask || !msgTask.isFullyFinalised();
    });

    const p = task.execute(batch, incompleteMessages, context);
    return whenDone(task, p, cancellable, context);
  });

  return Promises.every(promises, cancellable, context);
}
function executeAllProcessAllTasks ( batch , cancellable , context ) { const messages = batch . messages ; const m = messages . length ; const ms = toCountString ( m , 'message' ) ; const t = batch . taskDefs . processAllTaskDefs . length ; const ts = toCountString ( t , 'process all task' ) ; if ( m <= 0 ) { if ( context . debugEnabled ) context . debug ( ` ${ ts } ${ batch . describe ( true ) } ${ ms } ` ) ; return Promise . resolve ( [ ] ) ; } if ( t <= 0 ) { context . warn ( ` ${ ts } ${ batch . describe ( true ) } ${ ts } ` ) ; return Promise . resolve ( [ ] ) ; } // First get all of the batch's process all tasks const tasksByName = batch . getProcessAllTasks ( ) ; const tasks = taskUtils . getTasks ( tasksByName ) ; const incompleteTasks = tasks . filter ( task => ! task . isFullyFinalised ( ) ) ; const i = incompleteTasks . length ; if ( i <= 0 ) { const is = toCountString ( i , 'incomplete task' ) ; if ( context . debugEnabled ) context . debug ( ` ${ ts } ${ batch . describe ( true ) } ${ is } ` ) ; return Promise . resolve ( [ ] ) ; } else { if ( context . traceEnabled ) context . trace ( ` ${ i } ${ ts } ${ batch . describe ( true ) } ` ) ; } // Execute all of the incomplete processAll tasks on the batch and collect their promises const promises = incompleteTasks . map ( task => { // Collect all of the incomplete messages from the batch that are not fully finalised yet for the given task const incompleteMessages = messages . filter ( msg => { const msgTask = batch . getProcessAllTask ( msg , task . name ) ; return ! msgTask || ! msgTask . isFullyFinalised ( ) ; } ) ; const p = task . execute ( batch , incompleteMessages , context ) ; return whenDone ( task , p , cancellable , context ) ; } ) ; return Promises . every ( promises , cancellable , context ) ; }
Executes all of the incomplete "process all" tasks on the batch and returns a promise of an array with a Success or
Failure execution outcome for each task.

For each "process all" task to be executed, also resolves and passes all of its incomplete messages, which are all
of the batch's messages that are not yet fully finalised for the particular task. The incomplete messages are passed
as the 2nd argument to the task's `execute` function.

Any and all errors encountered along the way are logged, but no errors are allowed to escape from this function.

@param {Batch} batch - the batch on which to execute each of the "process all" tasks (passed as the 1st argument)
@param {Cancellable|Object|undefined} [cancellable] - a cancellable object onto which to install cancel functionality
@param {StreamConsumerContext} context - the context to use (passed as the 3rd argument)
@return {Promise.<Outcomes>} a promise to return all of the "process all" tasks' outcomes
Executes all of the incomplete process all tasks on the batch and returns a promise of an array with a Success or Failure execution outcome for each task .