Создайте Observable, который задерживает следующее значение

Это должно помочь.

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

Вот решение Кристофера, преобразованное в оператора.

Оператор 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)

Это удобно при обратном давлении на источники, где значение задержки известно только селектору.

Пример: учитывая мраморную диаграмму вопроса.

Предположим, что первым источником являются вызовы ajax с отображаемыми данными html, ajaxPages которые исходят из щелчков на панели навигации. И мы хотим визуализировать их вместе с анимацией входа animatePage, продолжительность которой является динамической.

ajaxPages.throttleImmediate(animatePage).subscribe();

Здесь мы анимируем страницы значениями из источника, пропуская все значения, которые генерируются в течение периода анимации, кроме самого последнего.

На практике мы получаем поток, который игнорирует щелчки, за которыми вскоре следуют другие щелчки, и бесполезно показывать пользователю, поскольку они будут анимироваться и немедленно анимируются.