A port of the rx-node project for v5 and beyond. Note that this is only for v5 and above. If you are still using RxJS 4 then you should be using: https://github.com/Reactive-Extensions/rx-node

This project provides Reactive Extensions for JavaScript (RxJS) bindings for Node.js and io.js to abstract over the EventEmitter, Streams and more.

There are a number of ways to get started with the RxJS bindings for Node.js.

To download the source of the Node.js Bindings for the Reactive Extensions for JavaScript, type in the following:

git clone https://github.com/paulpdaniels/rx-node-vx.git
cd ./rx-node-vx

npm install rx-node-vx

• toEventEmitter

• fromStream
• fromReadableStream
• fromReadLineStream
• fromWritableStream
• fromTransformStream
• writeToStream

# Ⓢ

Converts the given observable sequence to an event emitter with the given event name. The errors are handled on the 'error' event and completion on the 'end' event.

1. observable (Obsesrvable): The observable sequence to convert to an EventEmitter.
2. eventName (String): The event name to subscribe.

(EventEmitter): An EventEmitter which emits the given eventName for each onNext call in addition to 'error' and 'end' events.

var Rx = require('rxjs');
var RxNode = require('rx-node-vx');

var source = Rx.Observable.return(42);

var emitter = RxNode.toEventEmitter(source, 'data');

emitter.on('data', function (data) {
    console.log('Data: ' + data);
});

emitter.on('end', function () {
    console.log('End');
});

// Ensure to call publish to fire events from the observable
emitter.publish();

// => Data: 42
// => End

• index.js

# Ⓢ

Converts a flowing stream to an Observable sequence.

1. stream (Stream): A stream to convert to a observable sequence.
2. [finishEventName] (String): Event that notifies about closed stream. ("end" by default)
3. [dataEventName] (String): Event that notifies about incoming data. ("data" by default)

(Observable): An observable sequence which fires on each 'data' event as well as handling 'error' and finish events like end or finish.

var RxNode = require('rx-node-vx');

var subscription = RxNode.fromStream(process.stdin, 'end')
    .subscribe(function (x) { console.log(x); });

// => r<Buffer 72>
// => x<Buffer 78>

• index.js

# Ⓢ

Converts a flowing readable stream to an Observable sequence.

1. stream (Stream): A stream to convert to a observable sequence.
2. [dataEventName] (String): Event that notifies about incoming data. ("data" by default)

(Observable): An observable sequence which fires on each 'data' event as well as handling 'error' and 'end' events.

var RxNode = require('rx-node-vx');

var subscription = RxNode.fromReadableStream(process.stdin)
    .subscribe(function (x) { console.log(x); });

// => r<Buffer 72>
// => x<Buffer 78>

# Ⓢ

Converts a flowing readable stream to an Observable sequence.

1. stream (Stream): A stream to convert to a observable sequence.

(Observable): An observable sequence which fires on each 'line' event as well as handling 'error' and 'close' events.

var readline = require('readline');
var fs = require('fs');
var RxNode = require('rx-node-vx');

var rl = readline.createInterface({
  input: fs.createReadStream('sample.txt')
});

var subscription = RxNode.fromReadLineStream(rl)
    .subscribe(function (x) { console.log(x); });

// Prints contents of 'sample.txt' line by line:
// => rx
// => supports 'readline'

• index.js

# Ⓢ

Converts a flowing writeable stream to an Observable sequence.

1. stream (Stream): A stream to convert to a observable sequence.

(Observable): An observable sequence which fires on each 'data' event as well as handling 'error' and 'finish' events.

var RxNode = require('rx-node');

var subscription = RxNode.fromWritableStream(process.stdout)
    .subscribe(function (x) { console.log(x); });

// => r<Buffer 72>
// => x<Buffer 78>

• index.js

# Ⓢ

Converts a flowing transform stream to an Observable sequence.

1. stream (Stream): A stream to convert to a observable sequence.

(Observable): An observable sequence which fires on each 'data' event as well as handling 'error' and 'finish' events.

var RxNode = require('rx-node-vx');

var subscription = RxNode.fromTransformStream(getTransformStreamSomehow());

• index.js

# Ⓢ

Writes an observable sequence to a stream.

1. observable (Observable): Observable sequence to write to a stream.
2. stream (Stream): The stream to write to.
3. [encoding] (String): The encoding of the item to write.

(Subscription): The subscription handle.

var Rx = require('rxjs');
var RxNode = require('rx-node-vx');

var source = Rx.Observable.range(0, 5);

var subscription = RxNode.writeToStream(source, process.stdout, 'utf8');

// => 01234

• index.js