archan

Archan

Array-like channels for co. Similar to chan except it keeps track of the number of pending callbacks and thus allows a coroutine to exit.

Example

var archan = require('archan')
var fs = require('fs')

// create a new channel
var ch = archan()

co(function* () {
  fs.readFile('file1.txt', ch.push())
  fs.readFile('file2.txt', ch.push())
  fs.readFile('file3.txt', ch.push())

  var text
  while (text = yield* ch.shift()) {
    // contents of one of the three text files
    console.log(text.toString('utf8'))
  }

  console.log('all done!')
  process.exit()
})()

API

var archan = require('archan')

var ch = archan([options])

Creates a new channel instance. Options:

• concurrency: Infinity - See below for control flow handling

Array-like Properties

ch.length

The number of values in the channel. Note that if you ever push a "falsey" value in any of your callbacks, the while loop in the example will not work as expected. Instead, you should check ch.length:

var ch = archan()

setTimeout(ch.push(), 1)
setTimeout(ch.push(), 10)
setTimeout(ch.push(), 100)
setTimeout(ch.push(), 1000)

var count = 0
while (ch.length) {
  yield* ch.shift()
  count++
}

assert.equal(count, 4)

var cb = ch.push()

Returns a new callback you pass to asynchronous functions.

fs.readFile('file.txt', ch.push()) // push a callback
var buffer = yield ch.shift() // yield the result of the callback
request('https://github.com', ch.push()) // push a callback
var result = yield ch.shift() // yield the result of the callback

Note that each callback returned from ch.push() is single use only. If you use a callback more or less than once, things are going to go badly with your channel. For example, for emitters, you would want to do something like this:

var ch = archan()

var stream = fs.createWriteStream('file.txt')
var cb = ch.push() // create a callback
stream.once('error', cb)
stream.once('finish', cb)
yield* ch.shift() // wait until the stream is finished

ch.push(val...)

Push a value to the channel synchronously. If you push any values, a callback will not be returned. Multiple arguments will be combined into a single array.

ch.push(1) // adds 1
ch.push(1, 2, 3) // adds [1, 2, 3]
ch.push(new Error()) // adds an error, will be thrown on next .shift()

var val = yield* ch.shift([alwaysWait])

Returns the next value in the channel. The * is optional. If alwaysWait is true and there are no more values, it will indefinitely wait for the next value just like chan. Otherwise, undefined will be returned immediately.

Note that if any values were errors, .shift() will throw that error. See below for error handling.

Control Flow

Since archan keeps track of your callbacks, it can help you handle control flow much better.

ch.pending

The number of pending callbacks in the channel that have not yet returned.

ch.concurrency

This is the maximum number of pending callbacks you want to allow at once. This value only matters when you do yield* ch.drain(). By default, this value is Infinity.

yield* ch.drain()

If there are too many pending callbacks, this yields until the next drain event. Otherwise, it returns immediately.

For example, when saving files from a multipart upload to your server, you'd want to limit the number of open file descriptors per request. In this example, we'll limit the number to 5:

var parse = require('co-busboy')
var saveTo = require('save-to')

app.use(function* () {
  var ch = archan({
    concurrency: 5 // 5 maximum file descriptors
  })

  var parts = parse(this, {
    autoFields: true
  })
  var part

  while (part = yield parts) {
    yield* ch.drain()
    // save the stream to a file
    saveTo(part, 'randomFolder/' + part.filename, ch.push())
  }

  // wait until all the files are saved
  // and all the file descriptors are closed
  yield* ch.flush()

  // tell the client that everything went okay
  this.status = 204
})

var values = yield* ch.flush([returnValues])

Return all the pending values at once and clear the channel. This is a shortcut for:

var values = []
while (ch.length) {
  values.push(yield* ch.shift())
}
return values

If returnValues is false, it won't bother collecting the return values and returning it to you, which is most likely the case if you're just using archan for control flow.

Error Handling

ch.shift() will throw any errors that occured in any of the callbacks or that were pushed to the channel. These errors will be pushed to the beginning of the channel before any other value. Thus, if you want to handle errors on a per-shift basis, you should do the following:

var val
while (ch.length) {
  try {
    val = yield* ch.shift()
  } catch (err) {
    // do something with the error
    err.status = 400
    throw err
  }
}

or in a single callback:

fs.readFile('file.txt', ch.push())
try {
  var text = yield* ch.shift()
} catch (err) {
  console.log('this file probably doesn\'t exist')
}

Streams

Channels are pretty similar to readable object streams except:

• Order is not preserved
• Values are executed in parallel instead of in series

var val = yield* ch.read()

An alias for ch.shift(). This alias demonstrates archan's synonymity with Readable Streams in object mode:

var stream = archan()
stream.push('first')
stream.push('second')
stream.push('third')
stream.push(null)

assert.equal('first', yield stream.read())
assert.equal('second', yield stream.read())
assert.equal('third', yield stream.read())
assert.equal(null, yield stream.read())

Note that when passing callbacks, order is not preserved. In this case, order is preserved because values were pushed synchronously.

Piping to a Writable Stream

It's pretty easy to pipe to writable stream, even with back pressure. However, pipes won't be included in this repo since there are too many different ways you could want to do this. Here's a channel that pipes to a stream until it flushes with backpressure:

var stream = new Stream.Writable()
var ch = archan()

// add some values
// has to be added before the pipe begins
// otherwise `ch.length === 0`
fs.readFile('file1.txt', ch.push())
fs.readFile('file2.txt', ch.push())
fs.readFile('file3.txt', ch.push())
fs.readFile('file4.txt', ch.push())

// the pipe should be in its own coroutine
co(function* (){
  while (ch.length) {
    if (!stream.write(yield* ch.shift())) {
      // needs a drain event, so wait until the next one
      yield stream.once.bind(stream, 'drain')
    }
  }

  // end the stream
  stream.end()
})() // since we didn't specify an error handler, it'll throw on any errors

console.log('lets do something else in the mean time')

Suppose you want to indefinitely pipe values from a channel to a Writable Stream.

co(function* () {
  while (true) {
    if (!stream.write(yield* ch.shift(true))) {
      yield stream.once.bind(stream, 'drain')
    }
  }
})

License

The MIT License (MIT)

Copyright (c) 2013 Jonathan Ong [e-mail]

Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.