lib/Master.js
/**
   * Creates a new executable task pipeline.
   * @return {Object} A pipeline object.
   */
Creates a new executable task pipeline.
@return {Object} A pipeline object.
Creates a new executable task pipeline . @return { Object } A pipeline object .
createPipeline() {
    const tasks = [];
    let task = null;
    let removeTask = null;
    const execute = (data, ...rest) => new Promise((resolve, reject) => {
      const thisRunsTasks = tasks._.copy();
      const done = rest._.getAndStripCallback();
      let meta;
      if (typeof rest[0] === 'object') meta = rest[0];
      const doTask = (taskToPerform, dataForTask, metadataForTask) => {
        if (!(taskToPerform.slave instanceof Slave)) {
          const e = new Error(`Task "${taskToPerform.command}" is missing a slave. Did you forget to chain ".for"?`);
          done.call(this, e, null);
          reject(e);
          return;
        }
        taskToPerform.slave.exec(taskToPerform.command, dataForTask, metadataForTask)
          .then(res => {
            if (res.error) {
              done.call(this, res.error, null);
              return reject(res.error);
            }
            const nextTask = thisRunsTasks.shift();
            let finished = false;
            const end = (value) => {
              finished = true;
              if (value instanceof Error) {
                res.error = value;
              } else if (value !== undefined) {
                res.value = value;
              }
              done.call(this, null, res);
              resolve(res);
            };
            taskToPerform.intercepts._.every(intercept => {
              try {
                return intercept.call(taskToPerform.slave, res, end);
              } catch (e) {
                done.call(this, e, null);
                reject(e);
                finished = true;
                return false; 
              }
            });
            if (nextTask && !finished) {
              doTask(nextTask, res.value, meta);
            } else {
              done.call(this, null, res);
              resolve(res);
            }
            return null;
          })
          .catch(e => {
            done.call(this, e, null);
            reject(e);
          });
      };
      if (tasks.length === 0) {
        done.call(this, null, null);
        resolve(null);
        return;
      }
      doTask(thisRunsTasks.shift(), data, meta);
    });
    const intercept = (callback) => {
      if (callback instanceof Function) tasks[tasks.length - 1].intercepts.push(callback);
      return { addTask: task, removeTask, execute };
    };
    const slave = (s) => {
      const taskSlave = Slave.getSlave(s);
      if (!taskSlave) throw new TypeError('Master#pipeline.task.for.slave requires an instanceof Slave.');
      tasks[tasks.length - 1].slave = s;
      return { intercept, addTask: task, removeTask, execute, id: tasks[tasks.length - 1].reference };
    };
    task = (command) => {
      command = validateCommand(command);
      const taskReferencer = { id: Symbol(), for: slave, removeTask };
      tasks.push({ command, slave: null, intercepts: [], reference: taskReferencer.id });
      return taskReferencer;
    };
    removeTask = (ref) => {
      let comparator = null;
      if (typeof ref === 'symbol') {
        comparator = ref;
      } else if (typeof ref === 'object' && typeof ref.id === 'symbol') {
        comparator = ref.id;
      }
      let removed = false;
      if (comparator) {
        removed = !tasks._.every((t, i) => {
          if (t.reference === comparator) {
            tasks.splice(i, 1);
            return false;
          }
          return true;
        });
      }
      return { for: slave, addTask: task, removeTask, removed };
    };
    task.execute = execute;
    task.removeTask = removeTask;
    return { addTask: task, removeTask, execute, taskCount: () => tasks.length };
  }
