[stream] - callback based API additions

[spion]

Streams don't have an API that adheres to the node callback convention. This makes them awkward to use with other control flow tools. The following two methods would be the most useful:

readAll([encoding], function(err, data) {}) - on readable streams it collects all data, or is called with the first error
ended(function(err) {}) - Like the previous one, except it doesn't collect any data - its just called when everything ends (either successfully or at the first error). finished could be used for writable streams.

Both of the above would act as if an .on('error', cb) was attached.

The following would also be useful:
readNext(function(err, data) { ... }) Read the next packet or object. The callback is called with the current packet on the "next tick", with the "current packet" changing for subsequent requests right before the callback is called.

Would act as if a .once('error', cb) was attached

And the last method would also be useful, but I'm unsure of the semantics:

writeNext(data, [enc], function(err) { ...}) - The callback would be called when the source is allowed to continue writing. Now I'm not entirely sure what this would mean, but I suppose that for streams with buffers, this would mean "called immediately" if the buffer is not full and called after flush if it is, and for streams without buffers this would be called after flush.

This should be taken as a rough sketch, as I'm not sure I understand enough about node streams internals to give a good proposal :)

edit: sorry for the close/open noise, I accidentally the button.

[phpnode]

@spion doesn't your first example negate many of the benefits of using streams? i.e. if you have to buffer the whole response, what is gained by using streams at all? Also, these things could easily be userland functions, so, why not create a module and see how it works?

[spion]

Answering point 1, I don't think so. The reality is that there are valid use cases with relatively small data where the costs (of complexity and performance due to incremental processing) far outweight the benefits. Just look at all the http request modules - the most popular ones always provide ways to collect all the data.

Regarding point 2, I'm experimenting with a module that does some of this (and more) with promises. However I still believe that core should at least be self-consistent.

edit: I'll try and make a module that implements the methods and test out some of the use cases I have in mind.

[phpnode]

@spion cool, looking forward to seeing your module. I agree about the lack of consistency but I think node should be looking to (eventually) move to the whatwg streams spec which has pretty good promise integration baked in.

[spion]

@phpnode I agree. I'm just suggesting a relatively cheap (in terms of complexity / backward compatibility) ergonomic improvement that could be done in the meantime. (I want to be able to tell bluebird users to promisifyAll their stream prototypes, haha)

[benjamingr]

+1 this is an extremely common case that should be solved at the core level.

Most other platforms that provide streams have this method and it is immensely useful for a lot of use cases.

[zensh]

thunk-stream wrap stream to thunk function:

var thunkStream = require('thunk-stream'),
  stream = require('stream'),
  fs = require('fs');

var readableStream = fs.createReadStream('index.js');
var passStream = new stream.PassThrough();

thunkStream(readableStream)(function (error) {
  if (error) console.error('error', error);
  else console.log('read file end.')
});

thunkStream(passStream)(function (error) {
  console.log('file pass through finished.')
});

readableStream.pipe(passStream);

[chrisdickinson]

Streams don't have an API that adheres to the node callback convention.

Streams use callbacks where the operation in question is representable by a callback. A callback can only represent a single operation that eventually passes or fails. For instance, writable.write(data, enc, callback) uses a callback to represent the operation of having flushed data to the underlying resource, which may only happen once per .write, pass or fail.

As a counterexample, .read(n) does not represent a single read-from-resource operation, due to the highwatermark buffering nature of streams2+ -- a single .read may correspond to zero or more "chunks" of data being passed through the stream. There's no way for consumers of streams to know when the _read resulting from read has stopped .push'ing data.

readAll([encoding], function(err, data) {}) - on readable streams it collects all data, or is called with the first error

Because this can be done in userland with concat-stream; and there's currently no need to collect and buffer an entire stream internally in node, this is not a viable addition to the streams API. Additions to this API should either provide functionality unattainable through userland, or that is directly needed by other core modules.

ended(cb) is a more viable candidate, since it represents a result that may only happen once, pass or fail. However, it conflates two events that are currently emitted by readable streams. Since Streams are (for better or worse) implemented as event emitters, and their .pipe mechanism relies on those two events separately, it is unlikely that an alternate API to get to that state will be added.

Re: readNext and writeNext -- The above explanation of why .read lacks a callback should apply to readNext as well, but I confess I'm not sure what use case these methods serve.

