rxjs-stream

Note With the addition of Async Iternables, this library is no longer necessary.

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';letdata='This is a bit of text to have some fun with';letsrc=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';letdata='This is a bit of text to have some fun with';letwordObj=data.split(' ').map((text)=>({ text }));letsrc=Rx.Observable.from(wordObj);letstream=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 stdoutletob=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*asloremIpsumfrom'lorem-ipsum';import{rxToStream}from'rxjs-stream';letbook=loremIpsum({count: 1000,format: 'plain',units: 'paragraphs'});letwords=Rx.Observable.from(book.split(/\b/));letwordsBuffered=words.bufferCount(1000).map((words)=>words.join(''));letstream=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