pulls

pulls

pulls - readable streams for Node.js and Browsers.

Naturally readable streams are everything that implements async iterable protocol or iterable protocol.

Streams are:

• Array
• Async Iterable
• Iterable objects
• Generators
• and Node.js readable streams.

It's lightweight with 200LOC and fast - more than 2M/s throughput.

Basic usage examples

Consume array as stream for ex. and pipeline as map -> filter -> reduce

import {pipeline, stream} from "pulls";
import {map} from "pulls/map";
import {filter} from "pulls/filter";
import {reduce} from "pulls/reduce";
import {tap} from "pulls/tap";
import {sequence} from "pulls/sequence";


const expected = await pipeline(
    stream([1, 2, 3]), // automatically converts array to stream as async interable

    map((x) => x * 2), // multiple; = 2, 4, 6

    filter((x) => x > 2), // filter only x > 2; = 4, 6

    // types input:number, output:number[]
    reduce<number, number[]>((acc, value) => [...acc, value], []), // collect to array as expected

    tap(console.log) // prints output
);

// expected: [4, 6]
expect(expected).toEqual([4, 6])

One more example with async stream.

It takes 1s to calculate 1M dataset.

// consume async iterator as stream and sum some values of 1m
const expected = await pipeline(
    sequence(1000000), // creates low CPU/memory stream

    map((x) => x * 2), // multiple; = 2, 4, 6...

    filter((x) => x > 2), // filter only x > 2; = 4, 6....

    // types input:number, output:number
    reduce<number, number>((acc, value) => acc + value, 0), // sum values

    tap(console.log) // prints output; 999998999998
);

// expected: 999998999998
expect(expected).toEqual(999998999998)

It's pipeline to consume async generator as stream as ex.

import {pipeline} from "pulls";
import {tap} from "pulls/tap";

// consume generator/async iterable protocol as stream
async function* Generate123() {
    yield 1
    yield 2
    yield Promise.resolve(3)
}

const expected: number[] = [];
const collect = (value:number) => expected.push(value);

await pipeline(
    Generate123(), // consume generator as stream
    tap<number>(collect) // collect output to array as expected
)

// expected: [1, 2, 3]
expect(expected).toEqual([1, 2, 3])

Guide

TBD

Documentation

TBD

Stream examples

To get a point for a readable sources just check the information that I mentioned before. async iterable protocol or iterable protocol and more about iterating over async iterables. Or it could be enough to check examples here.

// stream 1
const asyncIterable = {
  [Symbol.asyncIterator]() {
    return {
      i: 0,
      next() {
        if (this.i < 3) {
          return Promise.resolve({ value: this.i++, done: false });
        }

        return Promise.resolve({ done: true });
      }
    };
  }
};

// stream 2
const myAsyncIterable = {
    async* [Symbol.asyncIterator]() {
        yield "hello";
        yield "async";
        yield "iteration!";
    }
};

// stream 3
let mySyncIterator = {
    next: function() {
        // ...
    },
    [Symbol.iterator]: function() { return this; }
};

// stream 4 / Node.js
const readableStream = fs.createReadStream(
  'tmp/test.txt', {encoding: 'utf8'});

// stream 5 / Node.js
const readableStream = fs.createReadStream(
  'tmp/test.txt', {encoding: 'utf8'});

// stream 6 / Node.js
import {Readable} from 'stream';
function* gen() {
  yield 'One line\n';
  yield 'Another line\n';
}
const readableStream = stream.Readable.from(gen(), {encoding: 'utf8'});

// stream 7 / Node.js
import {Readable} from 'stream';
const str = 'Some text!';
const readable = Readable.from(str, {encoding: 'utf8'});

Node.js streams cooperation

Node.js streams are readable and implements async iterable protocol and could be consumed in pipelines.

import {pipeline} from "pulls";

import {Readable} from 'stream';
function* gen() {
  yield 'One line\n';
  yield 'Another line\n';
}
const readableStream = Readable.from(gen(), {encoding: 'utf8'});

// example
await pipeline(readableStream, /* map(...), filter(...) */)

Good article how to convert async iterable to Node.js readable and back could help you to manage Node.js streams.

---

TODOs

More information will come soon about transformations, composition, and stream management and scalability.