[ Javascript / ES6 / Generators / Promises ]: Create a method to implement async flow control with ES6 generators

generator-flow-control.js

import EventEmitter from 'events'

class GeneratorEngine {
  constructor({ controller }) { this.controller = controller }

  iterate(progress, step) {
    if (step.done) { return step.value }

    this.route(step.value, result => {
      this.iterate(progress, progress.next(result))
    })
  }

  start(generator) {
    let progress = generator()
    let step = progress.next()

    this.iterate(progress, step)
  }

  route({ operation, inputs }, callback) {
    if (Reflect.has(this.controller, operation)) {
      let method = Reflect.get(this.controller, operation)
      return Reflect.apply(method, this.controller, [inputs, callback])
    }

    return new Promise(resolve => resolve(callback()))
  }
}

class Controller extends EventEmitter {
  constructor() {
    super()
    this.engine = new GeneratorEngine({ controller: this })
  }

  delay(inputs, callback) {
    let [ms] = inputs
    setTimeout(callback, ms)
  }

  fork(inputs, callback) {
    let [routine] = inputs
    this.engine.start(routine)
    callback()
  }

  subscribe(inputs, callback) {
    let [channel] = inputs
    let resolver = resolve => {
      this.once(channel, data => resolve(callback(data)))
    }

    return new Promise(resolver)
  }

  publish(inputs, callback) {
    let [channel, data] = inputs
    let resolver = resolve => resolve(callback(this.emit(channel, data)))
    return new Promise(resolver)
  }

  start(generator) { this.engine.start(generator) }
}

const delay = (ms) => ({ operation: `delay`, inputs: [ms] })
const fork = (routine) => ({ operation: `fork`, inputs: [routine] })
const publish = (...inputs) => ({ operation: `publish`, inputs })
const subscribe = (channel) => ({ operation: `subscribe`, inputs: [channel] })
const controller = new Controller

function* emitter() {
  while(true) {
    yield delay(3000)
    yield publish(`send:message`, "A delayed message")
  }
}

function* listener() {
  while(true) {
    let result = yield subscribe('send:message')
    console.log(result)
  }
}

controller.start(function* main() {
  yield fork(listener)
  yield fork(emitter)
})