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

來自： http://segmentfault.com/a/1190000004077117

概述

前文回顧： 迷之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;

}</pre> 
這里代碼的意思就是說，如果你返回的不是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);

}</pre> 
這里告訴我們，除了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));
        }
      });
}

}</pre> 
這里總算看見我們熟悉的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();
  }
}

}</pre> 
這里其實蠻簡單的，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)。