Rx 错误拦截和分发

/**
    * 获取验证码
    */
   @POST("/sms") Observable<ServerResult<Params>> fetchSmsCode(@Body Params params);

mRetrofit = new Retrofit.Builder().baseUrl(retrofitWrapperFactory.provideBaseUrl())
        .client(retrofitWrapperFactory.createOkHttpClient())
        .addConverterFactory(retrofitWrapperFactory.createGsonConverterFactory())
        .addCallAdapterFactory(retrofitWrapperFactory.createRxJavaCallAdapter()) // 设置 CallAdapterFactory
        .build();

/**
 * Adapts a {@link Call} with response type {@code R} into the type of {@code T}. Instances are
 * created by {@linkplain Factory a factory} which is
 * {@linkplain Retrofit.Builder#addCallAdapterFactory(Factory) installed} into the {@link Retrofit}
 * instance.
 */
public interface CallAdapter<R, T> {
  /**
   * Returns the value type that this adapter uses when converting the HTTP response body to a Java
   * object. For example, the response type for {@code Call<Repo>} is {@code Repo}. This type
   * is used to prepare the {@code call} passed to {@code #adapt}.
   * <p>
   * Note: This is typically not the same type as the {@code returnType} provided to this call
   * adapter's factory.
   */
  Type responseType();

  /**
   * Returns an instance of {@code T} which delegates to {@code call}.
   * <p>
   * For example, given an instance for a hypothetical utility, {@code Async}, this instance would
   * return a new {@code Async<R>} which invoked {@code call} when run.
   * <pre><code>
   * &#64;Override
   * public &lt;R&gt; Async&lt;R&gt; adapt(final Call&lt;R&gt; call) {
   *   return Async.create(new Callable&lt;Response&lt;R&gt;&gt;() {
   *     &#64;Override
   *     public Response&lt;R&gt; call() throws Exception {
   *       return call.execute();
   *     }
   *   });
   * }
   * </code></pre>
   */
  T adapt(Call<R> call);

  /**
   * Creates {@link CallAdapter} instances based on the return type of {@linkplain
   * Retrofit#create(Class) the service interface} methods.
   */
  abstract class Factory {
    /**
     * Returns a call adapter for interface methods that return {@code returnType}, or null if it
     * cannot be handled by this factory.
     */
    public abstract CallAdapter<?, ?> get(Type returnType, Annotation[] annotations,
        Retrofit retrofit);

    /**
     * Extract the upper bound of the generic parameter at {@code index} from {@code type}. For
     * example, index 1 of {@code Map<String, ? extends Runnable>} returns {@code Runnable}.
     */
    protected static Type getParameterUpperBound(int index, ParameterizedType type) {
      return Utils.getParameterUpperBound(index, type);
    }

    /**
     * Extract the raw class type from {@code type}. For example, the type representing
     * {@code List<? extends Runnable>} returns {@code List.class}.
     */
    protected static Class<?> getRawType(Type type) {
      return Utils.getRawType(type);
    }
  }
}

private static class RxCallAdapterWrapper<R> implements CallAdapter<R, Observable<?>> {

    private final Retrofit mRetrofit;

    private final CallAdapter<R, ?> mWrappedCallAdapter;
  
  	// ...省略 N 行

public final class RxErrorHandlerCallAdapterFactory extends CallAdapter.Factory {

  private final RxJavaCallAdapterFactory mRxJavaCallAdapterFactory;

  private final RxErrorDispatcher mRxErrorDispatcher;

  private RxErrorHandlerCallAdapterFactory(RxErrorDispatcher errorDispatcher) {

    mRxJavaCallAdapterFactory = RxJavaCallAdapterFactory.create();
    mRxErrorDispatcher = errorDispatcher;
  }
  // 省略 N 行

private static class RxCallAdapterWrapper<R> implements CallAdapter<R, Observable<?>> {

    private final Retrofit mRetrofit;

    private final CallAdapter<R, ?> mWrappedCallAdapter;

    private final RxErrorDispatcher mRxErrorDispatcher;

    RxCallAdapterWrapper(Retrofit retrofit, CallAdapter<R, ?> wrappedCallAdapter,
        RxErrorDispatcher rxErrorDispatcher) {

      mRetrofit = retrofit;
      mWrappedCallAdapter = wrappedCallAdapter;
      mRxErrorDispatcher = rxErrorDispatcher;
    }

    @Override public Type responseType() {

      return mWrappedCallAdapter.responseType();
    }

  	// 真正的转换在这里
    @SuppressWarnings("unchecked") @Override public Observable<?> adapt(Call<R> call) {

      return ((Observable) mWrappedCallAdapter.adapt(call)).onErrorResumeNext(

          new Func1<Throwable, Observable>() {

            @Override public Observable call(Throwable throwable) {

              return Observable.error(mRxErrorDispatcher.dispatchError(throwable, mRetrofit));
            }
          });
    }
  }

public interface RxErrorDispatcher {

  Throwable dispatchError(Throwable throwable, Retrofit retrofit);
}

private <T> T getErrorBodyAs(Response response, Class<T> type, Retrofit retrofit)
      throws IOException {

    if (response == null || response.errorBody() == null) {

      return null;
    }

    Converter<ResponseBody, T> converter = retrofit.responseBodyConverter(type, new Annotation[0]);

    return converter.convert(response.errorBody());
  }