如何创建一个随机间隔产生随机数的无限Observable？

Question

给定一个生成随机数的函数,您将如何创建一个随机间隔产生随机数的无限Observable？

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
}

以下是所需Observable的示例:

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

3ms     7ms       6ms   3ms     10ms

Answer 1

作为替代方案,如果您不想生活在令人困惑的超时世界中,您可以完全将其写为流:

// 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
}

这里的工作示例:http://jsbin.com/zipocaneya/edit？js,console

替代方法:首先创建随机数,然后添加延迟(如果执行时间无关紧要)

// 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));

附加说明:由于流之外没有并发或异步操作,而不是switchMap你也可以使用concatMap或flatMap- 在这种情况下它们都工作相同.

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