Example of making a complex runner capable of limited concurrency, disjoint continuation, and mixed concurrency/parallelism

runner.js

// an example of using async / await style scheduling using callbacks
// we can do things without promises and get some nicer (imho) results

// promises still work here, but are not shown for advanced cases
function* task(step) {
  
  //
  // simple callback style continuation
  //
  yield _ => setTimeout(_,0);
  
  
  
  //
  // simple promise continuation
  //
  yield new Promise((f,r) => {
    setTimeout(f, 0);                  
  });

  
  
  //
  // disjoint style continuation
  //
  // yielding the step function tells the runner that we have it covered for
  // this step, important for event listener style programming
  //
  setTimeout(step, 0);
  yield step;
  
  //
  // simple parallel tasks (always continuation style, to keep keys correct)
  //
  
  // indexed
  let [time100,time200s] = yield [
    _ => setTimeout(()=>_(null, Date.now()),100),
    _ => setTimeout(()=>_(null, Date.now()),200),
    _ => setTimeout(()=>_(null, Date.now()),200)
  ]
  
  console.log('indexed task', [time100,time200s]);
  
  
  
  // keyed
  let times = yield {
    time100: _ => setTimeout(()=>_(null, Date.now),100),
    time200: _ => setTimeout(()=>_(null, Date.now),200)
  };
 
  console.log('keyed task', times);
  
  
  
  // nested parallelism
  let nested = yield [
    {
      a: _ => setTimeout(()=>_(null, Date.now),0),
    },
    {
      b: _ => setTimeout(()=>_(null, Date.now),0),
    },
  ];
    
  console.log('nested task', nested);
  
    
    
  //
  // advanced uses
  //
    
  // mixed concurrency and parallelism
  let mixed = yield {
    series: _ => runner(function* (my_step) {
      yield _ => setTimeout(_,0)
      return 'series';
    }, _),
    parallel: [
      _ => setTimeout(()=>_(null, Date.now),0),
      _ => setTimeout(()=>_(null, Date.now),0),
    ]
  };
  
  console.log('mixed task', mixed);
    


  // task chaining
  let chained = yield yield _ => _(null, $ => $(null, 'in the chain'));
    
  console.log('chained task');
    


  // task delegation, important when we make repetitive tasks
  let delegated = yield* function* takeover(step) {
    yield _ => setTimeout(_,0);
    return 123;
  }(step);
    
  console.log('delegated task', delegated);
  


  // the following lets us listen to the next 3
  // clicks on an object before moving past the loop
  
  // simple way to have the same reference so event listener can be removed
  // since we need to make the first argument to step null
  let delegate = step.bind(null, null);
  document.body.addEventListener('click', delegate);
  
  for (let i = 0; i < 3; i++) {
    let click = yield step;
    console.log(click, 'clicked');
  }
  
  document.body.removeEventListener('click', delegate);  
  


  // example of wait guarding
  setTimeout(function () {
    try {
      step(null);
    }
    catch (e) {
      console.error('tried to step but...', e);
    }
  }, 0);
  // have to wait 100ms for the guard to release
  yield _ => setTimeout(_, 100); 
  


  // return something
  return [1,2,3];
}


// example of super fancy things going on (not shown due to traceur bug)
/*
function concurrent(concurrency, tasks) {
  return function* $concurrent(step) {
    let results = [];
    let queued = 0;
    for (let i = 0; i < tasks.length; i++) {
      if (queued === concurrency) {
        yield step;
      }
      tasks[i]((err, val) => {
        queued -= 1;
        results[i] = val;
        step(err);
      });
      queued += 1;
      i++;
    }
    while (queued) {
      yield step;
    }
    return results;
  }
}
*/

// make it run!
runner(task, (err, ...args) => {
  console.warn('DONE', err, args)
});












//
// -- RUNNER BOILERPLATE --
//
function toCallback(value, options) {
  // console.log('toCallback', value)
  if (value && typeof value.next === 'function' && typeof value.throw === 'function') {
    return function delegate(next) {
      runner(value, next);
    }
  }
  if (typeof value === 'function') {
    return value;
  }
  else if (value && typeof value.then === 'function') {
    return function (next) {
      value.then(
        v=>{next(null, v)},
        e=>{next(e, null)}
      )
    };
  }
  else if (Array.isArray(value)) {
    let steps = value.map(toCallback);
    let results = new Array(steps.length);
    // GC cleanup
    value = null;
    let todo = 0;
    return function (next) {
      function tick(err) {
        if (todo <= 0) return;
        if (err) {
          todo = 0;
          next(err);
          return;
        }
        todo--;
        if (todo == 0) {
          next(null, results);
        }
      }
      for (let step of steps) {
        let done = false;
        let i = todo;
        todo += 1;
        step(function (err, result) {
          if (done) return;
          done = true;
          results[i] = result;
          tick(err);
        })
      }
    }
  }
  else if (value && typeof value === 'object') {
    let steps = [];
    let results = Object.create(null);
    let todo = 0;
    for (let key in value) {
      todo += 1;
      steps.push(function (next) {
        toCallback(value[key])(function (err, value) {
          results[key] = value;
          next(err);
        });
      });
    }
    return function (next) {
      function tick(err) {
        if (todo <= 0) return;
        if (err) {
          todo = 0;
          next(err);
          return;
        }
        todo--;
        if (todo == 0) {
          next(null, results);
        }
      }
      for (let step of steps) {
        step(tick);
      }
    }
  }
  else {
    throw new Error('cannot convert to callback');
  }
}
function runner(generator_fn, cb) {
  let done;
  let first = true;
  let waiting_already = false;
  function step(err = null, ... args) {
    // console.log('step', err, args)
    let result;
    let value;
    if (done) throw new Error('already done');
    if (waiting_already) throw new Error('already waiting');
    try {
      if (err) {
        result = generator.throw(err);
      }
      else {
        // lol es-discuss
        if (first) {
          first = false;
          result = generator.next();
        }
        else result = generator.next(...args);
      }
      value = result.value;
      done = result.done;
    }
    catch (e) {
      done = true;
      cb && cb(e, null);
      return;
    }
    if (done) {
      cb && cb(null, value);
    }
    else if (value == step) {
      // do nothing, we were told that the generator has stuff setup;
    }
    else if (value == null) {
      generator.next(undefined);
    }
    else {
      waiting_already = true;
      let fn;
      try {
        fn = toCallback(value);
      }
      catch (e) {
        done = true;
        cb && cb(e, null);
        return;
      }
      let performed = false;
      fn((...args) => {
        if (performed) throw new Error('already performed this step');
        performed = true;
        waiting_already = false;
        step(...args);
      });
    }
  }
  let generator = generator_fn(step);
  step();
}