rx-stream-events

An experiment in using RxJS 5 to provide an interface for streaming in and out JSON events through stdin / stdout.

import {inputEvent$, subscribeToOutput} from "rx-stream-events";

// observable awaiting new line separated json messages from stdin of the form [string, any]
// the provided observable is shared, as the underlying stream is a shared resource itself.
var event$ = inputEvent$();

// Subscribes the first stream to stdout, writing each JSON event with a terminating new line.
// Uses sync write and does not buffer.
var logging = new Rx.Subject();
var subscription = subscribeToOutput(event$, logging);

logging.next("This will go to stderr!");

Closes itself when either of the input or output pipe closes (such as from SIGPIPE), and uses sync write on output, thus naturally chains itself and handles backpressure :)

Development

Provides a nix expression for quickly pulling up node and typescript.

nix-shell .