谜之RxJava(四)—— Retrofit和RxJava的基情

概述

前文回顾: 谜之RxJava (三)—— 线程切换java

今天来介绍下和RxJava搭配使用的好基友,就是咱们的Retrofit啦,Retrofit使用动态代理的机制,为咱们提供了一个简要的使用方法来获取网络上的资料,如今更新到2.0.0-beta2了(由于是beta,我也碰到一些坑,期待做者发布下一个版本)。segmentfault

Retrofit不算是一个网络库,它应该算是封装了okhttp,而后为咱们提供了一个友好的接口的一个工具库吧。网络

咱们今天着重来介绍下RxJavaCallAdapterFactory这个类。用过的朋友们都知道,它是用来把Retrofit转成RxJava可用的适配类。app

RxJavaCallAdapterFactory

OK,首先看下声明ide

public final class RxJavaCallAdapterFactory implements CallAdapter.Factory

CallAdapter.FactoryRetrofit这个库中的接口,用来给咱们自定义去解析咱们本身想要的类型用的。工具

举个栗子:this

@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接口干的事。code

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();
      }
    }
}

这里其实蛮简单的,callretrofitokhttp的一个代理,是一个同步网络请求,在这里就是典型的对网络进行数据请求,完了放到subscriberonNext里,完成网络请求。咱们能够看下,它把unsubscribe,也就是取消请求的状况处理的挺好。

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

这段代码是给subscribe增长一个unsubscribe的事件。 也就是请求完成的时候,会自动对call进行一个终止,也就是httpclose行为。

前方高能

今天被坑到这里好久,咱们对API调用了observeOn(MainThread)以后,线程会跑在主线程上,包括onComplete也是,unsubscribe也在主线程,而后若是这时候调用call.cancel会致使NetworkOnMainThreadException,因此必定要在retrofit的API调用ExampleAPI.subscribeOn(io).observeOn(MainThread)以后加一句unsubscribeOn(io)

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

相关文章
相关标签/搜索