我如何根据 属性 以不同的方式继续 Rx 链？

Question

我有一个基于 Flowable 数据值的方法，我需要以不同的方式继续 RX 链。 我的意思是，如果数据对象的 isOnline 属性 为真，那么我需要调用 scan(initial,selector) 但如果为假，则我需要调用 scan(selector)

 @NotNull
    public Flowable<Data> initialCall(
            @NotNull Flowable<Info> info, Data initial) {
        return  info
                .map()
                .switchMap(it -> call(Flowable.just(it),initial, it.isOnline));
    }



   private Flowable<Data> call (
            Flowable<A> a, 
            Data initial, boolean isOnline
    ) {
        return Flowable.combineLatest(
                a,
                b,
                (a, b) -> {
                    return ....;
                })
                .switchMap()
    ///here based on the Data isOnline property I need to call either 
    ///scan(initial, selector) or scan(selector) and then continue.... 
                .map()
                .distinctUntilChanged()
                .toObservable()
                .compose()
                .compose()
                .toFlowable(BUFFER)
    }

Answer 1

这是，你想要的吗？基于 isOnline 应用 scan 运算符，有或没有 seed 值。

import io.reactivex.rxjava3.core.BackpressureStrategy;
import io.reactivex.rxjava3.core.Flowable;
import io.reactivex.rxjava3.core.FlowableTransformer;
import org.junit.jupiter.api.Test;

class So65349760 {
  private static <C extends Data> FlowableTransformer<Data, Data> scan(
      Boolean isOnline, C initialValue) {
    if (isOnline) {
      return upstream -> {
        return upstream.scan(
            initialValue,
            (prev, current) -> {
              return new Data();
            });
      };
    } else {
      return upstream -> upstream.scan((prev, current) -> new Data());
    }
  }

  @Test
  void so65349760() {
    Flowable<Integer> a = Flowable.fromCallable(() -> 1);
    Flowable<String> b = Flowable.fromCallable(() -> "42");

    Data seed = new Data();

    call(a, b, seed, false).test().assertValueCount(1);
    call(a, b, seed, true).test().assertValueCount(2);
  }

  private <A, B, C extends Data> Flowable<Data> call(
      Flowable<A> a, Flowable<B> b, C init, boolean isOnline) {
    return Flowable.combineLatest(a, b, (v1, v2) -> 42)
        .switchMap(integer -> Flowable.just(new Data()))
        .compose(scan(isOnline, init))
        .map(d -> d)
        .distinctUntilChanged()
        .toObservable()
        .toFlowable(BackpressureStrategy.BUFFER);
  }

  private static class Data {}
}