rxjava2 基于发射数量的分支逻辑

Question

我想根据上游的排放数量来分支逻辑。

确切地说，我想要：

1. 上游为空时不会发生任何事情
2. 当上游只发出一个值然后完成时触发一个分支
3. 当上游发出多个值然后完成时触发一个分支。

我正在为如何解决这个问题而摸不着头脑，我想出了一些可行但看起来非常冗长的东西。我想知道是否有更简单的方法。

此解决方案基于 RxJava2Extensions 项目中的 valve 运算符。

解决方案的大纲如下：

1. 使用publish(foo)多次订阅上游
2. 将merge 用于两个逻辑分支
3. 对于“多个排放逻辑”使用最初关闭valve 并在第二次排放时打开它，如果没有排放或只有一个排放，则打破阀门。通过打破阀门，我的意思是终止控制Publisher
4. 对于“仅一种发射逻辑”，使用最初关闭 valve。使用ambArray 可以在没有排放或第二次排放时打破阀门，或者在恰好有一次排放时打开阀门。

所以这似乎可行，尽管我担心的是：

1. 它的功能看起来设计过头了。这能不能写得更简洁明了？
2. 整个阀门破坏业务将触发我刚刚吞下的异常，但可能还有其他与阀门无关的异常，我可能应该在这里区分并让它们沿流向下传播。 [编辑]阀门破裂很重要，因此单排放逻辑的阀门不会累积用于多排放逻辑的排放，也不会以这种方式占用内存[/EDIT]

代码如下：

Flowable.just(1,2,3,4,5) // +1 emissions
    //Flowable.just(1) // 1 emission
    //Flowable.empty() // 0 emissions
            .publish( //publish so that you get connectableFlowable inside
                f ->
                    Flowable.merge( //merge for the logic split
                        f.compose(
                            valve(f.scan(0, (sum, i) -> sum + 1) //scan to emit progressive count
                                   .filter(i -> i > 1) //filter for when count > 1
                                   .take(1) //take just first such count
                                   .concatMap(__ -> Flowable.<Boolean>never().startWith(true))  //and open the valve
                                   .switchIfEmpty(Flowable.empty()), //break the valve if there was just 1 element
                                  false) //start with the valve closed
                        )
                         .onErrorResumeNext(Flowable.empty()) //swallow the broken valve exception???
                         .map(__ -> "more than one elements!"), //here goes logic for +1 emissions
                        f.compose(
                            valve(
                                Flowable.ambArray(
                                    f.scan(0, (sum, i) -> sum + 1) //do progressive counts
                                     .switchIfEmpty(Flowable.never()) //if there was no elements then never end this guy
                                     .filter(i -> i > 1) //filter > 1
                                     .take(1) //take just first one
                                     .concatMap(
                                         __ -> Flowable.<Boolean>empty()) //if there was > 1 element then emit empty and break the valve so we
                                                                          //don't accumulate byte arrays that are meant for multipart upload
                                    ,
                                    f.count() //count the stream
                                     .map(c -> c == 1) //open valve if the count was 1
                                     .toFlowable()
                                     .concatWith(Flowable.never()) //and keep the stream opened forever
                                ),
                                false
                            )
                        )
                         .onErrorResumeNext(Flowable.empty())
                         .map(i -> "just one element") //here goes logic for just one emission
                    )
            )
            .doOnNext(i -> System.out.println("haya! " + i))
            .blockingSubscribe();
}

Answer 1

我怀疑我把它弄得太复杂了。我用这种方式更清洁、更简单地解决了这个问题：

 public static <U, D> FlowableTransformer<U, D> singleMultipleBranching(
    FlowableTransformer<U, D> singleBranchTransformer,
    FlowableTransformer<U, D> manyBranchTransformer
)
{
    return
        fl ->
            fl.replay( //replay so that you get connectableFlowable inside
                       f -> f.buffer(2)
                             .take(1)
                             .switchMap(
                                 buf -> {
                                     switch (buf.size()) {
                                     case 1:
                                         return f.compose(
                                             singleBranchTransformer);
                                     case 2:
                                         return f.compose(
                                             manyBranchTransformer);
                                     default:
                                         return Flowable.empty();
                                     }
                                 }
                             )
            );

}