privateCallAdapter<T, R>createCallAdapter() {Type returnType =method.getGenericReturnType();//获取返回值//校验返回值类型if (Utils.hasUnresolvableType(returnType)) {throwmethodError("Method return type must not include a type variable or wildcard: %s", returnType); }if (returnType ==void.class) {throwmethodError("Service methods cannot return void."); }Annotation[] annotations =method.getAnnotations();try {//noinspection unchecked//调用Retrofit的 callAdapter方法来获取CallAdapterreturn (CallAdapter<T, R>) retrofit.callAdapter(returnType, annotations); } catch (RuntimeException e) { // Wide exception range because factories are user code.throwmethodError(e,"Unable to create call adapter for %s", returnType); }}
public CallAdapter<?, ?> callAdapter(Type returnType, Annotation[] annotations) {
return nextCallAdapter(null, returnType, annotations);
}
public CallAdapter<?, ?> nextCallAdapter(CallAdapter.Factory skipPast, Type returnType,
Annotation[] annotations) {
checkNotNull(returnType, "returnType == null");
checkNotNull(annotations, "annotations == null");
int start = adapterFactories.indexOf(skipPast) + 1;
for (int i = start, count = adapterFactories.size(); i < count; i++) {
//调用CallAdapter.Factory的get方法来获取对应的CallAdapter
//这里看出是根据返回值类型来获取不同的CallAdapter
//
CallAdapter<?,?> adapter = adapterFactories.get(i).get(returnType, annotations, this);
if (adapter != null) {
return adapter;
}
}
StringBuilder builder = new StringBuilder("Could not locate call adapter for ")
.append(returnType)
.append(".\n");
if (skipPast != null) {
builder.append(" Skipped:");
for (int i = 0; i < start; i++) {
builder.append("\n * ").append(adapterFactories.get(i).getClass().getName());
}
builder.append('\n');
}
builder.append(" Tried:");
for (int i = start, count = adapterFactories.size(); i < count; i++) {
builder.append("\n * ").append(adapterFactories.get(i).getClass().getName());
}
throw new IllegalArgumentException(builder.toString());
}
public ServiceMethod build() {
callAdapter = createCallAdapter();//创建CallAdapter
responseType = callAdapter.responseType();//获取返回值类型
if (responseType == Response.class || responseType == okhttp3.Response.class) {
throw methodError("'"
+ Utils.getRawType(responseType).getName()
+ "' is not a valid response body type. Did you mean ResponseBody?");
}
responseConverter = createResponseConverter();//创建ResponseConverter
for (Annotation annotation : methodAnnotations) {
parseMethodAnnotation(annotation);//遍历解析方法注解
}
if (httpMethod == null) {
throw methodError("HTTP method annotation is required (e.g., @GET, @POST, etc.).");
}
if (!hasBody) {
if (isMultipart) {
throw methodError(
"Multipart can only be specified on HTTP methods with request body (e.g., @POST).");
}
if (isFormEncoded) {
throw methodError("FormUrlEncoded can only be specified on HTTP methods with "
+ "request body (e.g., @POST).");
}
}
int parameterCount = parameterAnnotationsArray.length;//获取参数注解长度
parameterHandlers = new ParameterHandler<?>[parameterCount];//创建参数处理的数组
//遍历每个参数
for (int p = 0; p < parameterCount; p++) {
Type parameterType = parameterTypes[p];
//校验每个参数的类型
if (Utils.hasUnresolvableType(parameterType)) {
throw parameterError(p, "Parameter type must not include a type variable or wildcard: %s",
parameterType);
}
//单个参数上的注解
Annotation[] parameterAnnotations = parameterAnnotationsArray[p];
if (parameterAnnotations == null) {
throw parameterError(p, "No Retrofit annotation found.");
}
parameterHandlers[p] = parseParameter(p, parameterType, parameterAnnotations);
}
if (relativeUrl == null && !gotUrl) {
throw methodError("Missing either @%s URL or @Url parameter.", httpMethod);
}
if (!isFormEncoded && !isMultipart && !hasBody && gotBody) {
throw methodError("Non-body HTTP method cannot contain @Body.");
}
if (isFormEncoded && !gotField) {
throw methodError("Form-encoded method must contain at least one @Field.");
}
if (isMultipart && !gotPart) {
throw methodError("Multipart method must contain at least one @Part.");
}
return new ServiceMethod<>(this);
}
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);
if (!Void.class.equals(responseType)) {
throw methodError("HEAD method must use Void as response type.");
}
} 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("@Headers annotation is empty.");
}
headers = parseHeaders(headersToParse);
} else if (annotation instanceof Multipart) {
if (isFormEncoded) {
throw methodError("Only one encoding annotation is allowed.");
}
isMultipart = true;
} else if (annotation instanceof FormUrlEncoded) {
if (isMultipart) {
throw methodError("Only one encoding annotation is allowed.");
}
isFormEncoded = true;
}
}
abstract void apply(RequestBuilder builder, T value) throws IOException;
final ParameterHandler<Iterable<T>> iterable() {
return new ParameterHandler<Iterable<T>>() {
@Override void apply(RequestBuilder builder, Iterable<T> values) throws IOException {
if (values == null) return; // Skip null values.
for (T value : values) {
ParameterHandler.this.apply(builder, value);
}
}
};
}
final ParameterHandler<Object> array() {
return new ParameterHandler<Object>() {
@Override void apply(RequestBuilder builder, Object values) throws IOException {
if (values == null) return; // Skip null values.
for (int i = 0, size = Array.getLength(values); i < size; i++) {
//noinspection unchecked
ParameterHandler.this.apply(builder, (T) Array.get(values, i));
}
}
};
}
/** Builds an HTTP request from method arguments. */
Request toRequest(Object... args) throws IOException {
RequestBuilder requestBuilder = new RequestBuilder(httpMethod, baseUrl, relativeUrl, headers,
contentType, hasBody, isFormEncoded, isMultipart);
@SuppressWarnings("unchecked") // It is an error to invoke a method with the wrong arg types.
ParameterHandler<Object>[] handlers = (ParameterHandler<Object>[]) parameterHandlers;
int argumentCount = args != null ? args.length : 0;
if (argumentCount != handlers.length) {
throw new IllegalArgumentException("Argument count (" + argumentCount
+ ") doesn't match expected count (" + handlers.length + ")");
}
for (int p = 0; p < argumentCount; p++) {
handlers[p].apply(requestBuilder, args[p]);
}
return requestBuilder.build();
}
/** Builds a method return value from an HTTP response body. */
R toResponse(ResponseBody body) throws IOException {
return responseConverter.convert(body);
}
final class RequestBuilder {
private static final char[] HEX_DIGITS =
{ '0', '1', '2', '3', '4', '5', '6', '7', '8', '9', 'A', 'B', 'C', 'D', 'E', 'F' };
private static final String PATH_SEGMENT_ALWAYS_ENCODE_SET = " \"<>^`{}|\\?#";
private final String method;
private final HttpUrl baseUrl;
private String relativeUrl;
private HttpUrl.Builder urlBuilder;
private final Request.Builder requestBuilder;
private MediaType contentType;
private final boolean hasBody;
private MultipartBody.Builder multipartBuilder;
private FormBody.Builder formBuilder;
private RequestBody body;
RequestBuilder(String method, HttpUrl baseUrl, String relativeUrl, Headers headers,
MediaType contentType, boolean hasBody, boolean isFormEncoded, boolean isMultipart) {
this.method = method;
this.baseUrl = baseUrl;
this.relativeUrl = relativeUrl;
this.requestBuilder = new Request.Builder();
this.contentType = contentType;
this.hasBody = hasBody;
if (headers != null) {
requestBuilder.headers(headers);//添加header
}
if (isFormEncoded) {//form请求
// Will be set to 'body' in 'build'.
formBuilder = new FormBody.Builder();
} else if (isMultipart) {
// Will be set to 'body' in 'build'.
multipartBuilder = new MultipartBody.Builder();
multipartBuilder.setType(MultipartBody.FORM);
}
}
void setRelativeUrl(Object relativeUrl) {
if (relativeUrl == null) throw new NullPointerException("@Url parameter is null.");
this.relativeUrl = relativeUrl.toString();
}
void addHeader(String name, String value) {
if ("Content-Type".equalsIgnoreCase(name)) {
MediaType type = MediaType.parse(value);
if (type == null) {
throw new IllegalArgumentException("Malformed content type: " + value);
}
contentType = type;
} else {
requestBuilder.addHeader(name, value);
}
}
void addPathParam(String name, String value, boolean encoded) {
if (relativeUrl == null) {
// The relative URL is cleared when the first query parameter is set.
throw new AssertionError();
}
relativeUrl = relativeUrl.replace("{" + name + "}", canonicalizeForPath(value, encoded));
}
private static String canonicalizeForPath(String input, boolean alreadyEncoded) {
int codePoint;
for (int i = 0, limit = input.length(); i < limit; i += Character.charCount(codePoint)) {
codePoint = input.codePointAt(i);
if (codePoint < 0x20 || codePoint >= 0x7f
|| PATH_SEGMENT_ALWAYS_ENCODE_SET.indexOf(codePoint) != -1
|| (!alreadyEncoded && (codePoint == '/' || codePoint == '%'))) {
// Slow path: the character at i requires encoding!
Buffer out = new Buffer();
out.writeUtf8(input, 0, i);
canonicalizeForPath(out, input, i, limit, alreadyEncoded);
return out.readUtf8();
}
}
// Fast path: no characters required encoding.
return input;
}
private static void canonicalizeForPath(Buffer out, String input, int pos, int limit,
boolean alreadyEncoded) {
Buffer utf8Buffer = null; // Lazily allocated.
int codePoint;
for (int i = pos; i < limit; i += Character.charCount(codePoint)) {
codePoint = input.codePointAt(i);
if (alreadyEncoded
&& (codePoint == '\t' || codePoint == '\n' || codePoint == '\f' || codePoint == '\r')) {
// Skip this character.
} else if (codePoint < 0x20 || codePoint >= 0x7f
|| PATH_SEGMENT_ALWAYS_ENCODE_SET.indexOf(codePoint) != -1
|| (!alreadyEncoded && (codePoint == '/' || codePoint == '%'))) {
// Percent encode this character.
if (utf8Buffer == null) {
utf8Buffer = new Buffer();
}
utf8Buffer.writeUtf8CodePoint(codePoint);
while (!utf8Buffer.exhausted()) {
int b = utf8Buffer.readByte() & 0xff;
out.writeByte('%');
out.writeByte(HEX_DIGITS[(b >> 4) & 0xf]);
out.writeByte(HEX_DIGITS[b & 0xf]);
}
} else {
// This character doesn't need encoding. Just copy it over.
out.writeUtf8CodePoint(codePoint);
}
}
}
//添加get请求参数
void addQueryParam(String name, String value, boolean encoded) {
if (relativeUrl != null) {
// Do a one-time combination of the built relative URL and the base URL.
urlBuilder = baseUrl.newBuilder(relativeUrl);
if (urlBuilder == null) {
throw new IllegalArgumentException(
"Malformed URL. Base: " + baseUrl + ", Relative: " + relativeUrl);
}
relativeUrl = null;
}
if (encoded) {
urlBuilder.addEncodedQueryParameter(name, value);
} else {
urlBuilder.addQueryParameter(name, value);
}
}
//添加form表单请求参数
void addFormField(String name, String value, boolean encoded) {
if (encoded) {
formBuilder.addEncoded(name, value);
} else {
formBuilder.add(name, value);
}
}
void addPart(Headers headers, RequestBody body) {
multipartBuilder.addPart(headers, body);
}
void addPart(MultipartBody.Part part) {
multipartBuilder.addPart(part);
}
void setBody(RequestBody body) {
this.body = body;
}
Request build() {
HttpUrl url;
HttpUrl.Builder urlBuilder = this.urlBuilder;
if (urlBuilder != null) {
url = urlBuilder.build();
} else {
// No query parameters triggered builder creation, just combine the relative URL and base URL.
url = baseUrl.resolve(relativeUrl);
if (url == null) {
throw new IllegalArgumentException(
"Malformed URL. Base: " + baseUrl + ", Relative: " + relativeUrl);
}
}
RequestBody body = this.body;
if (body == null) {
// Try to pull from one of the builders.
if (formBuilder != null) {
body = formBuilder.build();
} else if (multipartBuilder != null) {
body = multipartBuilder.build();
} else if (hasBody) {
// Body is absent, make an empty body.
body = RequestBody.create(null, new byte[0]);
}
}
MediaType contentType = this.contentType;
if (contentType != null) {
if (body != null) {
body = new ContentTypeOverridingRequestBody(body, contentType);
} else {
requestBuilder.addHeader("Content-Type", contentType.toString());
}
}
return requestBuilder
.url(url)
.method(method, body)
.build();
}
private static class ContentTypeOverridingRequestBody extends RequestBody {
private final RequestBody delegate;
private final MediaType contentType;
ContentTypeOverridingRequestBody(RequestBody delegate, MediaType contentType) {
this.delegate = delegate;
this.contentType = contentType;
}
@Override public MediaType contentType() {
return contentType;
}
@Override public long contentLength() throws IOException {
return delegate.contentLength();
}
@Override public void writeTo(BufferedSink sink) throws IOException {
delegate.writeTo(sink);
}
}
}
public final class RxJavaCallAdapterFactory extends CallAdapter.Factory {
/**
* Returns an instance which creates synchronous observables that do not operate on any scheduler
* by default.
*/
public static RxJavaCallAdapterFactory create() {
return new RxJavaCallAdapterFactory(null);
}
/**
* Returns an instance which creates synchronous observables that
* {@linkplain Observable#subscribeOn(Scheduler) subscribe on} {@code scheduler} by default.
*/
public static RxJavaCallAdapterFactory createWithScheduler(Scheduler scheduler) {
if (scheduler == null) throw new NullPointerException("scheduler == null");
return new RxJavaCallAdapterFactory(scheduler);
}
private final Scheduler scheduler;
private RxJavaCallAdapterFactory(Scheduler scheduler) {
this.scheduler = scheduler;
}
@Override
public CallAdapter<?, ?> get(Type returnType, Annotation[] annotations, Retrofit retrofit) {
Class<?> rawType = getRawType(returnType);
boolean isSingle = rawType == Single.class;//判断返回值是否是Single.class
boolean isCompletable = "rx.Completable".equals(rawType.getCanonicalName());
if (rawType != Observable.class && !isSingle && !isCompletable) {
return null;
}
if (isCompletable) {
return new RxJavaCallAdapter(Void.class, scheduler, false, true, false, true);
}
boolean isResult = false;
boolean isBody = false;
Type responseType;
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>");
}
Type observableType = getParameterUpperBound(0, (ParameterizedType) returnType);
Class<?> rawObservableType = getRawType(observableType);
if (rawObservableType == Response.class) {
if (!(observableType instanceof ParameterizedType)) {
throw new IllegalStateException("Response must be parameterized"
+ " as Response<Foo> or Response<? extends Foo>");
}
responseType = getParameterUpperBound(0, (ParameterizedType) observableType);
} else if (rawObservableType == Result.class) {
if (!(observableType instanceof ParameterizedType)) {
throw new IllegalStateException("Result must be parameterized"
+ " as Result<Foo> or Result<? extends Foo>");
}
responseType = getParameterUpperBound(0, (ParameterizedType) observableType);
isResult = true;
} else {
responseType = observableType;
isBody = true;
}
return new RxJavaCallAdapter(responseType, scheduler, isResult, isBody, isSingle, false);
}
}
@Override public Object adapt(Call<R> call) {
OnSubscribe<Response<R>> callFunc = new CallOnSubscribe<>(call);//创建CallOnSubscribe
//获取OnSubscribe对象
OnSubscribe<?> func;
if (isResult) {
func = new ResultOnSubscribe<>(callFunc);
} else if (isBody) {
func = new BodyOnSubscribe<>(callFunc);
} else {
func = callFunc;
}
Observable<?> observable = Observable.create(func);
if (scheduler != null) {
observable = observable.subscribeOn(scheduler);
}
if (isSingle) {
return observable.toSingle();//Observable转换为Single
}
if (isCompletable) {
return CompletableHelper.toCompletable(observable);//Observable转换为Completable
}
return observable;
}
final class CallOnSubscribe<T> implements OnSubscribe<Response<T>> {
private final Call<T> originalCall;
CallOnSubscribe(Call<T> originalCall) {
this.originalCall = originalCall;
}
@Override public void call(Subscriber<? super Response<T>> subscriber) {
// Since Call is a one-shot type, clone it for each new subscriber.
Call<T> call = originalCall.clone();//clone 了原来的 call，因为 okhttp3.Call 是只能用一次的，所以每次都是新 clone 一个进行网络请求；
CallArbiter<T> arbiter = new CallArbiter<>(call, subscriber);
subscriber.add(arbiter);
subscriber.setProducer(arbiter);//调用producer的request方法
Response<T> response;
try {
response = call.execute();//调用OkHttpCall的execute()方法返回Response
} catch (Throwable t) {
Exceptions.throwIfFatal(t);
arbiter.emitError(t);//发送错误
return;
}
arbiter.emitResponse(response);//发送response
}
static final class CallArbiter<T> extends AtomicInteger implements Subscription, Producer {
private static final int STATE_WAITING = 0;
private static final int STATE_REQUESTED = 1;
private static final int STATE_HAS_RESPONSE = 2;
private static final int STATE_TERMINATED = 3;
private final Call<T> call;
private final Subscriber<? super Response<T>> subscriber;
private volatile Response<T> response;
CallArbiter(Call<T> call, Subscriber<? super Response<T>> subscriber) {
super(STATE_WAITING);
this.call = call;
this.subscriber = subscriber;
}
@Override public void unsubscribe() {
call.cancel();
}
@Override public boolean isUnsubscribed() {
return call.isCanceled();
}
@Override public void request(long amount) {
if (amount == 0) {
return;
}
while (true) {
int state = get();
switch (state) {
case STATE_WAITING:
if (compareAndSet(STATE_WAITING, STATE_REQUESTED)) {
return;
}
break; // State transition failed. Try again.
case STATE_HAS_RESPONSE:
if (compareAndSet(STATE_HAS_RESPONSE, STATE_TERMINATED)) {
deliverResponse(response);
return;
}
break; // State transition failed. Try again.
case STATE_REQUESTED:
case STATE_TERMINATED:
return; // Nothing to do.
default:
throw new IllegalStateException("Unknown state: " + state);
}
}
}
void emitResponse(Response<T> response) {
while (true) {
int state = get();//获取当前值
switch (state) {
case STATE_WAITING:
this.response = response;
if (compareAndSet(STATE_WAITING, STATE_HAS_RESPONSE)) {//如果当前值==STATE_WAITING,设置值为STATE_HAS_RESPONSE
return;
}
break; //状态改变失败重试
case STATE_REQUESTED:
if (compareAndSet(STATE_REQUESTED, STATE_TERMINATED)) {
deliverResponse(response);
return;
}
break; // State transition failed. Try again.
case STATE_HAS_RESPONSE:
case STATE_TERMINATED:
throw new AssertionError();
default:
throw new IllegalStateException("Unknown state: " + state);
}
}
}
private void deliverResponse(Response<T> response) {
try {
if (!isUnsubscribed()) {
subscriber.onNext(response);
}
} catch (Throwable t) {
Exceptions.throwIfFatal(t);
try {
subscriber.onError(t);
} catch (Throwable inner) {
Exceptions.throwIfFatal(inner);
CompositeException composite = new CompositeException(t, inner);
RxJavaPlugins.getInstance().getErrorHandler().handleError(composite);
}
return;
}
try {
subscriber.onCompleted();//发送完成
} catch (Throwable t) {
Exceptions.throwIfFatal(t);
RxJavaPlugins.getInstance().getErrorHandler().handleError(t);
}
}
void emitError(Throwable t) {
set(STATE_TERMINATED);
if (!isUnsubscribed()) {
try {
subscriber.onError(t);
} catch (Throwable inner) {
Exceptions.throwIfFatal(inner);
CompositeException composite = new CompositeException(t, inner);
RxJavaPlugins.getInstance().getErrorHandler().handleError(composite);
}
}
}
}
}
final class BodyOnSubscribe<T> implements OnSubscribe<T> {
private final OnSubscribe<Response<T>> upstream;
BodyOnSubscribe(OnSubscribe<Response<T>> upstream) {
this.upstream = upstream;
}
@Override public void call(Subscriber<? super T> subscriber) {
upstream.call(new BodySubscriber<>(subscriber));
}
private static class BodySubscriber<R> extends Subscriber<Response<R>> {
private final Subscriber<? super R> subscriber;
/** Indicates whether a terminal event has been sent to {@link #subscriber}. */
private boolean subscriberTerminated;
BodySubscriber(Subscriber<? super R> subscriber) {
super(subscriber);
this.subscriber = subscriber;
}
@Override public void onNext(Response<R> response) {
if (response.isSuccessful()) {
subscriber.onNext(response.body());
} else {
subscriberTerminated = true;
Throwable t = new HttpException(response);
try {
subscriber.onError(t);
} catch (Throwable inner) {
Exceptions.throwIfFatal(inner);
CompositeException composite = new CompositeException(t, inner);
RxJavaPlugins.getInstance().getErrorHandler().handleError(composite);
}
}
}
@Override public void onError(Throwable throwable) {
if (!subscriberTerminated) {
subscriber.onError(throwable);
} else {
// This should never happen! onNext handles and forwards errors automatically.
Throwable broken = new AssertionError(
"This should never happen! Report as a Retrofit bug with the full stacktrace.");
//noinspection UnnecessaryInitCause Two-arg AssertionError constructor is 1.7+ only.
broken.initCause(throwable);
RxJavaPlugins.getInstance().getErrorHandler().handleError(broken);
}
}
@Override public void onCompleted() {
if (!subscriberTerminated) {
subscriber.onCompleted();
}
}
}
}
final class ResultOnSubscribe<T> implements OnSubscribe<Result<T>> {
private final OnSubscribe<Response<T>> upstream;
ResultOnSubscribe(OnSubscribe<Response<T>> upstream) {
this.upstream = upstream;
}
@Override public void call(Subscriber<? super Result<T>> subscriber) {
upstream.call(new ResultSubscriber<T>(subscriber));
}
private static class ResultSubscriber<R> extends Subscriber<Response<R>> {
private final Subscriber<? super Result<R>> subscriber;
ResultSubscriber(Subscriber<? super Result<R>> subscriber) {
super(subscriber);
this.subscriber = subscriber;
}
@Override public void onNext(Response<R> response) {
subscriber.onNext(Result.response(response));
}
@Override public void onError(Throwable throwable) {
try {
subscriber.onNext(Result.<R>error(throwable));
} catch (Throwable t) {
try {
subscriber.onError(t);
} catch (Throwable inner) {
Exceptions.throwIfFatal(inner);
CompositeException composite = new CompositeException(t, inner);
RxJavaPlugins.getInstance().getErrorHandler().handleError(composite);
}
return;
}
subscriber.onCompleted();
}
@Override public void onCompleted() {
subscriber.onCompleted();
}
}
}
