async-stream

The async stream spec.

This is a spec / manual and that's all you need for async streams, you don't need any modules to create a valid stream, not even for convenience.

Semantics

An async stream is simply a promise returning function. If you're familiar with other stream semantics: There's only readable streams, no need for writables and transforms.

• A readable is a function: var read = readable().
• A transform is a readable that takes a stream as argument: var read = transform(readable()).
• A writable is a while loop: var data; while (data = yield read()) {}.

It's all pulling and there's simply no need for base classes.

Comparison to node core streams

Reading

// corea.on('data',console.log)// async streamletdatawhile(data=awaita())console.log(data)

Piping

// coresource.pipe(transform).pipe(destionation)// async streamconstread=transform(source())letdatawhile(data=awaitread()){}

Error handling

// coresource.on('error',handle).pipe(transform).on('error',handle).pipe(destination)// async streamconstread=transform(source())try{letdatawhile(data=awaitread()){}}catch(err){handle(err)}

Implementing a source

// coreconst{ Readable }=require('stream')classreadableextendsReadable{_read(){this.push(String(Date.now()))}}// async streamconstreadable=()=>async()=>String(Date.now())

Implementing a transform

// coreconst{ Transform }=require('stream')classtransformextendsTransform{_transform(chunk,enc,done){done(null,Number(chunk).toString(16))}}// async streamconsttransform=read=>async()=>Number(awaitread()).toString(16)

A note on iteration

The cumbersome

letdatawhile(data=awaitread()){}

will become more elegant once the async iteration spec lands in JavaScript engines:

forawait(constdataofread()){}

Examples

All following examples are wrapped inside:

(async()=>{constsleep=dt=>newPromise(resolve=>setTimeout(resolve,dt))// ...})()

You can run the examples with node >= 7.6:

$ node examples/<name>.js

read

On each invocation the stream should return a String or Buffer of data, or a falsy value when there's nothing more to be read and the stream is done:

// readable stream that emits 3x the current date with 1 second delayconstdates=()=>{leti=0returnasync()=>{if(++i==3)return// endawaitsleep(1000)returnString(Date.now())}}letdataconstread=dates()while(data=awaitread()){console.log('data: %s',data)}console.log('done reading')

Outputs:

$ node examples/read.js
data: 1391519193735
data: 1391519194644
data: 1391519194663
done reading

end

The async function may take an end argument, which when truthy tells the stream to clean up its underlying resources, like tcp connections or file descriptors.

// readable stream with cleanup logicconstdates=()=>{leti=0constcleanup=()=>console.log('cleaning up')returnasyncend=>{if(end||++i==3)returncleanup()awaitsleep(1000)returnString(Date.now())}}letdataconstread=dates()console.log(`data: ${awaitread()}`)console.log(`data: ${awaitread()}`)awaitread(true)console.log('done reading')

Outputs:

$ node examples/read-end.js
data: 1391519193735
data: 1391519194644
cleaning up
done reading

pipe

"Pipe" streams into each other by letting them read from each other. Here we add a hex stream that converts date strings from decimal to hexadecimal:

constdates=()=>{leti=0returnasyncend=>{if(end||++i==3)returnawaitsleep(1000)returnString(Date.now())}}consthex=fn=>asyncend=>{conststr=awaitfn(end)if(!str)returnreturnNumber(str).toString(16)}letdataconstread=hex(dates())while(data=awaitread()){console.log(`data: ${data}`)}console.log('done reading')

Outputs:

$ node examples/pipe.js
data: 143fd169b4d
data: 143fd169f50
done reading

errors

Just throw inside the async function:

consterrors=()=>asyncend=>{thrownewError('not implemented')}letdataconstread=errors()while(true){try{data=awaitread()}catch(err){console.error('threw')break}}

Outputs:

$ node examples/error.js
threw

This means that a transform stream can decide itself if it wants to handle errors from it's source, by wrapping calls to read in a try/catch, or propagate them to the parent.

high water mark / buffering

Some streams - like node's or unix pipes - have the concept of high water marks / buffering, which means that a fast readable will be asked for data even if a slow writable isn't done consuming yet. This has the advantage of being potentially faster and evening out spikes in the streams' throughputs. However, it leads to more memory usage (in node max. 16kb per stream), complicates implementations and can be very unintuitive.

An example where you wouldn't expect that behavior is this:

http.createServer(function(req,res){fs.createReadStream('/dev/random').pipe(res);});

You'd think this would stop reading from the pseudo number generator /dev/random when the request ends, right? Unfortunately that's not the case. Node will read 16kb into an internal buffer first because you might want to later pipe that read stream into another stream and it can than immediately flush that out.

In that case the buffer will be filled up pretty quickly so that's not a huge problem. But imagine your source being slow, with low throughput. For example it could tail logs of an infrequently used system. In this case, with many requests to this http handler, it will keep a great number of streams open.

Currently async-streams have no concept of high water mark / buffering.

License

MIT