archan
v0.4.1
Published
Array-like generator-based channels
Downloads
56
Keywords
Readme
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 [email protected]
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.