stream-to-observable

What is stream-to-observable?

The stream-to-observable npm package allows you to convert Node.js streams into observables, which can be particularly useful for handling asynchronous data streams in a more functional and reactive programming style.

What are stream-to-observable's main functionalities?

Convert Readable Stream to Observable

This feature allows you to convert a Node.js Readable stream into an Observable. The code sample demonstrates how to create a Readable stream and convert it into an Observable, which can then be subscribed to for handling data, errors, and completion.

const streamToObservable = require('stream-to-observable');
const { Readable } = require('stream');
const { Observable } = require('rxjs');

const readableStream = new Readable({
  read() {
    this.push('data1');
    this.push('data2');
    this.push(null); // No more data
  }
});

const observable = streamToObservable(readableStream);

observable.subscribe({
  next: data => console.log(data),
  error: err => console.error(err),
  complete: () => console.log('Stream complete')
});

Convert Writable Stream to Observable

This feature allows you to convert a Node.js Writable stream into an Observable. The code sample demonstrates how to create a Writable stream and convert it into an Observable, which can then be subscribed to for handling data, errors, and completion.

const streamToObservable = require('stream-to-observable');
const { Writable } = require('stream');
const { Observable } = require('rxjs');

const writableStream = new Writable({
  write(chunk, encoding, callback) {
    console.log(chunk.toString());
    callback();
  }
});

const observable = streamToObservable(writableStream);

observable.subscribe({
  next: data => console.log(data),
  error: err => console.error(err),
  complete: () => console.log('Stream complete')
});

Other packages similar to stream-to-observable

rxjs

RxJS is a library for reactive programming using Observables, to make it easier to compose asynchronous or callback-based code. While RxJS does not directly convert streams to observables, it provides extensive functionality for creating and manipulating observables, which can be used in conjunction with Node.js streams.

most

Most.js is a high-performance FRP (Functional Reactive Programming) library. It provides a similar reactive programming model to RxJS but focuses on performance. Like RxJS, it does not directly convert streams to observables but offers powerful tools for working with asynchronous data streams.

baconjs

Bacon.js is a small functional reactive programming library for JavaScript. It provides a way to work with events and asynchronous data streams. While it does not directly convert Node.js streams to observables, it offers similar reactive programming capabilities.

README

stream-to-observable

Convert Node Streams into ECMAScript-Observables

Observables are rapidly gaining popularity. They have much in common with Streams, in that they both represent data that arrives over time. Most Observable implementations provide expressive methods for filtering and mutating incoming data. Methods like .map(), .filter(), and .forEach behave very similarly to their Array counterparts, so using Observables can be very intuitive.

Learn more about Observables

Install

$ npm install --save stream-to-observable

# You also need to install an Observable implementation (pick one):

$ npm install --save zen-observable rxjs

Usage

const fs = require('fs');
const split = require('split');

// You provide the Observable implmentation
const Observable = require('zen-observable')
const streamToObservable = require('stream-to-observable')(Observable);

const readStream = fs
  .createReadStream('./hello-world.txt', {encoding: 'utf8'})
  .pipe(split()); // chunks a stream into individual lines

streamToObservable(readStream)
  .filter(chunk => /hello/i.test(chunk))
  .map(chunk => chunk.toUpperCase())
  .forEach(chunk => {
    console.log(chunk); // only the lines containing "hello" - and they will be capitalized
  });

There are convenience imports for rxjs observables and zen-observables:

const streamToObservable = require('stream-to-observable/zen'); // zen-observables
// or
const streamToObservable = require('stream-to-observable/rxjs-all'); // full rxjs implementation
// or
const streamToObservable = require('stream-to-observable/rxjs'); // minimal rxjs implementation
// you can patch the minimal rxjs.
require('rxjs/add/operator/map');

None of the above implementations are included as dependencies of this package, so you still need to install them yourself using npm install. If using the minimal rxjs import, be sure to see the documentation regarding patching it with additional convenience methods.

API

streamToObservable(stream, [options])

stream

Type: ReadableStream

Note: stream can technically be any EventEmitter instance. By default the stream-to-observable listens to the standard Stream events (data, error, and end), but those are configurable via the options parameter. If you are using this with a standard Stream, you likely won't need the options parameter.

options

await

Type: Promies

If provided, the Observable will not "complete" until await is resolved. If await is rejected, the Observable will immediately emit an error event and disconnect from the stream. This is mostly useful when attaching to the stdin or stdout streams of a child_process. Those streams usually do not emit error events, even if the underlying process exits with an error. This provides a means to reject the Observable if the child process exits with an unexpected error code.

endEvent

Type: String or false
Default: "end"

If you are using an EventEmitter or non-standard Stream, you can change which event signals that the Observable should be completed.

Setting this to false will avoid listening for any end events.

Setting this to false and providing an await Promise will cause the Observable to resolve immediately with the await Promise (the Observable will remove all it's data event listeners from the stream once the Promise is resolved).

errorEvent

Type: String or false
Default: "error"

If you are using an EventEmitter or non-standard Stream, you can change which event signals that the Observable should be closed with an error.

Setting this to false will avoid listening for any error events.

dataEvent

Type: String
Default: "data"

If you are using an EventEmitter or non-standard Stream, you can change which event causes data to be emitted to the Observable.

Learn about Observables

• Overview: https://github.com/zenparsing/es-observable
• Formal Spec: https://zenparsing.github.io/es-observable/
• egghead.io lesson - Video
• rxjs observables - Observables Implementation
• zen-observables - Observables Implementation

Transform Streams

data events on the stream will be emitted as events in the Observable. Since most native streams emit chunks of binary data, you will likely want to use a TransformStream to convert those chunks of binary data into an object stream. split is just one popular TransformStream that splits streams into individual lines of text.

Caveats

It is important Note that using this module disables back-pressure controls on the stream. As such it should not be used where back-pressure throttling is required (i.e. high volume web servers). It still has value for larger projects, as it can make unit testing streams much cleaner.

License

MIT © James Talmage