RXJava-组合最新而不丢失任何结果

Question

我想把两个观测值结合起来，一个发射n个项目，另一个只发射1个。

组合测试（CombineTest）将等待，直到两个可观察对象都至少发出一个项目，然后组合最新发出的项目，直到两个可观察对象都完成。按以下时间顺序考虑：

• 可观测A-

组合测试（CombineTest）只会将可观察到的1的结果2与可观察到的2的结果1进行组合（可以在此处轻松测试：http://rxmarbles.com/#combineLatest).

我需要什么

我需要合并两个可观测的所有项目，无论哪一个更快。我该怎么做？

结果应该是（总是，独立的观察开始发射项目第一！）：

• A1与B1组合
• A2结合B1

Answer 1

注意：这是未经测试的：

从可观察序列a创建一个Replaysubject。对于序列b上发出的每个值，将该值与重放主题组合以创建新的可观察的Pair

public static <A, B> Observable<Pair<A, B>> permutation(
    Observable<A> observableA, 
    Observable<B> observableB, 
) {
    ReplaySubject<A> subjectA = ReplaySubject.create();
    observableA.subscribe(subjectA::onNext);
    return observableB.flatMap(b -> subjectA.map(a -> Pair.of(a, b)));
}

Answer 2

老问题，但我遇到了同样的问题。这是我的想法。首先，非工作版本：

    Observable<Integer> emitsMany = Observable.range( 1, 10 )
            .concatMap( i -> Observable.just( i ).delay( 1, TimeUnit.SECONDS ))
            .doOnNext( i -> System.out.println( "produced " + i ));

    Observable<Boolean> emitsOne = Observable.just( true )
            .delay( 3, TimeUnit.SECONDS )
            .doOnNext( b -> System.out.println( "produced " + b ));

    Observable.combineLatest(
            emitsMany, emitsOne,
            ( i, b ) -> "consumed " + i + " " + b )
    .blockingSubscribe( System.out::println );

果然，许多排放物的前几次排放量都会下降：

produced 1
produced 2
produced 3
produced true
consumed 3 true
produced 4
consumed 4 true
. . .

我想这是解决办法。。首先，我们需要将emitsOne打包成一个可以毫不延迟地继续返回先前观察到的值的东西。我不知道有哪个操作符能做到这一点，但行为主体（BehaviorSubject）能做到这一点。

接下来，我们可以将concatMap与嵌套的take（1）Observable一起使用：

    Observable<Integer> emitsMany = Observable.range( 1, 10 )
            .concatMap( i -> Observable.just( i ).delay( 1, TimeUnit.SECONDS ))
            .doOnNext( i -> System.out.println( "produced " + i ));

    Observable<Boolean> emitsOne = Observable.just( true )
            .delay( 3, TimeUnit.SECONDS )
            .doOnNext( b -> System.out.println( "produced " + b ));

    BehaviorSubject<Boolean> emitsOneSubject = BehaviorSubject.create();
    emitsOne.subscribe( emitsOneSubject::onNext );

    emitsMany.concatMap( i -> emitsOneSubject
            .take( 1 )
            .map( b -> "consumed " + i + " " + b ))
    .blockingSubscribe( System.out::println );

我们现在得到了所有的组合：

produced 1
produced 2
produced true
consumed 1 true
consumed 2 true
produced 3
consumed 3 true
produced 4
consumed 4 true
produced 5
consumed 5 true
produced 6
consumed 6 true
produced 7
consumed 7 true
produced 8
consumed 8 true
produced 9
consumed 9 true
produced 10
consumed 10 true