Bluebird: Streams and promises

[greghart]

I've a question on best practices when working with Promises and streams, with specific regards in how to "promisify" them. The example problem I'm trying to solve is with an API that returns a file stream. I then want to use Promise.map with concurrency on a large list of files, so that only c files are streaming at any given time. I've solved this to my needs below:

Promise.map(files, function(file){ 
  return new Promise(function(resolve, reject) {
    var stream = API.getStream(file); 
    // Pipe/use stream
    stream.pipe(endPoint);
    stream.on('end', resolve);
    stream.on('error', reject);    
  });
}, {concurrency: 3});

Main question is, is there a more semantic way to accomplish this? I've never before had to write new Promise(...) myself, so I have an instinctual feeling I'm missing something. On the other hand, I'm pretty sure it doesn't make sense to automatically promisify API.getStream.

• The stream is returned synchronously already
• You don't want a promise of a stream, you want a promise of a "finished" stream.
• You want the underlying stream for piping and other stream specific functions
• There's no guarantee that an input is a stream, and not just a general EventEmitter style object.

An example where the API returns a promise of a finished stream.

Promise.map(files, function(file){
  var streamPromise = API.getFinishedStreamAsync(file);
  streamPromise.pipe(streamPromise.stream); // Stream is attached synchronously
  // API now handles stream/promise resolution
  return streamPromise;
}, {concurrency: 3});

[bendrucker]

My time to shine! :smile:

https://www.npmjs.org/package/stream-to-promise

[petkaantonov]

It is generic code that can be tucked away in a utility function so it doesn't clutter application code and obstruct actual application logic.

In your example simplest utility function would be like

function streamToPromise(stream) {
    return new Promise(function(resolve, reject) {
        stream.on("end", resolve);
        stream.on("error", reject);
    });
}

And then the application code is simply

Promise.map(files, function(file) {
    var stream = API.getStream(file);
    stream.pipe(endPoint);
    return streamToPromise(stream);
});

[gurdotan]

@bendrucker nice module!

[jghaines]

Also:
https://github.com/spion/promise-streams

[beck]

A small addition to @petkaantonov's solution, suggest adding stream.resume();, otherwise there is no guarantee that the stream will drain and emit end.

Example:

const gs = require('glob-stream');
const through = require('through2').obj;

function streamToPromise(stream) {
  return new Promise(function(resolve, reject) {
    stream.on('end', resolve);
    stream.on('error', reject);
  });
}

function log(data, enc, cb) {
  console.log(data.path);
  cb();
}

const stream = gs.create('./*.txt')
  .pipe(through(log));

streamToPromise(stream)
  .then(() => { console.log('I never print, womp womp'); });

but the print will happen if one uses:

function streamToPromise(stream) {
  return new Promise(function(resolve, reject) {
    stream.on('end', resolve);
    stream.on('error', reject);
    stream.resume();
  });
}

[manast]

isn't there a risk that these event handlers:
return new Promise(function(resolve, reject) { stream.on('end', resolve); stream.on('error', reject);
leak memory?

[wmertens]

@manast EventEmitters (like Streams) should remove their listeners when they complete. That allows GCing them.