I agree about the lack of consistency but I think node should be looking to (eventually) move to the whatwg streams spec which has pretty good promise integration baked in.

I'm definitely interested in that approach, but unfortunately the whatwg spec is built around promises-as-a-core-primitive, so whatever (non-trivial) amount of work necessary to properly support promises as the primitive representing single-operation-pass-or-fail in node core would have to happen first. i.e., adopting the streams spec (or even cribbing from it) would be great, but it is a very long way off, IMO.

[spion]

Because this can be done in userland with concat-stream; and there's currently no need to collect and buffer an entire stream internally in node, this is not a viable addition to the streams API. Additions to this API should either provide functionality unattainable through userland, or that is directly needed by other core modules.

I disagree. As it is, the current node API is in a no-person's land. Its neither

• completely low-level / unopinionated, or
• ergnomic, well crafted and self-consistent.

Since the API can't be made more low-level and less complex without breaking backward compatibility, features that don't cost much yet add immense benefit for consumers should be considered even if they don't provide any benefit to core. Streams are already a "batteries included" feature - so why not provide decent batteries while at it? Otherwise I don't see the point.

That raises another interesting question, what is the actual strategy regarding the current core library? Will it be deprecated in favor of libuv.js or will it be developed further?

Re: readNext and writeNext -- The above explanation of why .read lacks a callback should apply to readNext as well, but I confess I'm not sure what use case these methods serve.

readNext and writeNext aren't really useful for current use cases of streams, but I was thinking they could probably be useful to make streams usable as channels for CSP. On second thought, they're not really designed to do that so its probably a bad idea.

ended(cb) is a more viable candidate, since it represents a result that may only happen once, pass or fail. However, it conflates two events that are currently emitted by readable streams. Since Streams are (for better or worse) implemented as event emitters, and their .pipe mechanism relies on those two events separately, it is unlikely that an alternate API to get to that state will be added.

You could also argue that fs.readFile(..., (err, data) => ...) starts a reading operation and conflates two separate events, error and data arrival. That doesn't make the API format of readFile less useful or practical.

I don't understand how this would interfere with pipe. Afaik, anyone can query the stream for the error or end event. Is that wrong? If its not IMO this is the single biggest improvement that the streams API could get, at least for end-users. I'd like to be able to do this:

var s1 = stream1(opts);
var s2 = stream2(opts);
var s3 = stream3(opts);
s1.pipe(s2).pipe(s3);
async.parallel(s1.ended.bind(s1), s2.ended.bind(s2), s3.ended.bind(s3), done);

[sonewman]

So do you mean from your example that each stream would need a define an ended method. Because having multiple things happening on ended would then become very manual, unless we still have the EventListener pattern.

Also this would work very well for people using the async library! But what if you don't want to use that library? Would this not have sufficed:

var s1 = stream1(opts);
var s2 = stream2(opts);
var s3 = stream3(opts);
s1.pipe(s2).pipe(s3);
s3.on('finish', done);

Unless we inherently needed to inform the next stream that the previous one had ended, (which it already knows about internally anyway).

It seems like we are missing the huge benefits of our stream abstractions.
OK so for a simple http-server responding with a blob of html, it's not so important and we probably want to provide a simpler solution for that use case, since it is much more common, but this has little to do with streams.

To reiterate, the point with streams is that the whole data is not dealt with in one chunk otherwise one would just use or provide an async callback API.

A stream allows a way of handling a series of async events returning data. As described by many before me an array through time. It doesn't really matter how big or small those chunks are, in some cases that stream my never end, because it is a socket and it is open forever.

