rxjs - 让 withLatestFrom 等待，直到所有源都产生一个值

Question

我正在以正常方式使用 RxJS 中的 withLatestFrom 运算符:

var combined = source1.withLatestFrom(source2, source3);

...主动收集来自 source2 和 source3 的最新发射，并仅在 source1 发射时发射所有三个值。

但我不能保证 source2 或 source3 会在 source1 产生值之前产生值。相反，我需要等到所有三个源都至少生成一个值，然后再让 withLatestFrom 执行其操作。

合约需要是:如果source1发出，那么当其他源最终产生时，combined将始终最终发出。如果 source1 在等待其他源时多次发出，我们可以使用最新的值并丢弃以前的值。 编辑:作为弹珠图:

--1------------2---- (source)
----a-----b--------- (other1)
------x-----y------- (other2)
------1ax------2by--


--1------------2---- (source)
------a---b--------- (other1)
--x---------y------- (other2)
------1ax------2by--


------1--------2---- (source)
----a-----b--------- (other1)
--x---------y------- (other2)
------1ax------2by--

我可以为此创建一个自定义运算符，但我想确保我没有错过使用普通运算符来执行此操作的明显方法。感觉几乎就像我想要 combineLatest 进行初始发射，然后切换到 withLatestFrom 从那时起，但我一直无法弄清楚知道如何做到这一点。

编辑:最终解决方案的完整代码示例:

var Dispatcher = new Rx.Subject();
var source1 = Dispatcher.filter(x => x === 'foo');
var source2 = Dispatcher.filter(x => x === 'bar');
var source3 = Dispatcher.filter(x => x === 'baz');

var combined = source1.publish(function(s1) {
    return source2.publish(function(s2) {
        return source3.publish(function(s3) {
            var cL = s1.combineLatest(s2, s3).take(1).do(() => console.log('cL'));
            var wLF = s1.skip(1).withLatestFrom(s2, s3).do(() => console.log('wLF'));

            return Rx.Observable.merge(cL, wLF);
        });
    });
});

var sub1 = combined.subscribe(x => console.log('x', x));

// These can arrive in any order
// and we can get multiple values from any one.
Dispatcher.onNext('foo');
Dispatcher.onNext('bar');
Dispatcher.onNext('foo');
Dispatcher.onNext('baz');

// combineLatest triggers once we have all values.
// cL
// x ["foo", "bar", "baz"]

// withLatestFrom takes over from there.
Dispatcher.onNext('foo');
Dispatcher.onNext('bar');
Dispatcher.onNext('foo');
// wLF
// x ["foo", "bar", "baz"]
// wLF
// x ["foo", "bar", "baz"]

Answer 1

我认为答案或多或少如您所描述的那样，让第一个值是combineLatest，然后切换到withLatestFrom。我的 JS 很模糊，但我认为它看起来像这样:

var selector = function(x,y,z) {};

var combined = Rx.Observable.concat(
    source1.combineLatest(source2, source3, selector).take(1),
    source1.withLatestFrom(source2, source3, selector)
);

您可能应该使用 publish 来避免多次订阅，因此看起来像这样:

var combined = source1.publish(function(s1)
{
    return source2.publish(function(s2)
    {
        return source3.publish(function(s3)
        {
            return Rx.Observable.concat(
                s1.combineLatest(s2, s3, selector).take(1),
                s1.withLatestFrom(s2, s3, selector)
            );
        });
    });
});

或使用箭头函数...

var combined = source1.publish(s1 => source2.publish(s2 => source3.publish(s3 => 
    Rx.Observable.concat(
        s1.combineLatest(s2, s3, selector).take(1),
        s1.withLatestFrom(s2, s3, selector)
    )
)));

编辑:

我发现 concat 存在问题，withLatestFrom 未获取值。我认为以下方法会起作用:

var combined = source1.publish(s1 => source2.publish(s2 => source3.publish(s3 => 
    Rx.Observable.merge(
        s1.combineLatest(s2, s3, selector).take(1),
        s1.skip(1).withLatestFrom(s2, s3, selector)
    )
)));

...因此使用 combineLatest 获取一个值，然后使用 withLatestFrom 获取其余值。