delayedRefCount-imperative.js

function delayedRefCount(delay) {
  return (source) => {
    let length = 0;
    let timeout;
    const onNewSubscriber = () => {
      const onUnsubscribe = () => {
        length--;
        if (length === 0) {
          timeout = setTimeout(() => {
            source.connect().unsubscribe();
          }, delay);
        }
      };
      length++;
      if (length === 1) {
        source.connect();
      }
      if (timeout) {
        clearTimeout(timeout);
        timeout = null;
      }
      return onUnsubscribe;
    };
    const obs = () => source;
    return Rx.Observable.using(onNewSubscriber, obs);
  };
}

const source = Rx.Observable
  .interval(1000)
  .do(x => console.log(x))
  .take(10)
  .publishReplay(1)
  .let(delayedRefCount(3000));