默認(rèn)添加異步調(diào)度

?源碼分析

final class RxJava2CallAdapter<R> implements CallAdapter<R, Object> {
 
  @Override public Object adapt(Call<R> call) {
    Observable<Response<R>> responseObservable = isAsync
        ? new CallEnqueueObservable<>(call)
        : new CallExecuteObservable<>(call);
    Observable<?> observable;
    if (isResult) {
      observable = new ResultObservable<>(responseObservable);
    } else if (isBody) {
      observable = new BodyObservable<>(responseObservable);
    } else {
      observable = responseObservable;
    }
    if (scheduler != null) {
      observable = observable.subscribeOn(scheduler);
    }
    if (isFlowable) {
      return observable.toFlowable(BackpressureStrategy.LATEST);
    }
     //省略...
    return observable;
  }
}

自定義CallAdapter.Factory實現(xiàn)委托RxJava2CallAdapterFactory對象實現(xiàn)添加異步調(diào)度:?

public final class ArnoldRxJava2CallAdapterFactory extends CallAdapter.Factory {

    private CallAdapter.Factory mFactory;

    public static ArnoldRxJava2CallAdapterFactory create() {
        return new ArnoldRxJava2CallAdapterFactory();
    }
    private ArnoldRxJava2CallAdapterFactory() {
        mFactory = RxJava2CallAdapterFactory.create();
    }
    @Override
    public CallAdapter<?, ?> get(@NonNull Type returnType, @NonNull Annotation[] annotations, @NonNull Retrofit retrofit) {
        final CallAdapter<?, ?> callAdapter = mFactory.get(returnType, annotations, retrofit);

        Class<?> rawType = getRawType(returnType);

        boolean isFlowable = rawType == Flowable.class;
        boolean isObservable = rawType == Observable.class;

        if (null == callAdapter) {
            return null;
        }
        //生產(chǎn)不同的CallAdapter
        if (isObservable) {
            return CallAdapterFactory.create((CallAdapter<Observable<?>, Observable<?>>) callAdapter, f -> f.subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()));
        }
        if (isFlowable) {
            return CallAdapterFactory.create((CallAdapter<Flowable<?>, Flowable<?>>) callAdapter, f -> f.subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()));
        }
        //省略...
        return callAdapter;
    }
}

創(chuàng)建CallAdapterFactory工廠類

public class CallAdapterFactory<R, W> implements CallAdapter<R, W> {

    public static <R, W> CallAdapterFactory<R, W> create(CallAdapter<R, W> mAdapter, Function<W, W> mFunction) {
        return new CallAdapterFactory<R, W>(mAdapter, mFunction);
    }
    private CallAdapterFactory(CallAdapter<R, W> mAdapter, Function<W, W> mFunction) {
        this.mAdapter = mAdapter;
        this.mFunction = mFunction;
    }
    private CallAdapter<R, W> mAdapter;

    private Function<W, W> mFunction;

    @Override
    public Type responseType() {
        return mAdapter.responseType();
    }
    @Override
    public W adapt(Call<R> call) {
        W adapt = mAdapter.adapt(call);
        if (mFunction == null) {
            return adapt;
        }
        try {
            adapt = mFunction.apply(mAdapter.adapt(call));
        } catch (Exception e) {
            e.printStackTrace();
        }
        return adapt;
    }
}

調(diào)用示例

        T service= new Retrofit.Builder()
                .client(OkHttpClient client)
                .addConverterFactory(ArnoldGsonConverterFactory.create())
                .addCallAdapterFactory(ArnoldRxJava2CallAdapterFactory .create())
                .baseUrl(BASE_HTTP_URL)
                .build()
                .create(Class<T> service);

在創(chuàng)建Retrofit時添加自定義RxJava2CallAdapterFactory，使用時不再需要添加異步調(diào)度代碼：?sevice.XXX().subscribe();