observable - 创建一个延迟下一个值的 Observable

Question

我正在尝试使用 RxJS 创建一个可观察的对象，它可以执行如图所示的操作。

获取一个值并等待一段固定的时间才能获得
下一个。

下一个将是该周期内发出的最后一个值
等等，跳过其余部分。

如果等待时间间隔没有发出任何值，则下一个
应该立即抓取作为图像的最后一个示例
描绘。

Answer 1

这应该可以解决问题。

var Rx      = require('rx'),
    source  = Rx.Observable.interval(10).take(100),
    log     = console.log.bind(console);

Rx.Observable.create(function (observer) {

    var delaying = false,
        hasValue = false,
        complete = false,
        value;

    function onNext (x) {
      value = x;
      if (delaying) {
        hasValue = true;
      } else {
        sendValue();
      }
    }

    function sendValue () {
      observer.onNext(value);
      if (complete) {
        observer.onCompleted();
      } else {
        setTimeout(callback, 1000); // exercise for the reader. Use a scheduler.
      }
      delaying = true;
    }

    function callback () {
      if (hasValue) {
        hasValue = false;
        sendValue();
      } else {
        delaying = false;
      }
    }

    return source.subscribe(
        onNext,
        observer.onError.bind(observer),
        function () {
          if (hasValue) {
            complete = true;
          } else {
            observer.onCompleted();
          }
        }
      );
  })
  .subscribe(log);