java - 如何帮助 Java 解析 RxJava 组合方法中的类型？

Question

我想链接两个 RxJava Single 实例来创建一个发出它们两个结果的 Observable 。此外，我需要第一个 Single 的结果来创建第二个。

这是我尝试过的:

public static <A extends C, B extends C, C> Observable<C> chain(final Single<A> a, final Function<A, Single<B>> f) {
    return Observable.concat(
        a.toObservable(),
        a.flatMap(f::apply).toObservable());
}

用法可能如下所示:

final Observable<Event> task = MoreObservables.chain(
    writeFile("Hello, world", "hello.txt"),
    writeFileEvent -> processFile(writeFileEvent.path));

但是，Java 提示它无法解析类型:

Error:(54, 61) java: incompatible types: cannot infer type-variable(s) A,B,C
(argument mismatch; bad return type in lambda expression
  io.reactivex.Single<ProcessFileEvent> cannot be converted to io.reactivex.Single<Event>)

当然，ProcessFileEvent 实现了Event。

如何编写函数以便 Java 能够识别类型？或者有更简单的方法来实现这个吗？

Answer 1

在不知道确切的 writeFile 和 processFile 签名的情况下，很难判断为什么会出现编译错误(通过简单的模拟，它应该可以编译)。

无论如何，更惯用的方法是将 compose() 方法与自定义 ObservableTransformer 结合使用，以便拥有单个链而不是包装方法，从而减少链的数量可读(read this)。
这里还存在逻辑问题，因为您使用 concat() 并使用两次 Observable，您实际上将执行两次 a 操作(a 将被订阅两次)，这可能会在最坏的情况下导致性能问题，或者导致重大的微妙错误。 (在您的示例中，您将写入同一个文件两次)。
我认为在这种情况下你应该使用publish，为了执行一次，与merge一起，这将导致Observable发出A的结果，然后用A的结果执行B，并发出这个结果:

变压器:

class PublishAndMergeTransformer<A extends C, B extends C, C> implements ObservableTransformer<A, C> {

        final Function<A, Single<B>> f;

        public PublishAndMergeTransformer(Function<A, Single<B>> f) {
            this.f = f;
        }

        @Override
        public ObservableSource<C> apply(Observable<A> a) {
            return a.publish(aObservable ->
                    Observable.merge(
                            aObservable,
                            aObservable
                                    .flatMap(a1 -> f.apply(a1).toObservable())
                    )
            );
        }
    }

以及使用示例:

writeFile("Hello, world", "hello.txt")
    .toObservable()
    .compose(new PublishAndMergeTransformer<>(writeFileEvent -> processFile(writeFileEvent.path)));