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

    1.1.0 • Public • Published

    rxlax

    npm version Build Status Coverage Status Dependencies Status

    Make Rx.js to relax a bit.

    This lib is useful to handle the backpressure problem with Rx.js. Currently there are other methods to handle this problem, but I've found that these methods are not always suitable for all cases, or the resulting code is just too complicated to achieve a simple solution. This lib try to solve this problematic with just a single operator.

    • Node.js & Browser support
    • Rx.js 6.x
    • Zero dependencies
    • TypeScript support

    Technically this operator map the source data to observable of resulting data, so you have to use another operator like mergeAll to retrieve the resulting data, or in other words convert a higher-order Observable into a first-order Observable.

    Example

    const { mergeAll, toArray } = require("rxjs/operators");
    const { rxlax } = require("rxlax");
     
    function getBigAndFastFiringObservable() {
      // Return a fast-firing observable, like an array or a file read
    }
     
    async function slowAsyncProcess(data) {
      await doSomethingCool(data);
      return {
        ts: new Date(),
        id: data.id
      };
    }
     
    const results = await getBigAndFastFiringObservable()
      .pipe(rxlax(slowAsyncProcess, { concurrency: 10 }))
      .pipe(mergeAll())
      .pipe(toArray())
      .toPromise();

    Mapper

    The first argument is a map function, that takes a stream element as argument and returns an observable input. This is the async process that have to be limited according to the speed of the data source.

    Options

    The second argument are the options.

    interface Options<T> {
      concurrency?: number;
      queue?: () => Queue<T>;
    }

    Concurrency

    Number of concurrent jobs, default to 16.

    Custom queue

    By default, all queued data is saved in memory, tecnically a simple array, but if the data to precess is enormous, or the processing time is very long, it is recommend to use a custom queue to buffer the data.

    interface Queue<T> {
      shift: () => Promise<T | undefined>;
      push: (entry: T) => Promise<any>;
      clear: () => Promise<any>;
    }

    A custom queue have to implement 3 methods used internally by rxlax, all of them are async, so you can use any DB as queue storage.

    push(entry: T): Promise

    Add one element to the end of the queue.

    shift(): Promise<T | undefined>

    Removes the first element from the queue and returns that removed element.

    clear(): Promise

    Clear the queue. This method is always fired just before the end of the overall process.

    Returns all errors

    If the option multiError is true, and more than one error are collected, the resulting error will be a custom instance whit an errors property.

    Install

    npm i rxlax

    DownloadsWeekly Downloads

    15

    Version

    1.1.0

    License

    MIT

    Unpacked Size

    13.1 kB

    Total Files

    13

    Last publish

    Collaborators

    • greguz