Promistreamus

Convert Stream into an Iterator yielding promises of values

Allows foreach of promises for a stream:

// preudo-code
foreach ( promise in promistreamus(stream) ) {
    promise.then(function(value) { /* use the value */ });
}

Promistreamus converts stream's values into promises. This allows you to treat streams in a sync "pull" fashion, even if the stream values are not ready yet. Wrapping a stream with Promistreamus provides an iterator function. Calling it returns a thenable promise of a value. Once available, the promise is resolved with the value. When the stream ends, all pending promises are resolved with the undefined value. On error, all pending promises are rejected with that error.

Using Promistreamus

1var promistreamus = require("promistreamus");
2var iterator = promistreamus(stream); // Create an iterator from a stream
3
4// Stream item processing function
5var processor = function() {
6    // Get the promise of the next stream value from the iterator
7    return iterator().then(function(value) {
8        if (value === undefined) {
9            // we are done, no more items in the stream
10            return;
11        }
12        // Process the value
13        ...
14        return processor(); // Continue to the next item
15    });
16};
17
18// Process stream one item at a time
19processor().then(function() {
20    // all items were successfully processed
21}, function(err) {
22    // processing failed
23});

Delayed initialization

In an edge case when iteration function is needed before the stream is available, promistreamus can be delay-created with an undefined call, and initialized later with init(stream) function.

1var promistreamus = require("promistreamus");
2// Create a non-initialized iterator which has .init(stream) method
3var iterator = promistreamus();
4...
5// init() can be called even after the iterator function has been called
6iterator.init(stream);

Note that if the filtering function is needed, it should still be passed as before:

1var iterator = promistreamus(undefined, function(row) {...});

Processing multiple values at once

The iterator function may be called more than once, without waiting for the first promise to be resolved.

1// Process all items, 3 items at a time (example uses bluebird npm)
2var threads = [processor(), processor(), processor()];
3return Promise.all(threads).then(function() {
4    // all items were successfully processed
5}, function(err) {
6    // processing failed
7});

Cancellation

Streaming can be stopped by calling cancel() function on the iterator. All pending promises will be rejected, and the stream will be paused.

1iterator.cancel();

Filtering and converting values of a promistreamus stream

One may wish to map all values of a given promistreamus stream by using a conversion function, and/or to filter them out.

1var iterator = promistreamus(stream); // Create an iterator from a stream
2
3var iterator2 = promistreamus.select(iterator, function (value) {
4     // process the value, and either return a new value, a promise of a new value, or undefined to skip it
5     return ...;
6});

Joining multiple streams

Given a promistreamus style stream of streams - an iterator function that returns promises of sub-iterators, one may wish to present the values from all sub-iterators together, just like the .NET's LINQ SelectMany() function.

1var items = promistreamus(stream); // Create an iterator from a stream
2
3// Convert each value into a separate promistreamus iterator
4var itemsOfItems = promistreamus.select(iterator, function (value) {
5    return promistreamus(createStream(value));
6});
7
8// Flatten out all subitems
9var flattenedItems = promistreamus.flatten(itemsOfItems);
10