promise-streams
    TypeScript icon, indicating that this package has built-in type declarations

    2.1.1 • Public • Published

    promise-streams

    A collection of streams that work well with promises (through, map, reduce)

    example

    Using ES6 arrow functions:

    var Promise = require('bluebird'),
        request = require('request'),
        path = require('path'),
        fs = require('fs'),
        ps = require('promise-streams'),
        select = require('./select-elements');
     
    Promise.promisifyAll(request);
     
    var download = url =>
      ps.wait(request('http:' + url).pipe(
        fs.createWriteStream(
            'images/' + path.basename(url))));
     
    var downloadAllFrom = url =>
        request(url)
        .pipe(select('.post a img', el => el.attributes.SRC))
        .pipe(ps.filter(url => /jpg$/.test(url.toLowerCase()))
        .pipe(ps.map({concurrent: 4}, imgurl => download(imgurl, url)))
        .reduce((count, stream) => count + 1, 0);
     
    downloadAllFrom('http://imgur.com/').done(
        total => console.log(total, "images downloaded"),
        err   => console.error(err.stack))

    api

    ps.through

    ([opts:Options,] fn:(data[, enc]) => Promise)) => PromiseStream

    Create a through-promise stream. Pass it a function that takes data and encoding and uses this.push to push values or promises. This function should return a promise that indicates when the object/chunk are fully processed.

    Returns a PromiseStream.

    Options:

    • concurrent - The maximum number of concurrent promises that are allowed. When this limit is reached, the stream will stop processing data and will start buffering incoming objects. Defaults to 1

    • highWatermark - the size (in objects) of the buffer mentioned above. When this buffer fills up, the backpressure mechanism will activate. Its passed to node's transform stream.

    The other options are also passed to node's Transform stream constructor.

    ps.map

    ([opts:Options,] fn: (data[, enc]) => Promise) => MapPromiseStream

    Create a new MapPromiseStream. The function should return a promise for the next object that will be pushed to the stream.

    Options: Same as ps.through

    ps.filter

    ([opts:Options,] fn: (data[, enc]) => boolean) => FilterPromiseStream

    Create a new FilterPromiseStream. The function should return a boolean to indicate whether the data value should pass to the next stream

    Options: Same as ps.through

    ps.reduce

    ([opts:Options,] fn: (acc, data[, enc]) => Promise) => ReducePromiseStream

    Reduces the objects in this promise stream. The function takes the resolved current accumulator and data object and should return the next accumulator or a promise for the next accumulator.

    The ReducePromiseStream has a promise() method which returns the final accumulator value

    process.stdin.pipe(split()).pipe(es.reduce(function(acc, el) {
        return acc + el;
    })).promise().then(function(sum) {
     
    });

    ps.wait

    (s: Stream) => Promise

    Wait for the stream to end. Also captures errors.

    ps.collect

    (s: Stream) => Promise<Buffer | string>

    Collects data from a stream into a single buffer or string, depending on the encoding of the passed stream.

    ps.pipe

    (source: Stream, destination: Stream) => Promise

    Pipes s1 to s2 and forwards all errors to the resulting promise. The promise is fulfilled without a value when the source stream ends.

    ps.pipeline

    (source: Stream, ...streams: Stream[]) => Promise

    Like pipe, but creates a pipeline of multiple streams.

    PromiseStream.prototype.push

    Like this.push in through2, but takes promise arguments. It returns a promise that resolves when the pushed promise resolves, to make it possible to use return this.push(data)

    PromiseStream.prototype.map

    ([opts:Options,] fn: (data[, enc]) => Promise) => MapPromiseStream

    Create a new MapPromiseStream and pipes this promise stream to it.

    PromiseStream.prototype.filter

    ([opts:Options,] fn: (data[, enc]) => boolean) => FilterPromiseStream

    Create a new FilterPromiseStream and pipes this promise stream to it.

    PromiseStream.prototype.reduce

    ([opts:Options,] fn: (acc, data[, enc]) => Promise) => Promise

    Reduces the objects in this promise stream. The function takes the resolved current accumulator and data object and should return the next accumulator or a promise for the next accumulator.

    Returns a promise for the final reduction result

    PromiseStream.promise

    () => Promise

    Returns a promise fulfilled at the end of the stream, rejected if any errors events are emitted by the stream.

    For ReducePromiseStreams, the promise is for the final reduction result. Any stream errors or exceptions encountered while reducing will result with a rejection of the promise.

    Install

    npm i promise-streams

    DownloadsWeekly Downloads

    2,240

    Version

    2.1.1

    License

    MIT

    Last publish

    Collaborators

    • spion