rxdable

Utility lib to work with Node.js streams and Rx.js.

• Node.js >= 10.x
• Rx.js 6.x
• Zero dependencies
• TypeScript support

Observable to Readable stream

const { getStreamByObservable } = require("rxdable")

const readableStream = getStreamByObservable(observable)

Readable stream to Observable

const { createReadStream } = require("fs")
const { subscribeToStream } = require("rxdable")
const { Observable } = require("rxjs")

function fileRead(file, encoding = "utf8") {
  return new Observable(subscriber => {
    return subscribeToStream(createReadStream(file, encoding), subscriber)
  })
}

Readable stream to Observable (faster way)

const { getObservableByStream } = require("rxdable")

const observable = getObservableByStream(readableStream)

WARNING: This function will create an Observable ables to be subscribed just one time.

Operator to Transform stream

const { count } = require("rxjs/operators")
const { getStreamByOperator } = require("rxdable")

const countStream = getStreamByOperator(count())

Multiple operators are supported:

const { map } = require("rxjs/operators")
const { getStreamByOperator } = require("rxdable")

const firstUppercasedLetterStream = getStreamByOperator(
  map(value => value.toString()),
  map(value => value.substr(0, 1)),
  map(value => value.toUpperCase())
)

Passthrough (no operators) is supported:

const { getStreamByOperator } = require("rxdable")

const passthroughStream = getStreamByOperator()

Writable/Duplex/Transform stream to operator

const { createWriteStream } = require("fs")
const { pipeObservableToStream } = require("rxdable")
const { Observable } = require("rxjs")

function fileWrite(file, encoding = "utf8") {
  return observable => {
    return new Observable(subscriber => {
      return pipeObservableToStream(
        observable,
        createWriteStream(file, encoding),
        subscriber
      )
    })
  }
}

Writable/Duplex/Transform stream to operator (faster way)

const { getOperatorByStream } = require("rxdable")

const operator = getOperatorByStream(transformStream)

WARNING: This function will create an Operator ables to be used just one time.