创建一个延迟下一个值的 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);

Answer 2

这是 Christopher 的解决方案，修改为运算符。

throttleImmediate 运算符仅存储来自源的最新值，直到给定选择器完成。它会在每次完成后立即触发缓存值（如果存在）。 它最适合在选择器有副作用（例如动画）时使用。

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

Rx.Observable.prototype.throttleImmediate = function (selector) {
    var source = this;

    return 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 () {
          delaying = true;
          selector(value).subscribe(
            observer.onNext.bind(observer),
            observer.onError.bind(observer),
            function(){
              if (hasValue) {
                hasValue = false;
                sendValue();
              } else {
                delaying = false;
              }
            }
          );
        }

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

source
  .throttleImmediate(function(data){
    var delay;

    if(data%2==0)
      delay=500;
    else
      delay=1000;

    return Rx.Observable.timer(delay).map(function(){ return data; });
  })
  .subscribe(log)

这在向只有选择器知道要延迟的值的源施加压力时派上用场。

示例： 鉴于问题的大理石图。

假设第一个来源是带有要显示的 html 数据的 ajax 调用，ajaxPages 源自对导航栏的点击。 我们希望将它们与入口动画一起渲染，animatePage，其持续时间是动态的。

ajaxPages.throttleImmediate(animatePage).subscribe();

在这里，我们使用来自源的值对页面进行动画处理，跳过动画期间发出的所有值，除了最新的值。

在实践中，我们得到的是一个忽略点击的流，这些点击紧随其后的是其他点击，并且对于向用户显示毫无用处，因为它们会动画化，然后立即动画化。