javascript - 让任何 Observable 表现得像 ReplaySubject

Question

我一直在尝试让我的流仅重播最新的值，但运气不佳。基本上，我有一个 replaySubject(1) 在我不能手动 .onNext 的地方。我想做的是接受它，将它映射到其他东西，然后添加一个初始值。

我的第一直觉是:

var testInStream = new Rx.ReplaySubject(1);
var testOutStream = testInStream.startWith(-1);

testInStream.onNext(0);

setTimeout(function(){
       testOutStream.subscribe(function(x){ 
       console.log("stream:", x);
   });
}, 1000);

哪个打印

stream: test -1
stream: test 0

但是，我想找到一种方法只打印stream: test 0。我能想到的最接近的事情是

var testInStream = new Rx.ReplaySubject(1);
var testOutStream = testInStream.startWith(-1);

testInStream.onNext(0);

var wrapperSubject = new Rx.ReplaySubject(1);
testOutStream.subscribe(function(new_val){ wrapperSubject.onNext(new_val) });

setTimeout(function(){
  wrapperSubject.subscribe(function(x){ 
    console.log("stream:", x);
  });
}, 1000);

哪个返回

stream: test 0

这就是我想要的。但是，我宁愿不创建中间 replaySubject，我宁愿只调用类似 testOutStream.latest() 或 testOutStream.replay(1) 但我还没有让他们中的任何一个工作。

感谢任何帮助，这里是JS Bin用我的例子。

Answer 1

您可以使用shareReplay(1)，示例代码为here

var validation$ = Rx.Observable
 .fromEvent(document.getElementById('validation'), 'click')
 .map(function(_, index){
   return ++counter;
   })
 .do(emits(ta_validation, 'index'))
 .shareReplay(1)

validation$.subscribe(function(){})

setTimeout(function() {
  validation$
  .do(emits(ta_result, 'second subscription'))
  .subscribe(function(){    });}, 3000)

我天真地尝试了 replay(1) 和 connect 但奇怪的是整个流值都被重播了。所以最好是使用 shareReplay。

另请记住，根据您要对“最新”值执行的操作，可以使用以下运算符:

• combineLatest
• withLatestFrom