javascript - 如何使 withLatestFrom() 与自身一起工作？

Question

我有中间流，它绑定(bind)到源，但也可以触发来自其他源的事件(例如用户输入)。在我的程序的其他地方，我有派生流，它需要将来自中间的新脉冲与源的最后一个值进行比较，所以这一切归结为这样的代码:

const source = new Rx.Subject;
const derived = new Rx.Subject;
derived.subscribe( () => console.log( "derived" ) );
const intermediate = new Rx.Subject;

//motivation of having "intermediate" is that sometimes it fires on it's own:
source.subscribe( intermediate );

intermediate
    .withLatestFrom( source )
    .subscribe( derived );

source.next();

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/5.4.3/Rx.min.js"></script>

问题是“派生”消息永远不会被打印(源中的第一个事件被忽略)。我怎样才能创建一些流，使来自中间流的每条消息都获得源的最后一个值，即使它当前是源的传播？

Answer 1

如果我理解正确的话，您有一个source流，然后是一个intermediate流，它订阅并发出所有的source 的项目并混合来自用户输入的其他一些项目。那么您有两个不同的请求，我将分别提供建议:

发射源的值以及来自中间的计时: combineLatest会轻松完成这项工作:

const source_when_intermediate : Observable<TSource> = intermediate.combineLatest(source, (i, s) => s);

将 intermediate 的最新值与 source 进行比较:虽然这听起来非常相似，但 combineLatest 却并非如此非常安全，因为如果你做一些简单的事情，比如:

const is_different : Observable<bool> = intermediate.combineLatest(source, (i, s) => i !== s);

在 source 发出新值之前，

intermediate 可能不会从 source 发出给定值，并且您可能会认为该过时值是唯一的到中级。

相反，为了获得最大的安全性，您需要缓冲并使用主题:

// untested
const derived : Observable<TSource> = (function() {
  const source_buffer = new Queue();
  const subj = new Subject();
  source.forEach(s => {
    source_buffer.enqueue(s);
  });
  intermediate.forEach(i => {
    subj.onNext(source_queue.peek() === i);
    if(source_buffer.peek() === i) source_buffer.dequeue();
  });
  Promise.all([
    source.toPromise(),
    intermediate.toPromise()
  ]).then(() => subj.onClose());
  return subj.asObservable();
})();