Skip to content

paulpdaniels/rx-node-vx

Folders and files

NameName
Last commit message
Last commit date

Latest commit

 

History

6 Commits
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 

Repository files navigation

rx-node-vx

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

OVERVIEW

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

GETTING STARTED

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

Download the Source

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

Installing with NPM

npm install rx-node-vx

API

RxNode Methods

Event Handlers

Stream Handlers

RxNode.toEventEmitter(observable, eventName)

#

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.

Arguments

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

Returns

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

Example

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

Location

  • index.js

Stream Handlers

RxNode.fromStream(stream, finishEventName, dataEventName)

#

Converts a flowing stream to an Observable sequence.

Arguments

  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)

Returns

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

Example

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>

Location

  • index.js

RxNode.fromReadableStream(stream, dataEventName)

#

Converts a flowing readable stream to an Observable sequence.

Arguments

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

Returns

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

Example

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

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

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

RxNode.fromReadLineStream(stream)

#

Converts a flowing readable stream to an Observable sequence.

Arguments

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

Returns

(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'

Location

  • index.js

RxNode.fromWritableStream(stream)

#

Converts a flowing writeable stream to an Observable sequence.

Arguments

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

Returns

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

Example

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

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

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

Location

  • index.js

RxNode.fromTransformStream(stream)

#

Converts a flowing transform stream to an Observable sequence.

Arguments

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

Returns

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

Example

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

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

Location

  • index.js

RxNode.writeToStream(observable, stream, [encoding])

#

Writes an observable sequence to a stream.

Arguments

  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.

Returns

(Subscription): The subscription handle.

Example

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

Location

  • index.js

About

A port of the rx-node project for v5 and beyond

Topics

Resources

License

Stars

Watchers

Forks

Releases

No releases published

Packages

No packages published