Similarly doing readable.on('data', data => { }) is a lot like doing [1, 2, 3, 4].forEach(no => { }) We say the forEach takes a callback but it doesn't just happen once. It's like saying we expect forEach to be called only once, containing all the values in an array (isn't that what we started with?).

OK so in other languages we can do the equivalent of readAll but we also have threads, coroutines, forks, etc. We can afford to do infinite loops until the stream ends or errors, because although it appears linear when written, behind the scenes threads and schedulers are parallelising all of this for us.

Don't get me wrong having a way to do light-weight threading (in our case isolated super light-weight JS runtime) would be be awesome, however the single-threaded runtime is what make currently makes iojs / Node!

This kind of defeats the whole flexibility of JavaScript's asynchronous model.

There are also multiple ways to buffer a whole stream, some of which are async and others which would block the rest of the runtime.

IMO it's like we are taking two steps backwards from everything that iojs / Node has given us with it's evented (process.nextTick/setImmediate throw this onto the 'list') model.

What we need is to agree on an abstraction that handles asynchronous and it's internals that deals with this.

with regards to:

I disagree. As it is, the current node API is in a no-person's land. Its neither

completely low-level / unopinionated, or
ergnomic, well crafted and self-consistent.

It may not be as simple as a callback but it is pretty low-level handling piece of data in such granular pieces, that is the point.
It is not most peoples use case, because their use case is a much higher level.

ergonomically perhaps it is a little weird since we have to inherit from a stream and override implementation detail which is then in turn called by internals, which is not a pattern which we would all be used to.

No one is saying that the streams API is perfect but I strongly disagree with it not being well crafted. @isaacs et al. put a great deal of time and work into forming what we currently know of as streams and what we have now is a pattern that makes all streams following that pattern automatically compatible with one another, which IMO is awesome!

[chrisdickinson]

Streams are already a "batteries included" feature - so why not provide decent batteries while at it?

Streams are not a "batteries included" feature. They are an abstraction core uses internally to represent the /data*((end|error)?close)?/ abstraction -- which will always be required by core. The current implementation has features that appear to be "batteries included" at first glance: for instance, highwatermark buffering. However, their raison d'être is not to provide a "batteries included" API, but because other parts of the core API required that implementation for performance considerations.

That raises another interesting question, what is the actual strategy regarding the current core library?

libuv.js would become a dependency of node/iojs, one which the current core library is built on top of. Alternate implementations of core functionality could be built in pure JS.

However, since there's a massive package ecosystem built around the current set of abstractions, there's going to be a strong incentive to stay compatible with what exists at present. In other words, libuv.js won't solve the problem of "how to evolve core" so much as it will make maintaining core more straightforward, and open up the possibility of building different abstractions on top of libuv to more people. Those abstractions are not guaranteed to be "npm-ecosystem-compatible."

You could also argue that fs.readFile(..., (err, data) => ...) starts a reading operation and conflates two separate events, error and data arrival. That doesn't make the API format of readFile less useful or practical.

An errored stream is not the same as an errored callback operation. A stream may have a backing resource which may be inspected on error -- for example, to inspect the buffered items, or to look at a TCP socket's remote address -- before disposal. A callback operation can only represent the final state of the entire transaction.

For example:

     fs.createReadStream
                           finish/
     flow    data   data  end/error    close
      |       |      |       |          |
      V       V      V       V          V
 time ------------------------------------->
      ^                                 ^
      |                                 |
     fs.readFile                      (err, data)

* "finish" is the writable equivalent to the readable "end" event.

Callbacks can't represent the temporal difference between "end/error" and "close". .pipe's machinery relies on being able to differentiate between close, end, error, and finish. This is not to say that callbacks are not useful abstractions, but instead that they are not, by themselves, sufficient for all purposes (NB: s/callbacks/promises/g applies equally well here).

I don't understand how this would interfere with pipe.

It's not that .ended(cb) would interfere with .pipe, it's that it could not replace the operations that .pipe currently relies upon (notably, being able to differentiate close/error/end temporally) -- it could only be implemented as sugar on top of what's there.

One other consideration to keep in mind is that it's not trivial to expand the API surface area of streams reliably: for instance, imagine getting a stream from a package that has a pegged version of readable-stream. The .ended method will not be available on it. There's not a single streams interface -- there are four in core ([v0.10, v0.12].length * 2), and 40 (!) releases of readable-stream. Maintaining compatibility and consistency between these implementations is a delicate process.

Closing this issue for now:

1. readAll is solved by userland's concat-stream package, and no core module currently requires that functionality.
2. ended(cb) is API expansion that would be difficult to ensure across versions of readable-stream and builtin streams, and could not replace existing mechanisms used by .pipe.

If you'd like to discuss further, feel free to comment on this issue or find me in IRC on freenode (in #io.js) — I'm always happy to talk streams!

[sonewman]

@chrisdickinson That ascii diagram is a perfect explanation, and should be in the documentation somewhere, if it is not already!