Implement asNodeWritable.

[vqvu]

Implements asNodeWritable as discussed in #358.

This PR also adds a callback argument to write that gets called when the value written is sent downstream.

Advice needed
Currently, write has the signature write(token, encoding, callback) to match the Writable#write signature. Since we don't care about the encoding, and since Highland streams are not Writable anyway, should we even have the encoding argument? We have asNodeWritable, so we should be deprecating the use of writable Highland streams as a Writable anyway.

[apaleslimghost]

This won't be compatible with pre-0.10 Nodes, do we support those?

[svozza]

We're not running any of our tests on Node 0.8 so I guess by virtue of that we don't.

[vqvu]

Once upon a time we did (back in 1.x according to git history). I think we dropped it because one of our devDependencies is no longer compatible with the version of npm that comes with 0.8 on travis.

If you're concerned we could make asNodeWritable return this if require('stream').Writable doesn't exist. Thoughts?

[apaleslimghost]

Well, there's always readable-stream, but I'm not too concerned. I think we're fine ignoring old Nodes, as long as we make it clear what we support.

[vqvu]

Ok. We'll keep using Writable then.

[vqvu]

Any thoughts on the encoding argument of write? My feeling is we don't need to have it, since Highland streams are EventEmitters and not Writables. It just so happens that piping to highland streams works.

The only reason to have it is for slightly more compatibility with Writable than we have now. But not having it might make it clear to people that Highland streams should not be used as a Writable in general.

[shaunc]

My case was passing a stream to a third-party library which passed in a callback in the third argument. The callback called "result()" of a promise wrapper, so needed to get called. For me it would be great if, whether or not you accepted a callback as the 2nd argument, if the 2nd argument wasn't a function you'd treat the 3rd argument as a callback (if present).

(Also might not you sometimes wrap a node stream which could make use of the encoding? Doesn't matter for me but just a thought.)

Thanks for the quick work here! (PS end() with a callback in third argument possible would also be great :))

[shaunc]

(suggested edit -- see below): If it is callable, it is taken as the callback argument.

[vqvu]

For me it would be great if, whether or not you accepted a callback as the 2nd argument, if the 2nd argument wasn't a function you'd treat the 3rd argument as a callback (if present).

My point is that you can always use asNodeWritable to get a working writable, so there's no need to support the Writable#write signature at all. The reason to have the encoding argument is for compatibility with Writable, but users should really be using asNodeWritable if they need a Writable anyway; you never know when the library call will be changed to use a feature of Writable that Highland streams don't support. Allowing an optional encoding argument in the way you suggest would muddy the message.

Also might not you sometimes wrap a node stream which could make use of the encoding?

We don't wrap node streams. We consume from them. Even the _(nodeStream) constructor actually calls pipe to get the data, so there's no way for us to make use of the encoding. Highland streams are more like objectMode node streams anyway, so the encoding is pointless. We don't treat strings or buffers in any special way.

PS end() with a callback in third argument possible would also be great :)

You should be able to get this behavior from asNodeWritable. Just be aware that the callback may never be called. This is a result of Highland's laziness; it won't fetch data until it absolutely needs it. The source stream may never be consumed to the end. For instance, in this situation

var s = _();
var writable = s.asNodeWritable();
writable.write(1);
writable.end(2, null, function () {
    console.log('finished');
});

s.take(1).each(_.log);
// => 1
// Won't see the finished.

[shaunc]

Ok -- thats fine. (One might say that having "write" at all already muddies the message. :))

The laziness you describe seems ok in your example, but "end" is called explicitly on "s" shouldn't "writable" listen for that and call the callback then, as the data has been explicitly discarded?

...
s.end()
// => finished

UPDATE: Probably the answer is "no", as the data is still available. End just means "no more data". I guess there is no "shutdown" interface expect to read all the data?
UPDATE 2 (I should read your documentation more thoroughly. :)) -- I guess there is a shutdown interface: "destroy". So my question should be phrased:

...
s.destroy()
// => finished

[vqvu]

This brings up an interesting question. What should happen here?

var s = _();
var writable = s.asNodeWritable();
writable.write(1, null, callback);
writable.write(2, null, callback);
writable.write(3, null, callback);
s.destroy();

Should those callbacks be called? I see three choices

1. Call the callbacks, because a destroy "handles" those values by throwing them away.
2. Don't call the callbacks, because the values were thrown away, and that doesn't count as being "handled".
3. Call the callbacks with an error that says "Stream was destroyed". This will cause the Writable to emit('error'), which doesn't seem like what you would want.

[vqvu]

I guess there is no "shutdown" interface expect to read all the data?

By the way, destroy is not expected to read all the data and throw it away. It's simply a "I don't care about the stream any more" call.

[shaunc]

Ok ... I wouldn't expect the data would be read. But I would expect a stream whose sink has hung up on it would say something --either "finish" event with an error (though the node documentation doesn't say anything about this) or an "error" event. After all, if your task is to send a message, you can wait patiently (maybe forever unless you timeout) until its read but might want to take some action if its guaranteed not to be read.

I would think that destroy should emit some sort of event, (though it seems from the code that it doesn't), which sources could listen for. In particular, I guess here writable could use it to generate an error event.

[vqvu]

I would think that destroy should emit some sort of event

The event is exposed in v3.0 (in development) as onDestroy.

But I would expect a stream whose sink has hung up on it would say something

In general, I agree. Here's my concern. In 3.0, consumers will automatically destroy the parent once it's done. This allows the source to clean up any resources that it is holding. Consumers may hang up on their source for reasons other than an error, and the source should respond by simply stopping to generate data. There is no equivalent concept for Writables; the only way to signal that a Writable won't be accepting more data is with an error.

For example, consider this

var s = _();
libraryCall(s.asNodeWritable());
s.take(1).each(_.log);

function libraryCall(writable) {
    writable.write(1);
    writable.end(2);
}

In the above example, stream will be destroyed after it emits its first value. If it is set up to cause an emit('error'), you would end up with spurious errors thrown all over the place.

So...

1. We can have the stream end the Writable and fire all of the pending callbacks on destroy, but that would cause the Writable to be ended out-of-band of the library call. If the library is not expecting this and calls write, you'd get an exception.
2. We can have the Writable throw away all values written after the stream is destroyed. It will simply fire the callback immediately. This can potentially cause the library call to do a lot of unnecessary work and breaks laziness.
3. We can have the stream do nothing to the Writable when it is destroyed. This will look to the Writable like its writes are pending forever. The user of the library must then listen to onDestroy and take the appropriate cleanup actions (if needed) when it is called.

Now that I think about it, the approach of least surprise is probably (2) even though it doesn't preserve laziness. In the end, we're exposing a Highland stream via an different API that doesn't support the same semantics, so something has to give.