Neutral Political Machine

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

    1.7.47 • Public • Published

    pulls

    pulls - readable streams for Node.js and Browsers.

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

    Streams are:

    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.

    Keywords

    none

    Install

    npm i pulls

    DownloadsWeekly Downloads

    9

    Version

    1.7.47

    License

    ISC

    Unpacked Size

    300 kB

    Total Files

    294

    Last publish

    Collaborators

    • slava.hatnuke