rxjs-stream

rxjs-stream

This is a simple library for converting to and from NodeJS stream and RxJS 7.

This was created to fill the gap left by rx-node, which only works with rxjs 4.

Installation

npm install --save rxjs rxjs-stream

Usage

Writing to a stream.

import { rxToStream } from 'rxjs-stream';

let data = 'This is a bit of text to have some fun with';
let src = Rx.Observable.from(data.split(' '));
rxToStream(src).pipe(process.stdout);

Writing objects to a stream

To write objects, you must pass in the ReadableOptions with objectMode to be true: { objectMode: true }

import { rxToStream } from 'rxjs-stream';

let data = 'This is a bit of text to have some fun with';
let wordObj = data.split(' ').map((text) => ({ text }));
let src = Rx.Observable.from(wordObj);
let stream = rxToStream(src, { objectMode: true });

Read from a stream

import { rxToStream, streamToStringRx } from 'rxjs-stream';

// Read stdin and make it upper case then send it to stdout
let ob = streamToStringRx(process.stdin).map((text) => text.toUpperCase());

rxToStream(ob).pipe(process.stdout);

Performance

It is recommended to buffer observable values before sending them to the stream. Node streams work better with fewer calls of a large amount of data than with many calls with a small amount of data.

Example:

import * as loremIpsum from 'lorem-ipsum';
import { rxToStream } from 'rxjs-stream';

let book = loremIpsum({ count: 1000, format: 'plain', units: 'paragraphs' });
let words = Rx.Observable.from(book.split(/\b/));
let wordsBuffered = words.bufferCount(1000).map((words) => words.join(''));
let stream = rxToStream(wordsBuffered);

stream.pipe(process.stdout);

Compatibility

This library is tested with Node 12 and above.

rx-stream	RxJS	Node
4.x	7.x	>=12
3.x	6.x	>=10

Changelog

5.0.0 (2022-06-07)

⚠ BREAKING CHANGES

• Drop support for Node 12

Note: The code has not changed, but the cost of maintaining the tools is too high.

Features

• Drop support for Node 12 (#349) (db331d6)

Bug Fixes

• Update devDependencies (#258) (df64568)
• Update dependencies and workflows (#348) (3d54289)
• Update dev dependencies (#304) (6b6a404)
• upgrade rxjs from 7.5.4 to 7.5.5 (#291) (4e89239)