我如何自定义自己的可观察对象？

Question

我正在使用reverfit2、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());
        }
    }

我在https://github.com/square/reverfit/blob/master/reverfit-adapters/rxjava/src/main/java/reverfit2/adapter/rxjava/rxjavaalladapterfactory.java上跟踪了rxjavacalladapterfactory.java代码。我在第100行发现了RxJavaCallAdapterFactory,它似乎只允许Observable类通过这个检查点。我无法扩展和重写这个方法，因为这个类是final类。

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

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

Answer 1

虽然在RxJava2中您可以安全地扩展Observable，但它不太可能适合这种情况，即您希望重用公共代码而不是复制公共代码（而是从头开始创建Observable，通常用于包装外部异步回调代码）。
相反，您可以使用compose()运算符来转换自定义代码中的Observable，这是将公共逻辑添加到Observable中的经典方法。
您可以根据Dan Lew的文章提供一个完全符合您需要的示例（添加调度器)。

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

如果使用compose()对您来说还不够（因为您仍然需要将它添加到每个API中），那么正式的方法是创建您自己的CallAdapter.Factory并通过包装RXJava2CallAdapterFactory委托Adapting，实现改版CallAdapter，然后用您的自定义代码/Operators/Schdeuler包装返回的Observable。请参阅本教程。或者在我正在处理的某个库中使用RxJava2的示例（相当相同）。