RXJAVA2分支逻辑基于排放次数

Question

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

确切地说，我想要：

1. 上游是空的
2. 当上游发出一个值时，一个分支要触发，然后完成
3. 当上游发出多个值然后完成时，一个分支要触发。

我在如何解决这个问题的情况下挠头，我想出了一些有效但似乎冗长的东西。我想知道是否有更简单的方法。

此解决方案基于rxjava2extensions项目的阀门操作员。

解决方案的轮廓如下：

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

所以这似乎有效，尽管我的担忧是：

1. 它看了为自己的工作而设计的。可以将其编码更简单，Clener？
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();
                                     }
                                 }
                             )
            );
}