var archan =var fs =// create a new channelvar ch =
var archan =
var ch = archan([options])
Creates a new channel instance. Options:
concurrency: Infinity- See below for control flow handling
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
var ch =var count = 0while chlengthchcount++assert
var cb = ch.push()
Returns a new callback you pass to asynchronous functions.
fs // push a callbackvar buffer = ch // yield the result of the callback// push a callbackvar result = ch // 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 =var stream = fsvar cb = ch // create a callbackstreamstreamch // wait until the stream is finished
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 // adds 1ch // adds [1, 2, 3]ch // adds an error, will be thrown on next .shift()
var val = yield* ch.shift([alwaysWait])
Returns the next value in the channel.
* is optional.
true and there are no more values,
it will indefinitely wait for the next value just like chan.
undefined will be returned immediately.
Note that if any values were errors,
.shift() will throw that error.
See below for error handling.
Since archan keeps track of your callbacks, it can help you handle control flow much better.
The number of pending callbacks in the channel that have not yet returned.
This is the maximum number of pending callbacks you want to allow at once.
This value only matters when you do
By default, this value is
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 =var saveTo =app
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 chlengthvaluesreturn values
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.
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 valwhile chlengthtryval = chcatch err// do something with the errorerrstatus = 400throw err
or in a single callback:
fstryvar text = chcatch errconsole
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
This alias demonstrates archan's synonymity with Readable Streams in object mode:
var stream =streamstreamstreamstreamassertassertassertassert
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 =var ch =// add some values// has to be added before the pipe begins// otherwise `ch.length === 0`fsfsfsfs// the pipe should be in its own coroutine// since we didn't specify an error handler, it'll throw on any errorsconsole
Suppose you want to indefinitely pipe values from a channel to a Writable Stream.
The MIT License (MIT)
Copyright (c) 2013 Jonathan Ong firstname.lastname@example.org
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.