Go routines for JS

go.js

// Utility function for detecting generators.
let isGenerator = x => {
  return Function.isGenerator &&
         Function.isGenerator.call(x)
}

// Data type represents channel into which values
// can be `put`, or `received` from. Channel is
// very much like queue where reads and writes are
// synchronized via continuation passing.
function Channel() {
    this.active = false
    this.inbox = []
    this.outbox = []
}

// Drain is operation that is invoked with channel any time
// new value is send to it or request is received.
function drain(channel) {
  let { active, inbox, outbox } = channel
  if (!active) {
    channel.active = true
    while (inbox.length && outbox.length) {
      let message = inbox.shift()
      let deliver = inbox.shift()
      let send = outbox.shift()
      
      let receipt = send(message)
      if (deliver) deliver(receipt)
    }
    channel.active = false
  } 
}

// Queues message into a channel and blocks routine until
// consumer receives it. Note that `put` must be used with
// yield operator in go routines:
//
// let numbers = (start, end) => {
//   let output = new Channel()
//   go(function*() {
//     let x = start
//     while (x < end) {
//       yield put(output, x)
//       x = x + 1
//     }
//   })
//   return output
// }
function put(channel, message) {
  return function(resume) {
    channel.inbox.push(message, resume)
    drain(channel)
  }
}
exports.put = put

// Receives message from the input channel & blocks routine until
// message is received. Note that `receive` must be used with a
// yield operator in a go routines:
//
// go(function*(input) {
//   let event = void(0)
//   while (event = yield receive(input)) {
//     console.log(event.keyCode)
//   }
// }, keypress)
function receive(channel) {
  return function(resume) {
    channel.outbox.push(resume)
    drain(channel)
  }
}
exports.receive = receive

// Queues message into given `channel` but without blocking
// a go routine. Send can be used inside or outside go
// routines.
//
// function keypress(target) {
//   let output = new Channel()
//   target.addEventListener("keydown", function(event) {
//     send(output, event)
//   }, false)
//   return output
// }
function send(channel, message) {
  put(channel, message)()
}
exports.send = send


// Funtion that starts a go routine with a given arguments.
// `routine` must be a ES6 generator, which will be invoked
// with rest of the arguments. Go routines are suspended on
// each `yield` until scheduled task is resumed. For example
// `yeild receive(input)` will block routine until message is
// received on the given `input`, once message is received
// routine is resumed and result of expression is `message`
// recived. Similarily `yield put(channel, message)` in routine
// will block it until `message` is recieved on the other end.
function go(routine, ...args) {
  if (isGenerator(routine)) {
    let task = routine(...args)
    let next = data => {
      let {done, value} = task.next(data)
      return done ? null :
             value ? value(next) :
             next()
    }
    next()
  }
  else {
    throw new TypeError("routine must be a generator")
  }
}
exports.go = go



// Implementation of fadeIn example from:
// Google I/O 2012 - Go Concurrency Patterns 
// http://www.youtube.com/watch?v=f6kdp27TYZs

let forward = function*(input, output) {
  let message = void(0)
  // send messages to an output until received
  // message is falsy. Note `send` is used intentionally
  // instead of `put` since messages need to be forwarded
  // as they arrive not as they are received.
  while (message = yield receive(input))
    send(output, message)
}

let fanIn = (input1, input2) => {
  let output = new Channel()
  
  // Lunch two go routines that just
  // send values to an output.
  go(forward, input1, output)
  go(forward, input2, output)
  
  return output
}
exports.fanIn = fanIn

let print = input => {
  go(function*() {
    while (true)
      console.log(yield receive(input))
  })
}
exports.print = print


// Example
  
let a = new Channel()
let b = new Channel()
let c = fanIn(a, b)

    
send(a, 1)
send(a, 2)
send(b, 10)


print(c)
  
send(a, 3)
send(b, 11)