stream-consumer.js
function createCompletedPromise(task, completingPromise, batch, timeoutCancellable, context) {
  const mustResolve = false;

  return completingPromise.then(
    outcomes => {
      // The completing promise has completed
      // 1. Try to cancel the timeout with which the completing promise was racing
      const timedOut = timeoutCancellable.cancelTimeout(mustResolve);
      if (timedOut) {
        context.warn(`Timed out before ${task.name} completed for ${batch.describe(true)}`);
      } else {
        // 2. Mark the completing task as failed if any of its outcomes (or any of its sub-tasks' outcomes) is a Failure
        const failure = Try.findFailure(outcomes);
        if (failure) {
          task.fail(failure.error, false);
        } else {
          task.complete(outcomes, dontOverrideTimedOut, false);
        }
        context.debug(`Completed ${task.name} for ${batch.describe(true)} - final state (${task.state})`);
      }
      return outcomes;
    },
    err => {
      if (isInstanceOf(err, CancelledError)) {
        // The completing promise was cancelled, which means the timeout must have triggered and cancelled it
        context.warn(`Cancelling ${task.name} for ${batch.describe(true)}`, err);
        const timeoutOpts = {overrideCompleted: false, overrideUnstarted: false, reverseAttempt: true};
        const timeoutError = new TimeoutError(`Ran out of time to complete ${task.name}`);
        if (!task.timedOut) {
          task.timeout(timeoutError, timeoutOpts, true);
        }
        // timeoutError.cancelledError = err;
        // throw timeoutError;
        throw err;

      } else {
        // The completing promise rejected with a non-cancelled error, so attempt to cancel the timeout promise
        const timedOut = timeoutCancellable.cancelTimeout(mustResolve);
        if (timedOut) {
          context.warn(`Timed out before failed ${task.name} for ${batch.describe(true)}`, err);
        } else {
          context.info(`Failed ${task.name} for ${batch.describe(true)}`, err);
          task.fail(err, false);
        }
        throw err;
      }
    }
  );
}
function createCompletedPromise(task, completingPromise, batch, timeoutCancellable, context) {
  const mustResolve = false;

  return completingPromise.then(
    outcomes => {
      // The completing promise has completed
      // 1. Try to cancel the timeout with which the completing promise was racing
      const timedOut = timeoutCancellable.cancelTimeout(mustResolve);
      if (timedOut) {
        context.warn(`Timed out before ${task.name} completed for ${batch.describe(true)}`);
      } else {
        // 2. Mark the completing task as failed if any of its outcomes (or any of its sub-tasks' outcomes) is a Failure
        const failure = Try.findFailure(outcomes);
        if (failure) {
          task.fail(failure.error, false);
        } else {
          task.complete(outcomes, dontOverrideTimedOut, false);
        }
        context.debug(`Completed ${task.name} for ${batch.describe(true)} - final state (${task.state})`);
      }
      return outcomes;
    },
    err => {
      if (isInstanceOf(err, CancelledError)) {
        // The completing promise was cancelled, which means the timeout must have triggered and cancelled it
        context.warn(`Cancelling ${task.name} for ${batch.describe(true)}`, err);
        const timeoutOpts = {overrideCompleted: false, overrideUnstarted: false, reverseAttempt: true};
        const timeoutError = new TimeoutError(`Ran out of time to complete ${task.name}`);
        if (!task.timedOut) {
          task.timeout(timeoutError, timeoutOpts, true);
        }
        // timeoutError.cancelledError = err;
        // throw timeoutError;
        throw err;

      } else {
        // The completing promise rejected with a non-cancelled error, so attempt to cancel the timeout promise
        const timedOut = timeoutCancellable.cancelTimeout(mustResolve);
        if (timedOut) {
          context.warn(`Timed out before failed ${task.name} for ${batch.describe(true)}`, err);
        } else {
          context.info(`Failed ${task.name} for ${batch.describe(true)}`, err);
          task.fail(err, false);
        }
        throw err;
      }
    }
  );
}
function createCompletedPromise ( task , completingPromise , batch , timeoutCancellable , context ) { const mustResolve = false ; return completingPromise . then ( outcomes => { // The completing promise has completed // 1. Try to cancel the timeout with which the completing promise was racing const timedOut = timeoutCancellable . cancelTimeout ( mustResolve ) ; if ( timedOut ) { context . warn ( ` ${ task . name } ${ batch . describe ( true ) } ` ) ; } else { // 2. Mark the completing task as failed if any of its outcomes (or any of its sub-tasks' outcomes) is a Failure const failure = Try . findFailure ( outcomes ) ; if ( failure ) { task . fail ( failure . error , false ) ; } else { task . complete ( outcomes , dontOverrideTimedOut , false ) ; } context . debug ( ` ${ task . name } ${ batch . describe ( true ) } ${ task . state } ` ) ; } return outcomes ; } , err => { if ( isInstanceOf ( err , CancelledError ) ) { // The completing promise was cancelled, which means the timeout must have triggered and cancelled it context . warn ( ` ${ task . name } ${ batch . describe ( true ) } ` , err ) ; const timeoutOpts = { overrideCompleted : false , overrideUnstarted : false , reverseAttempt : true } ; const timeoutError = new TimeoutError ( ` ${ task . name } ` ) ; if ( ! task . timedOut ) { task . timeout ( timeoutError , timeoutOpts , true ) ; } // timeoutError.cancelledError = err; // throw timeoutError; throw err ; } else { // The completing promise rejected with a non-cancelled error, so attempt to cancel the timeout promise const timedOut = timeoutCancellable . cancelTimeout ( mustResolve ) ; if ( timedOut ) { context . warn ( ` ${ task . name } ${ batch . describe ( true ) } ` , err ) ; } else { context . info ( ` ${ task . name } ${ batch . describe ( true ) } ` , err ) ; task . fail ( err , false ) ; } throw err ; } } ) ; }
Build a completed promise that will only complete when the given completingPromise has completed.
@param {Task} task - a task to be completed or failed depending on the outcome of the promise and if the timeout has not triggered yet
@param {Promise} completingPromise - the promise that will complete when all processing has been completed for the current phase
@param {Batch} batch - the batch being processed
@param {Object} timeoutCancellable - a cancellable object that enables cancellation of the timeout promise on completion
@param {StreamConsumerContext} context - the context to use
@returns {Promise.<Outcomes>} a promise to return the outcomes of the completing promise when the completingPromise resolves
Build a completed promise that will only complete when the given completingPromise has completed .