lib/Master.js
/**
   * Creates a new parallel directive.
   * @return {Object} A parallel object, for building out parallel tasks.
   */
Creates a new parallel directive.
@return {Object} A parallel object, for building out parallel tasks.
Creates a new parallel directive . @return { Object } A parallel object for building out parallel tasks .
createParallel() {
    const tasks = [];
    let addTask = null;
    let removeTask = null;
    let forSlave = null;
    let setTimes = null;
    let times = 1;
    const executeTasks = (...args) => {
      const tasksThisExecution = tasks._.copy();
      const done = args._.getAndStripCallback();
      return new Promise((resolve, reject) => {
        const total = tasksThisExecution.length * times;
        const responses = [];
        let completed = 0;
        let broken = false;
        if (tasksThisExecution.length === 0 || times <= 0) {
          const res = new ResponseArray();
          done.call(this, null, res);
          resolve(res);
          return;
        }
        const singleResponse = times === 1;
        const doIteration = (n) => {
          tasksThisExecution.forEach((t, i) => {
            if (!responses[n]) responses[n] = new ResponseArray();
            if (!(t.slave instanceof Slave)) {
              const e = new TypeError(`Task #${i} is missing a slave. Did you forget to call ".for"?`);
              times = 1;
              done.call(this, e, null);
              reject(e);
              return;
            }
            t.slave.exec(t.command, t.data, t.meta)
              .then(res => {
                responses[n].push(res);
                if (!broken && ++completed === total) {
                  const result = singleResponse ? responses[0] : responses;
                  times = 1;
                  done.call(this, null, result);
                  resolve(result);
                }
              })
              .catch(e => {
                if (!broken) {
                  broken = true;
                  times = 1;
                  done.call(this, e, null);
                  reject(e);
                }
              });
          });
        };
        for (let n = 0; n < times; n++) doIteration(n);
      });
    };
    setTimes = (n) => {
      if (typeof n === 'number') times = n;
      return { execute: executeTasks };
    };
    forSlave = (slave) => {
      slave = Slave.getSlave(slave);
      if (!slave) throw new TypeError('Master#paralle.addTask.for expected argument #0 to be an instanceof Slave.');
      tasks[tasks.length - 1].slave = slave;
      return { addTask, removeTask, execute: executeTasks, times: setTimes, id: tasks[tasks.length - 1].reference };
    };
    addTask = (command, data, meta) => {
      validateCommand(command);
      const id = Symbol();
      tasks.push({ command, data, meta, slave: null, reference: id });
      return { for: forSlave, removeTask, id };
    };
    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: forSlave, removeTask, addTask, removed };
    };
    return { addTask, removeTask, execute: executeTasks, taskCount: () => tasks.length };
  }
createParallel ( ) { const tasks = [ ] ; let addTask = null ; let removeTask = null ; let forSlave = null ; let setTimes = null ; let times = 1 ; const executeTasks = ( ... args ) => { const tasksThisExecution = tasks . _ . copy ( ) ; const done = args . _ . getAndStripCallback ( ) ; return new Promise ( ( resolve , reject ) => { const total = tasksThisExecution . length * times ; const responses = [ ] ; let completed = 0 ; let broken = false ; if ( tasksThisExecution . length === 0 || times <= 0 ) { const res = new ResponseArray ( ) ; done . call ( this , null , res ) ; resolve ( res ) ; return ; } const singleResponse = times === 1 ; const doIteration = ( n ) => { tasksThisExecution . forEach ( ( t , i ) => { if ( ! responses [ n ] ) responses [ n ] = new ResponseArray ( ) ; if ( ! ( t . slave instanceof Slave ) ) { const e = new TypeError ( ` ${ i } ` ) ; times = 1 ; done . call ( this , e , null ) ; reject ( e ) ; return ; } t . slave . exec ( t . command , t . data , t . meta ) . then ( res => { responses [ n ] . push ( res ) ; if ( ! broken && ++ completed === total ) { const result = singleResponse ? responses [ 0 ] : responses ; times = 1 ; done . call ( this , null , result ) ; resolve ( result ) ; } } ) . catch ( e => { if ( ! broken ) { broken = true ; times = 1 ; done . call ( this , e , null ) ; reject ( e ) ; } } ) ; } ) ; } ; for ( let n = 0 ; n < times ; n ++ ) doIteration ( n ) ; } ) ; } ; setTimes = ( n ) => { if ( typeof n === 'number' ) times = n ; return { execute : executeTasks } ; } ; forSlave = ( slave ) => { slave = Slave . getSlave ( slave ) ; if ( ! slave ) throw new TypeError ( 'Master#paralle.addTask.for expected argument #0 to be an instanceof Slave.' ) ; tasks [ tasks . length - 1 ] . slave = slave ; return { addTask , removeTask , execute : executeTasks , times : setTimes , id : tasks [ tasks . length - 1 ] . reference } ; } ; addTask = ( command , data , meta ) => { validateCommand ( command ) ; const id = Symbol ( ) ; tasks . push ( { command , data , meta , slave : null , reference : id } ) ; return { for : forSlave , removeTask , id } ; } ; 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 : forSlave , removeTask , addTask , removed } ; } ; return { addTask , removeTask , execute : executeTasks , taskCount : ( ) => tasks . length } ; }
Creates a new parallel directive.
Creates a new parallel directive .
/**
     * 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.
     */