Rx-stream NPM | npm.io

RxStream

TypeScript port of Monix reactive streams module. All credits goes to monix authors!

WARNING! Work in progress!

This library is Work in Progress, api may change until version 1.0.0.

Usage

Install rx-stream library:

npm install --save rx-stream

Hello World

Create an Observable and observe it's events:

import Observable from 'rx-stream';
import {Future} from 'funfix';

Observable.range(0, 10)
    .mapFuture(n => Future.pure(n).delayResult(1000))
    .foreach((e) => {
        console.log('got item', e);
    })

`takeUntil` for synchronous sources

For synchronous sources, in order to use takeUntil and onErrorRestart - need to add asyncBoundary (ex: bufferWithPressure), otherwise event loop may never reach takeUntil

let failed = false;

Observable.loop()
  .map((n): number => {
    // will throw here
    if (n == 3  && !failed) {
      failed = true;
      throw new Error('something went wrong');
    }

    return n;
  })
  .bufferWithPressure(10) // this will break synchronous loop, to "make room" async events (sigTrigger)
  .onErrorRestartUnlimited()
  .takeUntil(sigTrigger)