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

Single operator:

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

const countStream = getStreamByOperator(count())

Multiple operators:

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

No operators (passthrough):

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.