1

特定のコンポーネントからの送信変更フィードである Rx ストリームがあります。

onNextに渡されたアイテムが蓄積され、バッチ モードが終了したときにのみ渡されるように、ストリームでバッチ モードに入ることができるようにしたい場合があります。

通常、ストリームを介して単一のアイテムを渡します。

stream.onNext(1); 
stream.onNext(2);

onNextに渡された項目とsubscribeで受信された項目の間には 1 対 1 のマッピングがあるため、前のスニペットでは、値 1 と 2 を使用してsubscribeを 2 回呼び出すことになります。

私が探しているバッチモードは、次のように機能する可能性があります。

stream.enterBatchMode();
stream.onNext(1);
stream.onNext(2);
stream.exitBatchMode();

この場合、サブスクライブが単一の連結配列 [1, 2] で 1 回だけ呼び出されるようにします。

繰り返しますが、バッチ モードが必要な場合は、連結されていないアイテムを渡す場合もあります。

Rx でこの動作を実現するにはどうすればよいですか?

注: 以前はonNextを介して配列を渡していましたが、これはほとんどの場合、個別モードでもバッチ モードでも型が同じになるようにするためです。

例えば：

stream.onNext([1, 2, 3]); 
stream.onNext([4, 5, 6]);

Subscribe は [1, 2, 3] の次に [4, 5, 6] を受け取りますが、バッチモードでは、Subscribe は連結された結果 [1, 2, 3, 4, 5, 6] を受け取ります。

したがって、このように見ると、非連結モードと連結モードに似ています(個別モードとバッチモードではなく)。しかし、問題は非常に似ていると思います。

4

3 に答える 3

2

これは、James World の c# ソリューションの精神を JavaScript に変換し、特定の質問に合わせて調整したものです。

var source = new Rx.Subject();
source.batchMode = new Rx.BehaviorSubject(false);
source.enterBatchMode = function() { this.batchMode.onNext(true); };
source.exitBatchMode = function() { this.batchMode.onNext(false); };
var stream = source
  .window(source.batchMode
    .distinctUntilChanged()
    .skipWhile(function(mode) { return !mode; }))
  .map(function(window, i) {
    if ((i % 2) === 0) {
      // even windows are unbatched windows
      return window;
    }

    // odd windows are batched
    // collect the entries in an array
    // (e.g. an array of arrays)
    // then use Array.concat() to join
    // them together
    return window
      .toArray()
      .filter(function(array) { return array.length > 0; })
      .map(function(array) {
        return Array.prototype.concat.apply([], array);
      });
  })
  .concatAll();


stream.subscribe(function(value) {
  console.log("received ", JSON.stringify(value));
});

source.onNext([1, 2, 3]);

source.enterBatchMode();
source.onNext([4]);
source.onNext([5, 6]);
source.onNext([7, 8, 9]);
source.exitBatchMode();

source.onNext([10, 11]);

source.enterBatchMode();
source.exitBatchMode();

source.onNext([12]);

source.enterBatchMode();
source.onNext([13]);
source.onNext([14, 15]);
source.exitBatchMode();
<script src="https://getfirebug.com/firebug-lite-debug.js"></script>
<script src="https://rawgit.com/Reactive-Extensions/RxJS/master/dist/rx.all.js"></script>

于 2014-11-26T19:35:12.120 に答える