How to create an infinite Observable that produces random numbers at random intervals?

Question

Given a function that generates random numbers, how would you create an infinite Observable that produces random numbers at random intervals?

function getRandomNumber() {
  // Assume this function returns a random number, e.g. 198
}

function getRandomDelay() {
  // Assume this function returns a random delay in ms, e.g. 2000
}

Here is an example of a desired Observable:

---198--------64-------3---2----------18------->  (indefinitely)

3ms     7ms       6ms   3ms     10ms

Answer 1

As an alternative, if you don't want to live in a confusing timeout-world, you could write this entirely as a stream:

// the stream
const randomizer$ = Rx.Observable.of("")
  .switchMap(() => Rx.Observable
             .timer(getRandomDelay())
             .mapTo(getRandomNumber()))
  .repeat();

// subscribe to it
randomizer$.subscribe(num => console.log("Random number after random delay" + num));


// your utility functions
function getRandomNumber() {
  return ~~(Math.random() * 200)
}

function getRandomDelay() {
  return Math.random() * 1000
}

Working example here: http://jsbin.com/zipocaneya/edit?js,console

---

Alternative: Create the random number first and then add a delay (if the time of execution does not matter)

// the stream
const randomizer$ = Rx.Observable.of("")
  .switchMap(() => Rx.Observable
             .of(getRandomNumber())
             .delay(getRandomDelay()
  )
  .repeat();

// subscribe to it
randomizer$.subscribe(num => console.log("Random number after random delay" + num));

---

Additional note: Since there is no concurrency or async-operation going on outside of the stream, instead of switchMap you could just as well use concatMap or flatMap - in this case they all work the same.

Answer 2

const { Observable }  = require("rxjs");

const ob = new Observable(sub => {
  let timeout = null;
  
  // recursively send a random number to the subscriber
  // after a random delay
  (function push() {
    timeout = setTimeout(
      () => {
        sub.next(getRandomNumber());
        push();
      },
      getRandomDelay()
    );
  })();
  
  // clear any pending timeout on teardown
  return () => clearTimeout(timeout);
});

ob.subscribe(console.log);