createPipeline ( ) { const tasks = [ ] ; let task = null ; let removeTask = null ; const execute = ( data , ... rest ) => new Promise ( ( resolve , reject ) => { const thisRunsTasks = tasks . _ . copy ( ) ; const done = rest . _ . getAndStripCallback ( ) ; let meta ; if ( typeof rest [ 0 ] === 'object' ) meta = rest [ 0 ] ; const doTask = ( taskToPerform , dataForTask , metadataForTask ) => { if ( ! ( taskToPerform . slave instanceof Slave ) ) { const e = new Error ( ` ${ taskToPerform . command } ` ) ; done . call ( this , e , null ) ; reject ( e ) ; return ; } taskToPerform . slave . exec ( taskToPerform . command , dataForTask , metadataForTask ) . then ( res => { if ( res . error ) { done . call ( this , res . error , null ) ; return reject ( res . error ) ; } const nextTask = thisRunsTasks . shift ( ) ; let finished = false ; const end = ( value ) => { finished = true ; if ( value instanceof Error ) { res . error = value ; } else if ( value !== undefined ) { res . value = value ; } done . call ( this , null , res ) ; resolve ( res ) ; } ; taskToPerform . intercepts . _ . every ( intercept => { try { return intercept . call ( taskToPerform . slave , res , end ) ; } catch ( e ) { done . call ( this , e , null ) ; reject ( e ) ; finished = true ; return false ; } } ) ; if ( nextTask && ! finished ) { doTask ( nextTask , res . value , meta ) ; } else { done . call ( this , null , res ) ; resolve ( res ) ; } return null ; } ) . catch ( e => { done . call ( this , e , null ) ; reject ( e ) ; } ) ; } ; if ( tasks . length === 0 ) { done . call ( this , null , null ) ; resolve ( null ) ; return ; } doTask ( thisRunsTasks . shift ( ) , data , meta ) ; } ) ; const intercept = ( callback ) => { if ( callback instanceof Function ) tasks [ tasks . length - 1 ] . intercepts . push ( callback ) ; return { addTask : task , removeTask , execute } ; } ; const slave = ( s ) => { const taskSlave = Slave . getSlave ( s ) ; if ( ! taskSlave ) throw new TypeError ( 'Master#pipeline.task.for.slave requires an instanceof Slave.' ) ; tasks [ tasks . length - 1 ] . slave = s ; return { intercept , addTask : task , removeTask , execute , id : tasks [ tasks . length - 1 ] . reference } ; } ; task = ( command ) => { command = validateCommand ( command ) ; const taskReferencer = { id : Symbol ( ) , for : slave , removeTask } ; tasks . push ( { command , slave : null , intercepts : [ ] , reference : taskReferencer . id } ) ; return taskReferencer ; } ; removeTask = ( ref ) => { let comparator = null ; if ( typeof ref === 'symbol' ) { comparator = ref ; } else if ( typeof ref === 'object' && typeof ref . id === 'symbol' ) { comparator = ref . id ; } let removed = false ; if ( comparator ) { removed = ! tasks . _ . every ( ( t , i ) => { if ( t . reference === comparator ) { tasks . splice ( i , 1 ) ; return false ; } return true ; } ) ; } return { for : slave , addTask : task , removeTask , removed } ; } ; task . execute = execute ; task . removeTask = removeTask ; return { addTask : task , removeTask , execute , taskCount : ( ) => tasks . length } ; }
Creates a new executable task pipeline.
Creates a new executable task pipeline .
/**
     * Executes the pipeline with the given data.
     * @param {*} data The initial data to start the pipeline with.
     * @param {Object} meta The metadata to pass to the pipeline.
     * @return {Promise} A promise for completion.
     */ // Copy the task queue to allow for multi-execution /**
       * Performs the tasks in the task queue recursively until no tasks remain.
       * @param {String|Number|Symbol} taskToPerform A task object to perform.
       * @param {*} dataForTask The data to perform the task with.
       * @param {Object} metadataForTask The metadata to pass to this task.
       * @return {undefined}
       */ // Reject pipeline on error. /**
             * Cuts off pipeline execution.
             * @param {*} value The value to end execution with.
             * @return {undefined}
             */ // Execute all the intercepts... // Break every loop // Perform the next task in the queue... // Execute the first task... /**
     * Allows the user to intercept responses between pipeline sections and modify it.
     * @param {Function} callback The function to execute on task response.
     * @return {Object} Various chaining objects to continue defining the pipeline.
     */ /**
     * Adds a slave to the task queue.
     * @param {Slave|Number|String} s The slave (or slave id, or slave alias) to add.
     * @return {Object} Various chaining objects to continue defining the pipeline.
     */ /**
     * Adds a new task to the pipeline.
     * @param {String|Number|Symbol} command The task command to execute.
     * @return {Object} Various chaining objects to continue defining the pipeline.
     */ /**
     * Removes a task from the pipeline.
     * @param {Symbol|Object<Symbol>} ref The reference to the task to remove.
     * @return {Boolean} True if the task was removed, false otherwise.
     */