謎之RxJava（四）—— Retrofit和RxJava的基情

概述

前文回顧： 謎之RxJava （三）—— 線程切換

今天來介紹下和RxJava搭配使用的好基友，就是咱們的Retrofit啦，Retrofit使用動態代理的機制，爲咱們提供了一個簡要的使用方法來獲取網絡上的資料，如今更新到2.0.0-beta2了（由於是beta，我也碰到一些坑，期待做者發佈下一個版本）。

Retrofit不算是一個網絡庫，它應該算是封裝了okhttp，而後爲咱們提供了一個友好的接口的一個工具庫吧。

咱們今天着重來介紹下RxJavaCallAdapterFactory這個類。用過的朋友們都知道，它是用來把Retrofit轉成RxJava可用的適配類。

RxJavaCallAdapterFactory

OK，首先看下聲明

public final class RxJavaCallAdapterFactory implements CallAdapter.Factory

CallAdapter.Factory是Retrofit這個庫中的接口，用來給咱們自定義去解析咱們本身想要的類型用的。

舉個栗子：

@GET("/aaa")
Observable<QuestionListData> getQuestionNewestList();

這麼一個接口，retrofit自己是沒法識別Observable<QuestionListData>而後去工做的，若是沒有這個適配器就根本沒法工做，所以咱們的適配器的做用，就是生成咱們想要的Observable。

看下它的實現。

@Override
public CallAdapter<?> get(Type returnType, Annotation[] annotations, Retrofit retrofit) {
    Class<?> rawType = Utils.getRawType(returnType);
    boolean isSingle = "rx.Single".equals(rawType.getCanonicalName());
    if (rawType != Observable.class && !isSingle) {
      return null;
    }
    if (!(returnType instanceof ParameterizedType)) {
      String name = isSingle ? "Single" : "Observable";
      throw new IllegalStateException(name + " return type must be parameterized"
          + " as " + name + "<Foo> or " + name + "<? extends Foo>");
    }
    
    CallAdapter<Observable<?>> callAdapter = getCallAdapter(returnType);
    if (isSingle) {
      // Add Single-converter wrapper from a separate class. This defers classloading such that
      // regular Observable operation can be leveraged without relying on this unstable RxJava API.
      return SingleHelper.makeSingle(callAdapter);
    }
    return callAdapter;
}

這裏代碼的意思就是說，若是你返回的不是Observable<T>這種類型，我就不幹！
若是是的話，那我再來詳細看下模板類是哪一個，也就是getCallAdapter接口乾的事。

private CallAdapter<Observable<?>> getCallAdapter(Type returnType) {
    Type observableType = Utils.getSingleParameterUpperBound((ParameterizedType) returnType);
    Class<?> rawObservableType = Utils.getRawType(observableType);
    if (rawObservableType == Response.class) {
      if (!(observableType instanceof ParameterizedType)) {
        throw new IllegalStateException("Response must be parameterized"
            + " as Response<Foo> or Response<? extends Foo>");
      }
      Type responseType = Utils.getSingleParameterUpperBound((ParameterizedType) observableType);
      return new ResponseCallAdapter(responseType);
    }

    if (rawObservableType == Result.class) {
      if (!(observableType instanceof ParameterizedType)) {
        throw new IllegalStateException("Result must be parameterized"
            + " as Result<Foo> or Result<? extends Foo>");
      }
      Type responseType = Utils.getSingleParameterUpperBound((ParameterizedType) observableType);
      return new ResultCallAdapter(responseType);
    }

    return new SimpleCallAdapter(observableType);
}

這裏告訴咱們，除了Observable<Response>和Observable<Result>須要不一樣的Adapter處理外，其餘的都讓SimpleCallAdapter處理。

OK，咱們就不看別的，直搗黃龍，看SimpleCallAdapter！

SimpleCallAdapter 建立Observable的類

static final class SimpleCallAdapter implements CallAdapter<Observable<?>> {
    private final Type responseType;

    SimpleCallAdapter(Type responseType) {
      this.responseType = responseType;
    }

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

    @Override public <R> Observable<R> adapt(Call<R> call) {
      return Observable.create(new CallOnSubscribe<>(call)) //
          .flatMap(new Func1<Response<R>, Observable<R>>() {
            @Override public Observable<R> call(Response<R> response) {
              if (response.isSuccess()) {
                return Observable.just(response.body());
              }
              return Observable.error(new HttpException(response));
            }
          });
    }
  }

這裏總算看見咱們熟悉的Observable接口咯，原來是本身定義了一個OnSubscribe，而後把Response經過flatMap轉爲咱們想要的對象了。 這裏同時也判斷請求是否成功，進入Observable的工做流裏了。

好，咱們最終能夠看下CallOnSubscribe幹了啥

static final class CallOnSubscribe<T> implements Observable.OnSubscribe<Response<T>> {
    private final Call<T> originalCall;

    private CallOnSubscribe(Call<T> originalCall) {
      this.originalCall = originalCall;
    }

    @Override public void call(final Subscriber<? super Response<T>> subscriber) {
      // Since Call is a one-shot type, clone it for each new subscriber.
      final Call<T> call = originalCall.clone();

      // Attempt to cancel the call if it is still in-flight on unsubscription.
      subscriber.add(Subscriptions.create(new Action0() {
        @Override public void call() {
          call.cancel();
        }
      }));

      if (subscriber.isUnsubscribed()) {
        return;
      }

      try {
        Response<T> response = call.execute();
        if (!subscriber.isUnsubscribed()) {
          subscriber.onNext(response);
        }
      } catch (Throwable t) {
        Exceptions.throwIfFatal(t);
        if (!subscriber.isUnsubscribed()) {
          subscriber.onError(t);
        }
        return;
      }

      if (!subscriber.isUnsubscribed()) {
        subscriber.onCompleted();
      }
    }
}

這裏其實蠻簡單的，call是retrofit對okhttp的一個代理，是一個同步網絡請求，在這裏就是典型的對網絡進行數據請求，完了放到subscriber的onNext裏，完成網絡請求。咱們能夠看下，它把unsubscribe，也就是取消請求的狀況處理的挺好。

subscriber.add(Subscriptions.create(new Action0() {
    @Override public void call() {
      call.cancel();
    }
}));

這段代碼是給subscribe增長一個unsubscribe的事件。 也就是請求完成的時候，會自動對call進行一個終止，也就是http的close行爲。

前方高能

今天被坑到這裏好久，咱們對API調用了observeOn(MainThread)以後，線程會跑在主線程上，包括onComplete也是，unsubscribe也在主線程，而後若是這時候調用call.cancel會致使NetworkOnMainThreadException，因此必定要在retrofit的API調用ExampleAPI.subscribeOn(io).observeOn(MainThread)以後加一句unsubscribeOn(io)。

完整的就是ExampleAPI.subscribeOn(io).observeOn(MainThread).unsubscribeOn(io)。