javascript - RxJS と到達不能要素

Question

オブザーバブルが与えられた場合、サブスクライバーに到達しない要素 (フィルターで除外されるなど) を知り、それに基づいてアクションを実行したいというシナリオがあります。このようなことを達成するための最良のアプローチは何でしょうか?

Answer 1

オプション1

パブリッシュと関数合成を組み合わせて使用​​します。

var Rx        = require('rx')
    log       = console.log.bind(console),
    source    = Rx.Observable.interval(100).take(100);
    published = source.publish(),
    accepted  = published.where(condition),
    rejected  = published.where(not(condition));

accepted.subscribe(log.bind(undefined, 'accepted: '));
rejected.subscribe(log.bind(undefined, 'rejected: '));

published.connect();

function condition (x) {
  return x % 4 === 0;
}

function not (func) {
  return function () {
    return !func.apply(this, arguments);
  }
}

オプション # 2

イベントにタグ付け (変更) し、最後の 1 秒でフィルター処理します。

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

source
  .map(toAcceptable)
  .doAction(function (x) {
    x.acceptable = x.acceptable && condition1(x.value);
  })
  .doAction(function (x) {
    x.acceptable = x.acceptable && condition2(x.value);
  })
  .doAction(function (x) {
    log(x.acceptable ? 'accepted:' : 'rejected:', x);
  })
  .subscribe();

function condition1 (x) {
  return x % 4 === 0;
}

function condition2 (x) {
  return x % 3 === 0;
}

function toAcceptable (x) {
  return {
    acceptable: true,
    value: x
  };
}