Create coprophagus streams

coprophagus.js

const Rx = require('Rx')
const O = Rx.Observable
const R = require('ramda')

function SlimShady (signal$) {
  return signal$
    .delay(500).map((x) => x + 1)
}

const coprophagus = R.curry((funct, signal$) => {
  const sub = new Rx.Subject()
  const out$ = funct(O.merge(sub.asObservable(), signal$))
  return out$.multicast(sub).refCount()
})

const slimShady = coprophagus(SlimShady)

slimShady(O.just(0))
  .take(4)
  .subscribe((x) => console.log(x))