reactive-programming - Rx 操作符，以 combineLatest 开头但随后的行为类似于 withLatestFrom

Question

我正在寻找结合两个 Observables 的运算符直到 Observables 才发出任何东西已经发射了一个元素(类似于 combineLatest)，但随后仅通过组合来自一个 Observable 的元素来发射元素与另一个 Observable 最近发出的元素(类似于 withLatestFrom )。结果看起来像这样(y observable 是“控制”):



有这样的运营商吗？

Answer 1

我用 Java 解决了这个问题，但同样的理论应该对你有用。

你真正拥有的是两种基本模式； A combineLatest值后跟 withLatestFrom值。如 withLatestFrom首先触发然后你想跳过 combineLatest值(value)。

首先制作 withLatestFrom可观察:

Observable<Result> wlf = o1.withLatestFrom(o2, f::apply);

接下来我们要创建 combineLatest发出单个值的 observable。我们也想在 wlf 时停止这个 observable触发器:

Observable<Result> cl = Observable.combineLatest(o1, o2, f::apply)
    .take(1).takeUntil(wlf);

最后将这两个 observable 加在一起......为了方便起见，我制作了一个辅助方法来接受任何两个 observable 和一个双功能运算符:

public static <Result,
    Param1, Source1 extends Param1,
    Param2, Source2 extends Param2>
Observable<Result> combineThenLatestFrom(
    final Observable<Source1> o1,
    final Observable<Source2> o2,
    final BiFunction<Param1, Param2, Result> f
) {
    final Observable<Result> base = o1
        .withLatestFrom(o2, f::apply);
    return Observable
        .combineLatest(o1, o2, f::apply)
        .take(1).takeUntil(base)
        .mergeWith(base);
}

这是我用来验证该方法的测试代码:

public static void main(final String[] args) {
    final TestScheduler scheduler = new TestScheduler();
    final TestSubject<String> o1 = TestSubject.create(scheduler);
    final TestSubject<String> o2 = TestSubject.create(scheduler);
    final Observable<String> r = combineThenLatestFrom(o1, o2, (a, b) -> a + b);
    r.subscribe(System.out::println);
    o1.onNext("1");
    o1.onNext("2");
    o2.onNext("A");
    o2.onNext("B");
    o2.onNext("C");
    o2.onNext("D");
    o1.onNext("3");
    o2.onNext("E");
    scheduler.triggerActions();
}

哪些输出:

2A
3D