如何自定义我自己的 Observable？

Question

我正在使用retrofit2、rxjava2 和adapter-rxjava 来实现我的http api 调用。

//API definition
Observable<String> queryProducts(@Body Query query);

//API implementation.
serviceApi.queryProducts(query)
                .subscribeOn(new Scheduler().ioThread())
                .observeOn(new Scheduler().mainThread())
                .subscribe(new Observer());

如果我有很多api需要实现，每个单独的api实现都需要添加这两行：

.subscribeOn(new Scheduler().ioThread())
.observeOn(new Scheduler().mainThread())

我不想在每个 api 实现中添加它们。我想使用 MyObservable 作为我的 api 定义的结果类型。

我的想法如下：

//API definition
 MyObservable<String> queryProducts(@Body Query query);

//MyObservable definition
    public class MyObservable<T> extends Observable<T> {
        /**
         * Creates an Observable with a Function to execute when it is subscribed to.
         * <p>
         * <em>Note:</em> Use {@link #create(OnSubscribe)} to create an Observable, instead of this constructor,
         * unless you specifically have a need for inheritance.
         *
         * @param f {@link OnSubscribe} to be executed when {@link #subscribe(Subscriber)} is called
         */
        protected MyObservable(OnSubscribe<T> f) {
            super(f);
            this.subscribeOn(new Scheduler().ioThread());
            this.observeOn(new Scheduler().mainThread());
        }
    }

当我运行它时，我得到了以下异常：

java.lang.IllegalArgumentException：无法创建调用适配器 MyObservable。

我在以下位置跟踪了 RxJavaCallAdapterFactory.java 代码 https://github.com/square/retrofit/blob/master/retrofit-adapters/rxjava/src/main/java/retrofit2/adapter/rxjava/RxJavaCallAdapterFactory.java。我在第 100 行找到了 RxJavaCallAdapterFactory，它似乎只让 Observable 类通过这个检查点。我不能扩展和覆盖这个方法，因为这个类是一个最终类。

if (rawType != Observable.class && !isSingle && !isCompletable) {
      return null;
}

有没有办法在超类中添加这两行，我不想在每个 api 实现中添加它们？非常感谢。

Answer 1

虽然在 RxJava2 中你可以safely extend Observable，但它不太适合这种你想重用公共代码而不是复制它的情况（但从头开始创建 Observable，通常用于包装外部异步回调代码）。 相反，您可以使用 compose() 运算符，它使用您的自定义代码转换 Observable，这是向 Observable 添加通用逻辑的经典方法。
您可以关注 Dan Lew's article 以获取完全符合您需要的示例（添加 Schedulers）。

重新升级改造适配器，因为它使用反射创建服务，它不支持自定义类，但会生成现有的 RxJava 类。
顺便说一句，您正在寻找/使用带有 RxJava2 的改造的 RxJava1 适配器，您需要切换到 RxJava2 适配器，使用 RxJava2 适配器您可以看到实际上改造使用了自己的自定义 Observable 类。

如果使用 compose() 对您来说还不够（因为您仍然需要将其添加到每个 API），它的官方方式是创建您自己的 CallAdapter.Factory 并通过包装 RxJava2CallAdapterFactory 委托来实现改造 CallAdapter适应它，然后用您的自定义代码/运算符/schdeulers 包装返回 Observable。看到这个tutorial。或在library I'm working on 上使用 RxJava2（几乎相同）的示例。