helpers: introduce Stream()

[Fishrock123]

Refs: #26

This turned out quite reasonable, I think. I might just like it. (It was pretty easy to implement, worked first try??)

[Raynos]

Are you sure you want to implement a thenable ?

What about adding a method that returns a promise like stream.wait()

[Fishrock123]

Ok, a little more complexity and this is now capable of this kind of thing:

  const streamSource = new Stream(fileSource, zlibTransform)
  const streamSink = new Stream(passThrough, fileSink)
  const stream = new Stream(streamSource, new PassThrough(), streamSink)

  await stream.start()

In which Streams are now also a partial passthough for conditional complexity reduction...

I also tried to get rid of bindSource and bindSink but it didn't go too well. Relatedly, I think that bindCb should move to start() (and or start should be thenable or return a promise).

---

@Raynos Is there anything wrong with it being thenable? I guess start could just return a promise, if it's bad...

[Fishrock123]

PR for the start(exitCb) thing: #32

[Raynos]

Is there anything wrong with it being thenable?

I'm not sure if it impacts debugging. For example, a thenable will not play well with unhandledRejections as it's invisible. However your implementation does throw the error if no-one is waiting on it with then()

The other issue with thenable is that someone might refactor

await stream

To

stream.catch(handleError)

And then be confused

[Fishrock123]

I think that start() should maybe just take a callback and if none is provided return a promise. Should solve those concerns and make things simpler and more sensical.

[Raynos]

👍 for adding to start() thanks.

[Fishrock123]

Simplified a lot.

Also stuff like Stream(Stream(Stream(...), ...), ...) now works.

[Raynos]

This looks nice to me.

[Raynos]

consider checking for components.length >= 2 here with some kind of assertion.

[Fishrock123]

It could do more typechecking, but for now that will just make things needlessly more work to read.

Is there an issue with doing Stream(<single component>) though? Maybe there's a use case for that. But also I suppose we could only expand to that later if necessary.

[Fishrock123]

landed as 5a460b4

[Fishrock123]

To be clear, await new Stream(streamSource, new PassThrough(), streamSink).start() doesn't quite work today. I've left the Promise returning out for simplicity for now, but the same is achievable by:

const stream = new Stream(streamSource, new PassThrough(), streamSink)
await util.promisify(stream.start.bind(stream))