如何不在 RxJava 中重复相同的操作？

Question

我有如下代码：

authRepository.login(userName, password)
              .doOnSubscribe(__ -> apiProgress.setValue(ApiProgress.start()))
              .doFinally(() -> apiProgress.setValue(ApiProgress.stop()))
              .subscribe(login -> loginData.setValue(login), 
                           err -> apiError.setValue(ApiError.create(err)))

我需要对所有 api 调用重复 doOnSubscribe(..) 和 doFinally。

有没有办法实现这个东西？

Answer 1

欢迎来到 StackOverflow！ https://stackoverflow.com/conduct

您可以使用Transformer (http://reactivex.io/RxJava/javadoc/rx/Single.Transformer.html) 创建类似的内容

static <T> SingleTransformer<T, T> subscribeAndFinalTransformer() {
        return new SingleTransformer<T, T>() {
            @Override
            public SingleSource<T> apply(Single<T> upstream) {
                return upstream.doOnSubscribe(disposable -> {
                    // Your doOnSubscribe Block
                }).doFinally(() -> {
                    // Your doFinally Block 
                });
            }
        };
    }

及以上可重复使用的Transformer 可以使用compose 方法为所有Single 附加。

authRepository.login(userName, password).compose(subscribeAndFinalTransformer())
.subscribe()

authRepository.anotherApi().compose(subscribeAndFinalTransformer()).subscribe()

如果您使用Observable 或Completable，则应使用等效的Transformer 而不是SingleTransformer

编辑：

如果您只想对某些调用重复使用某些操作，则上述方法很方便。

如果您想将操作附加到所有 API 调用，您可以创建 Retrofit CallAdapter

class RxStreamAdapter implements CallAdapter {

    private final Class rawType;
    private final CallAdapter<Object, Object> nextAdapter;
    private final Type returnType;

    RxStreamAdapter(Class rawType,
                    Type returnType,
                    CallAdapter nextAdapter) {
        this.rawType = rawType;
        this.returnType = returnType;
        this.nextAdapter = nextAdapter;
    }

    @Override
    public Type responseType() {
        return nextAdapter.responseType();
    }

    @Override
    public Object adapt(Call call) {
        if (rawType == Single.class) {
            return ((Single) nextAdapter.adapt(call))
                    .doOnSubscribe(getDoOnSubscribe())
                    .doFinally(getDoFinally());
        } else if (returnType == Completable.class) {
            return ((Completable) nextAdapter.adapt(call))
                    .doOnSubscribe(getDoOnSubscribe())
                    .doFinally(getDoFinally());
        } else {
            // Observable
            return ((Observable<Object>) nextAdapter.adapt(call))
                    .doOnSubscribe(getDoOnSubscribe())
                    .doFinally(getDoFinally());
        }
    }

    @NotNull
    private Consumer<Disposable> getDoOnSubscribe() {
        return disposable -> {

        };
    }

    @NotNull
    private Action getDoFinally() {
        return () -> {

        };
    }
}

然后在创建Retrofit Object时添加（RxJava2CallAdapterFactory之前）

RetrofitApi retrofitApi = new Retrofit
                .Builder()
                .addCallAdapterFactory(new CallAdapter.Factory() {
                    @Override
                    public CallAdapter<?, ?> get(Type returnType, Annotation[] annotations, Retrofit retrofit) {
                        CallAdapter<?, ?> nextAdapter = retrofit.nextCallAdapter(this, returnType, annotations);
                        Class<?> rawType = getRawType(returnType);
                        if (rawType == Single.class || rawType == Observable.class || rawType == Completable.class) {
                            return new RxStreamAdapter(getRawType(returnType), returnType, nextAdapter);
                        } else {
                            return nextAdapter;
                        }
                    }
                })
                .addCallAdapterFactory(RxJava2CallAdapterFactory.create())
                .build()

您还可以使用RxJavaPlugins 设置挂钩。但是你无法区分黑白正常流和改造流。

希望对你有帮助！