What？
它是Square公司开发的一个强大的网络库！基于Okhttp，支持RxJava。
Why?
功能强大
1.基于Okhttp&遵循Restful API设计风格
2.通过注解配置网络请求参数
4.支持同步&异步网络请求
5.支持多种数据解析&序列化格式（Gson,Json,XML，Protobuf）
6.提供RxJava支持优点
功能强大，简洁易用，可扩展性好
HOW?
说了这么多用个试试！
public interface ApiService {
@GET("/")
Call<ResponseBody> getRetrofitTest();
}
public void retrofit(){
Retrofit retrofit =new Retrofit.Builder().baseUrl("https://www.baidu.com").build();
ApiService apiService=retrofit.create(ApiService.class);
apiService.getRetrofitTest().enqueue(new Callback<ResponseBody>() {
@Override
public void onResponse(Call<ResponseBody> call, Response<ResponseBody> response) {
Log.i("retrofit",response.message());
}
@Override
public void onFailure(Call<ResponseBody> call, Throwable t) {
Log.i("retrofit",t.toString());
}
});
}
Code分析
基本流程
1.使用建造者模式实例化Retrofit
Retrofit retrofit = new Retrofit.Builder().baseUrl("https://www.baidu.com").addConverterFactory
(GsonConverterFactory.create()).build();
2.使用注解方式构建Call请求方法| 参数...
@GET("/")
Call<ResponseBody> getRetrofitTest();
3.使用动态代理实现请求接口
ApiService apiService = retrofit.create(ApiService.class);
4.异步请求方法调用，回调返回结果
apiService.getRetrofitTest().enqueue(new Callback<ResponseBody>() {
@Override
public void onResponse(Call<ResponseBody> call, Response<ResponseBody> response) {
Log.i("retrofit", response.message());
}
@Override
public void onFailure(Call<ResponseBody> call, Throwable t) {
Log.i("retrofit", t.toString());
}
});
流程详解
流程1
Builder模式实例化retrofit三步骤
Retrofit retrofit = new Retrofit.Builder().baseUrl("https://www.baidu.com").addConverterFactory
(GsonConverterFactory.create()).build();
步骤 1
这个函数做了什么呢？
1.对变量是否为空判断，并附上默认的值
2.然后new 出retrofit
public Retrofit build() {
if (baseUrl == null) {
throw new IllegalStateException("Base URL required.");
}
okhttp3.Call.Factory callFactory = this.callFactory;
if (callFactory == null) {
callFactory = new OkHttpClient();
}
Executor callbackExecutor = this.callbackExecutor;
if (callbackExecutor == null) {
callbackExecutor = platform.defaultCallbackExecutor();
}
// Make a defensive copy of the adapters and add the default Call adapter.
List<CallAdapter.Factory> callAdapterFactories = new ArrayList<>(this.callAdapterFactories);
callAdapterFactories.addAll(platform.defaultCallAdapterFactories(callbackExecutor));
// Make a defensive copy of the converters.
List<Converter.Factory> converterFactories = new ArrayList<>(
1 + this.converterFactories.size() + platform.defaultConverterFactoriesSize());
// Add the built-in converter factory first. This prevents overriding its behavior but also
// ensures correct behavior when using converters that consume all types.
converterFactories.add(new BuiltInConverters());
converterFactories.addAll(this.converterFactories);
converterFactories.addAll(platform.defaultConverterFactories());
return new Retrofit(callFactory, baseUrl, unmodifiableList(converterFactories),
unmodifiableList(callAdapterFactories), callbackExecutor, validateEagerly);
}
}
步骤2 设置配置参数
这里只是举两个配置参数的例子
public Builder baseUrl(HttpUrl baseUrl) {
checkNotNull(baseUrl, "baseUrl == null");
List<String> pathSegments = baseUrl.pathSegments();
if (!"".equals(pathSegments.get(pathSegments.size() - 1))) {
throw new IllegalArgumentException("baseUrl must end in /: " + baseUrl);
}
this.baseUrl = baseUrl;
return this;
}
/** Add converter factory for serialization and deserialization of objects. */
public Builder addConverterFactory(Converter.Factory factory) {
converterFactories.add(checkNotNull(factory, "factory == null"));
return this;
}
步骤3 分析Retrofit常量&方法
public final class Retrofit {
private final Map<Method, ServiceMethod<?>> serviceMethodCache = new ConcurrentHashMap<>();
final okhttp3.Call.Factory callFactory;
final HttpUrl baseUrl;//请求URL
final List<Converter.Factory> converterFactories;//数据转换器集合
final List<CallAdapter.Factory> callAdapterFactories;//网络请求适配器集合
final @Nullable Executor callbackExecutor;//线程切换执行器
final boolean validateEagerly;//验证
Retrofit(okhttp3.Call.Factory callFactory, HttpUrl baseUrl,
List<Converter.Factory> converterFactories, List<CallAdapter.Factory> callAdapterFactories,
@Nullable Executor callbackExecutor, boolean validateEagerly) {
this.callFactory = callFactory;
this.baseUrl = baseUrl;
this.converterFactories = converterFactories; // Copy+unmodifiable at call site.
this.callAdapterFactories = callAdapterFactories; // Copy+unmodifiable at call site.
this.callbackExecutor = callbackExecutor;
this.validateEagerly = validateEagerly;
}
变量详解
1.HttpUrl baseUrl
/**
* Set the API base URL.
*
* @see #baseUrl(HttpUrl)
*把String类型的url参数转化为适合OKhttp的HttpUrl类型
*/
public Builder baseUrl(String baseUrl) {
checkNotNull(baseUrl, "baseUrl == null");
return baseUrl(HttpUrl.get(baseUrl));
}
/**
* Set the API base URL.
*
* @see #baseUrl(HttpUrl)
*/
public Builder baseUrl(URL baseUrl) {
checkNotNull(baseUrl, "baseUrl == null");
return baseUrl(HttpUrl.get(baseUrl.toString()));
}
/**
* Set the API base URL.
*
* @see #baseUrl(URL)
*/
public Builder baseUrl(HttpUrl baseUrl) {
checkNotNull(baseUrl, "baseUrl == null");//检测是否空
List<String> pathSegments = baseUrl.pathSegments();//切割判断
//判断最后一个字段要以 / 结尾
if (!"".equals(pathSegments.get(pathSegments.size() - 1))) {
throw new IllegalArgumentException("baseUrl must end in /: " + baseUrl);
}
this.baseUrl = baseUrl;
return this;
}
2. final List<Converter.Factory> converterFactories;
数据转换器工厂的集合
作用：放置数据转换器工厂
数据转换器工厂作用：生产数据转换器（converter）
/** Add converter factory for serialization and deserialization of objects.
*添加转换器工厂以进行对象的序列化和反序列化。
* */
public Builder addConverterFactory(Converter.Factory factory) {
converterFactories.add(checkNotNull(factory, "factory == null"));
return this;
}
所有定义只要实现Converter这个接口,数据F转换T
public interface Converter<F, T> {
//实现从 F(rom) 到 T(o)的转换
@Nullable T convert(F value) throws IOException;
/** Creates {@link Converter} instances based on a type and target usage. */
abstract class Factory {
/**
* 这里创建从ResponseBody其它类型的Converter，如果不能处理返回null
* 主要用于对响应体的处理
*/
public @Nullable Converter<ResponseBody, ?> responseBodyConverter(Type type,
Annotation[] annotations, Retrofit retrofit) {
return null;
}
/**
在这里创建 从自定类型到RequestBody 的Converter,不能处理就返回null，
主要用于对Part、PartMap、Body注解的处理
*/
public @Nullable Converter<?, RequestBody> requestBodyConverter(Type type,
Annotation[] parameterAnnotations, Annotation[] methodAnnotations, Retrofit retrofit) {
return null;
}
/**
这里用于对Field、FieldMap、Header、Path、Query、QueryMap注解的处理
Retrfofit对于上面的几个注解默认使用的是调用toString方法
*/
public @Nullable Converter<?, String> stringConverter(Type type, Annotation[] annotations,
Retrofit retrofit) {
return null;
}
/**
* 根据下标index,提取泛型参数
*/
protected static Type getParameterUpperBound(int index, ParameterizedType type) {
return Utils.getParameterUpperBound(index, type);
}
/**
* 根据泛型提取类
*/
protected static Class<?> getRawType(Type type) {
return Utils.getRawType(type);
}
}
}
3. final List<CallAdapter.Factory> callAdapterFactories;
网络请求适配器工厂的集合
作用：放置网络请求适配器工厂
网络请求适配器工厂作用：生产网络请求适配器（CallAdapter）
Call在Retrofit里默认是OkHttpCall
/**
* 网络请求适配器
* @param <T>
*/
final class OkHttpCall<T> implements Call<T> {
private final RequestFactory requestFactory;
private final Object[] args;
private final okhttp3.Call.Factory callFactory;
private final Converter<ResponseBody, T> responseConverter;
private volatile boolean canceled;
@GuardedBy("this")
private @Nullable okhttp3.Call rawCall;
@GuardedBy("this") // Either a RuntimeException, non-fatal Error, or IOException.
private @Nullable Throwable creationFailure;
@GuardedBy("this")
private boolean executed;
// 构造方法
OkHttpCall(RequestFactory requestFactory, Object[] args,
okhttp3.Call.Factory callFactory, Converter<ResponseBody, T> responseConverter) {
this.requestFactory = requestFactory;
this.args = args;
this.callFactory = callFactory;
this.responseConverter = responseConverter;
}
@SuppressWarnings("CloneDoesntCallSuperClone") // We are a final type & this saves clearing state.
@Override public retrofit2.OkHttpCall<T> clone() {
return new retrofit2.OkHttpCall<>(requestFactory, args, callFactory, responseConverter);
}
@Override public synchronized Request request() {
okhttp3.Call call = rawCall;
if (call != null) {
return call.request();
}
if (creationFailure != null) {
if (creationFailure instanceof IOException) {
throw new RuntimeException("Unable to create request.", creationFailure);
} else if (creationFailure instanceof RuntimeException) {
throw (RuntimeException) creationFailure;
} else {
throw (Error) creationFailure;
}
}
try {
return (rawCall = createRawCall()).request();
} catch (RuntimeException | Error e) {
throwIfFatal(e); // Do not assign a fatal error to creationFailure.
creationFailure = e;
throw e;
} catch (IOException e) {
creationFailure = e;
throw new RuntimeException("Unable to create request.", e);
}
}
//异步执行网络请求
@Override public void enqueue(final Callback<T> callback) {
checkNotNull(callback, "callback == null");
okhttp3.Call call;
Throwable failure;
synchronized (this) {
if (executed) throw new IllegalStateException("Already executed.");
executed = true;
call = rawCall;
failure = creationFailure;
if (call == null && failure == null) {
try {
//获取真正的网络请求对象
call = rawCall = createRawCall();
} catch (Throwable t) {
throwIfFatal(t);
failure = creationFailure = t;
}
}
}
if (failure != null) {
callback.onFailure(this, failure);
return;
}
if (canceled) {
call.cancel();
}
call.enqueue(new okhttp3.Callback() {
@Override public void onResponse(okhttp3.Call call, okhttp3.Response rawResponse) {
Response<T> response;
try {
response = parseResponse(rawResponse);
} catch (Throwable e) {
throwIfFatal(e);
callFailure(e);
return;
}
try {
callback.onResponse(retrofit2.OkHttpCall.this, response);
} catch (Throwable t) {
throwIfFatal(t);
t.printStackTrace(); // TODO this is not great
}
}
@Override public void onFailure(okhttp3.Call call, IOException e) {
callFailure(e);
}
private void callFailure(Throwable e) {
try {
callback.onFailure(retrofit2.OkHttpCall.this, e);
} catch (Throwable t) {
throwIfFatal(t);
t.printStackTrace(); // TODO this is not great
}
}
});
}
@Override public synchronized boolean isExecuted() {
return executed;
}
//同步执行网络请求方法
@Override public Response<T> execute() throws IOException {
okhttp3.Call call;
synchronized (this) {
if (executed) throw new IllegalStateException("Already executed.");
executed = true;
if (creationFailure != null) {
if (creationFailure instanceof IOException) {
throw (IOException) creationFailure;
} else if (creationFailure instanceof RuntimeException) {
throw (RuntimeException) creationFailure;
} else {
throw (Error) creationFailure;
}
}
call = rawCall;
if (call == null) {
try {
call = rawCall = createRawCall();
} catch (IOException | RuntimeException | Error e) {
throwIfFatal(e); // Do not assign a fatal error to creationFailure.
creationFailure = e;
throw e;
}
}
}
if (canceled) {
call.cancel();
}
return parseResponse(call.execute());
}
private okhttp3.Call createRawCall() throws IOException {
okhttp3.Call call = callFactory.newCall(requestFactory.create(args));
if (call == null) {
throw new NullPointerException("Call.Factory returned null.");
}
return call;
}
//解析网络返回数据
Response<T> parseResponse(okhttp3.Response rawResponse) throws IOException {
ResponseBody rawBody = rawResponse.body();
// Remove the body's source (the only stateful object) so we can pass the response along.
rawResponse = rawResponse.newBuilder()
.body(new retrofit2.OkHttpCall.NoContentResponseBody(rawBody.contentType(), rawBody.contentLength()))
.build();
int code = rawResponse.code();
if (code < 200 || code >= 300) {
try {
// Buffer the entire body to avoid future I/O.
ResponseBody bufferedBody = Utils.buffer(rawBody);
return Response.error(bufferedBody, rawResponse);
} finally {
rawBody.close();
}
}
if (code == 204 || code == 205) {
rawBody.close();
return Response.success(null, rawResponse);
}
retrofit2.OkHttpCall.ExceptionCatchingResponseBody catchingBody = new retrofit2.OkHttpCall.ExceptionCatchingResponseBody(rawBody);
try {
//格式转换
T body = responseConverter.convert(catchingBody);
return Response.success(body, rawResponse);
} catch (RuntimeException e) {
// If the underlying source threw an exception, propagate that rather than indicating it was
// a runtime exception.
catchingBody.throwIfCaught();
throw e;
}
}
public void cancel() {
canceled = true;
okhttp3.Call call;
synchronized (this) {
call = rawCall;
}
if (call != null) {
call.cancel();
}
}
@Override public boolean isCanceled() {
if (canceled) {
return true;
}
synchronized (this) {
return rawCall != null && rawCall.isCanceled();
}
}
static final class NoContentResponseBody extends ResponseBody {
private final @Nullable
MediaType contentType;
private final long contentLength;
NoContentResponseBody(@Nullable MediaType contentType, long contentLength) {
this.contentType = contentType;
this.contentLength = contentLength;
}
@Override public MediaType contentType() {
return contentType;
}
@Override public long contentLength() {
return contentLength;
}
@Override public BufferedSource source() {
throw new IllegalStateException("Cannot read raw response body of a converted body.");
}
}
static final class ExceptionCatchingResponseBody extends ResponseBody {
private final ResponseBody delegate;
private final BufferedSource delegateSource;
@Nullable IOException thrownException;
ExceptionCatchingResponseBody(ResponseBody delegate) {
this.delegate = delegate;
this.delegateSource = Okio.buffer(new ForwardingSource(delegate.source()) {
@Override public long read(Buffer sink, long byteCount) throws IOException {
try {
return super.read(sink, byteCount);
} catch (IOException e) {
thrownException = e;
throw e;
}
}
});
}
@Override public MediaType contentType() {
return delegate.contentType();
}
@Override public long contentLength() {
return delegate.contentLength();
}
@Override public BufferedSource source() {
return delegateSource;
}
@Override public void close() {
delegate.close();
}
void throwIfCaught() throws IOException {
if (thrownException != null) {
throw thrownException;
}
}
}
}
在Retrofit中提供了四种CallAdapterFactory： ExecutorCallAdapterFactory（默认）、GuavaCallAdapterFactory、Java8CallAdapterFactory、RxJavaCallAdapterFactory
4. private @Nullable Executor callbackExecutor;线程切换
@Override public Executor defaultCallbackExecutor() {
return new MainThreadExecutor();
}
static class MainThreadExecutor implements Executor {
private final Handler handler = new Handler(Looper.getMainLooper());
//切换到主线程显示结果
@Override public void execute(Runnable r) {
handler.post(r);
}
}
@Override public void enqueue(final Callback<T> callback) {
checkNotNull(callback, "callback == null");
delegate.enqueue(new Callback<T>() {
@Override public void onResponse(Call<T> call, final Response<T> response) {
//线程切换
callbackExecutor.execute(new Runnable() {
@Override public void run() {
if (delegate.isCanceled()) {
// Emulate OkHttp's behavior of throwing/delivering an IOException on cancellation.
callback.onFailure(ExecutorCallbackCall.this, new IOException("Canceled"));
} else {
callback.onResponse(ExecutorCallbackCall.this, response);
}
}
});
}
@Override public void onFailure(Call<T> call, final Throwable t) {
callbackExecutor.execute(new Runnable() {
@Override public void run() {
callback.onFailure(ExecutorCallbackCall.this, t);
}
});
}
});
}
5. final boolean validateEagerly;判断是否需要验证
if (validateEagerly) {
eagerlyValidateMethods(service);
}
6.private final Map<Method, ServiceMethod<?>> serviceMethodCache = new ConcurrentHashMap<>();
网络请求配置对象（对网络请求接口中方法注解进行解析后得到的对象）
作用：存储网络请求相关的配置，如网络请求的方法、数据转换器、网络请求适配器、网络请求//工厂、基地址等
/**
*通过注解方式获取配置对象 实现缓存
*/
ServiceMethod<?> loadServiceMethod(Method method) {
ServiceMethod<?> result = serviceMethodCache.get(method);
if (result != null) return result;
synchronized (serviceMethodCache) {
result = serviceMethodCache.get(method);
if (result == null) {
result = ServiceMethod.parseAnnotations(this, method);
serviceMethodCache.put(method, result);
}
}
return result;
}
private void parseMethodAnnotation(Annotation annotation) {
if (annotation instanceof DELETE) {
parseHttpMethodAndPath("DELETE", ((DELETE) annotation).value(), false);
} else if (annotation instanceof GET) {
parseHttpMethodAndPath("GET", ((GET) annotation).value(), false);
} else if (annotation instanceof HEAD) {
parseHttpMethodAndPath("HEAD", ((HEAD) annotation).value(), false);
} else if (annotation instanceof PATCH) {
parseHttpMethodAndPath("PATCH", ((PATCH) annotation).value(), true);
} else if (annotation instanceof POST) {
parseHttpMethodAndPath("POST", ((POST) annotation).value(), true);
} else if (annotation instanceof PUT) {
parseHttpMethodAndPath("PUT", ((PUT) annotation).value(), true);
} else if (annotation instanceof OPTIONS) {
parseHttpMethodAndPath("OPTIONS", ((OPTIONS) annotation).value(), false);
} else if (annotation instanceof HTTP) {
HTTP http = (HTTP) annotation;
parseHttpMethodAndPath(http.method(), http.path(), http.hasBody());
} else if (annotation instanceof retrofit2.http.Headers) {
String[] headersToParse = ((retrofit2.http.Headers) annotation).value();
if (headersToParse.length == 0) {
throw methodError(method, "@Headers annotation is empty.");
}
headers = parseHeaders(headersToParse);
} else if (annotation instanceof Multipart) {
if (isFormEncoded) {
throw methodError(method, "Only one encoding annotation is allowed.");
}
isMultipart = true;
} else if (annotation instanceof FormUrlEncoded) {
if (isMultipart) {
throw methodError(method, "Only one encoding annotation is allowed.");
}
isFormEncoded = true;
}
}
主要分方法
1.动态方法实现注解接口
public <T> T create(final Class<T> service) {
Utils.validateServiceInterface(service);
if (validateEagerly) {
eagerlyValidateMethods(service);
}
return (T) Proxy.newProxyInstance(service.getClassLoader(), new Class<?>[] { service },
new InvocationHandler() {
private final Platform platform = Platform.get();
private final Object[] emptyArgs = new Object[0];
@Override public Object invoke(Object proxy, Method method, @Nullable Object[] args)
throws Throwable {
// If the method is a method from Object then defer to normal invocation.
if (method.getDeclaringClass() == Object.class) {
return method.invoke(this, args);
}
if (platform.isDefaultMethod(method)) {
return platform.invokeDefaultMethod(method, service, proxy, args);
}
return loadServiceMethod(method).invoke(args != null ? args : emptyArgs);
}
});
}
2.解析缓存注解等参数
ServiceMethod<?> loadServiceMethod(Method method) {
ServiceMethod<?> result = serviceMethodCache.get(method);
if (result != null) return result;
synchronized (serviceMethodCache) {
result = serviceMethodCache.get(method);
if (result == null) {
result = ServiceMethod.parseAnnotations(this, method);
serviceMethodCache.put(method, result);
}
}
return result;
}
3.网络请求适配器
@Override ReturnT invoke(Object[] args) {
return callAdapter.adapt(
new OkHttpCall<>(requestFactory, args, callFactory, responseConverter));
}
4.实现适配器方法
@Override public Call<Object> adapt(Call<Object> call) {
return new ExecutorCallbackCall<>(callbackExecutor, call);
}
5.正真的请求和返回参数的类
final class ExecutorCallAdapterFactory extends CallAdapter.Factory {
final Executor callbackExecutor;
ExecutorCallAdapterFactory(Executor callbackExecutor) {
this.callbackExecutor = callbackExecutor;
}
@Override public @Nullable CallAdapter<?, ?> get(
Type returnType, Annotation[] annotations, Retrofit retrofit) {
if (getRawType(returnType) != Call.class) {
return null;
}
final Type responseType = Utils.getCallResponseType(returnType);
return new CallAdapter<Object, Call<?>>() {
@Override public Type responseType() {
return responseType;
}
@Override public Call<Object> adapt(Call<Object> call) {
return new ExecutorCallbackCall<>(callbackExecutor, call);
}
};
}
static final class ExecutorCallbackCall<T> implements Call<T> {
final Executor callbackExecutor;
final Call<T> delegate;
ExecutorCallbackCall(Executor callbackExecutor, Call<T> delegate) {
this.callbackExecutor = callbackExecutor;
this.delegate = delegate;
}
@Override public void enqueue(final Callback<T> callback) {
checkNotNull(callback, "callback == null");
delegate.enqueue(new Callback<T>() {
@Override public void onResponse(Call<T> call, final Response<T> response) {
callbackExecutor.execute(new Runnable() {
@Override public void run() {
if (delegate.isCanceled()) {
// Emulate OkHttp's behavior of throwing/delivering an IOException on cancellation.
callback.onFailure(ExecutorCallbackCall.this, new IOException("Canceled"));
} else {
callback.onResponse(ExecutorCallbackCall.this, response);
}
}
});
}
@Override public void onFailure(Call<T> call, final Throwable t) {
callbackExecutor.execute(new Runnable() {
@Override public void run() {
callback.onFailure(ExecutorCallbackCall.this, t);
}
});
}
});
}
@Override public boolean isExecuted() {
return delegate.isExecuted();
}
@Override public Response<T> execute() throws IOException {
return delegate.execute();
}
@Override public void cancel() {
delegate.cancel();
}
@Override public boolean isCanceled() {
return delegate.isCanceled();
}
@SuppressWarnings("CloneDoesntCallSuperClone") // Performing deep clone.
@Override public Call<T> clone() {
return new ExecutorCallbackCall<>(callbackExecutor, delegate.clone());
}
@Override public Request request() {
return delegate.request();
}
}
}
结合RxJava用RxJava2CallAdapter
//第一步创建工厂CallAdapterFactory
.addCallAdapterFactory(RxJava2CallAdapterFactory.create())
//第二步实例化工厂方法
public static RxJava2CallAdapterFactory create() {
return new RxJava2CallAdapterFactory(null, false);
}
//第三步创建网络适配器RxJava2CallAdapter
public CallAdapter<?, ?> get(Type returnType, Annotation[] annotations, Retrofit retrofit) {
Class<?> rawType = getRawType(returnType);
if (rawType == Completable.class) {
// Completable is not parameterized (which is what the rest of this method deals with) so it
// can only be created with a single configuration.
return new RxJava2CallAdapter(Void.class, scheduler, isAsync, false, true, false, false,
false, true);
}
//第四步传入Call请求器到RxJava2CallAdapter
@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);
}
if (isSingle) {
return observable.singleOrError();
}
if (isMaybe) {
return observable.singleElement();
}
if (isCompletable) {
return observable.ignoreElements();
}
return observable;
}
//第五步具体实现方法
final class CallEnqueueObservable<T> extends Observable<Response<T>> {
private final Call<T> originalCall;
CallEnqueueObservable(Call<T> originalCall) {
this.originalCall = originalCall;
}
@Override protected void subscribeActual(Observer<? super Response<T>> observer) {
// Since Call is a one-shot type, clone it for each new observer.
Call<T> call = originalCall.clone();
CallCallback<T> callback = new CallCallback<>(call, observer);
observer.onSubscribe(callback);
call.enqueue(callback);
}
private static final class CallCallback<T> implements Disposable, Callback<T> {
private final Call<?> call;
private final Observer<? super Response<T>> observer;
boolean terminated = false;
CallCallback(Call<?> call, Observer<? super Response<T>> observer) {
this.call = call;
this.observer = observer;
}
@Override public void onResponse(Call<T> call, Response<T> response) {
if (call.isCanceled()) return;
try {
observer.onNext(response);
if (!call.isCanceled()) {
terminated = true;
observer.onComplete();
}
} catch (Throwable t) {
if (terminated) {
RxJavaPlugins.onError(t);
} else if (!call.isCanceled()) {
try {
observer.onError(t);
} catch (Throwable inner) {
Exceptions.throwIfFatal(inner);
RxJavaPlugins.onError(new CompositeException(t, inner));
}
}
}
}
@Override public void onFailure(Call<T> call, Throwable t) {
if (call.isCanceled()) return;
try {
observer.onError(t);
} catch (Throwable inner) {
Exceptions.throwIfFatal(inner);
RxJavaPlugins.onError(new CompositeException(t, inner));
}
}
@Override public void dispose() {
call.cancel();
}
@Override public boolean isDisposed() {
return call.isCanceled();
}
}
}
image.png
