Add annotation for skipping the callback executor

retrofit/src/main/java/retrofit2/DefaultCallAdapterFactory.java

@@ -15,33 +15,110 @@
  */
 package retrofit2;
 
+import java.io.IOException;
 import java.lang.annotation.Annotation;
+import java.lang.reflect.ParameterizedType;
 import java.lang.reflect.Type;
+import java.util.concurrent.Executor;
 import javax.annotation.Nullable;
+import okhttp3.Request;
+
+import static retrofit2.Utils.checkNotNull;
 
-/**
- * Creates call adapters for that uses the same thread for both I/O and application-level
- * callbacks. For synchronous calls this is the application thread making the request; for
- * asynchronous calls this is a thread provided by OkHttp's dispatcher.
- */
 final class DefaultCallAdapterFactory extends CallAdapter.Factory {
-  static final CallAdapter.Factory INSTANCE = new DefaultCallAdapterFactory();
+  private final @Nullable Executor callbackExecutor;
+
+  DefaultCallAdapterFactory(@Nullable Executor callbackExecutor) {
+    this.callbackExecutor = callbackExecutor;
+  }
 
   @Override public @Nullable CallAdapter<?, ?> get(
       Type returnType, Annotation[] annotations, Retrofit retrofit) {
     if (getRawType(returnType) != Call.class) {
       return null;
     }
+    if (!(returnType instanceof ParameterizedType)) {
+      throw new IllegalArgumentException(
+          "Call return type must be parameterized as Call<Foo> or Call<? extends Foo>");
+    }
+    final Type responseType = Utils.getParameterUpperBound(0, (ParameterizedType) returnType);
+
+    final Executor executor = Utils.isAnnotationPresent(annotations, SkipCallbackExecutor.class)
+        ? null
+        : callbackExecutor;
 
-    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 call;
+        return executor == null
+            ? call
+            : new ExecutorCallbackCall<>(executor, 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();
+    }
+  }
 }

retrofit/src/main/java/retrofit2/ExecutorCallAdapterFactory.java

-/*
- * Copyright (C) 2015 Square, Inc.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package retrofit2;
-
-import java.io.IOException;
-import java.lang.annotation.Annotation;
-import java.lang.reflect.Type;
-import java.util.concurrent.Executor;
-import javax.annotation.Nullable;
-import okhttp3.Request;
-
-import static retrofit2.Utils.checkNotNull;
-
-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();
-    }
-  }
-}

retrofit/src/main/java/retrofit2/Platform.java

@@ -62,10 +62,7 @@ class Platform {
 
   List<? extends CallAdapter.Factory> defaultCallAdapterFactories(
       @Nullable Executor callbackExecutor) {
-    if (callbackExecutor != null) {
-      return singletonList(new ExecutorCallAdapterFactory(callbackExecutor));
-    }
-    return singletonList(DefaultCallAdapterFactory.INSTANCE);
+    return singletonList(new DefaultCallAdapterFactory(callbackExecutor));
   }
 
   int defaultCallAdapterFactoriesSize() {
@@ -111,11 +108,7 @@ class Platform {
         @Nullable Executor callbackExecutor) {
       List<CallAdapter.Factory> factories = new ArrayList<>(2);
       factories.add(CompletableFutureCallAdapterFactory.INSTANCE);
-      if (callbackExecutor != null) {
-        factories.add(new ExecutorCallAdapterFactory(callbackExecutor));
-      } else {
-        factories.add(DefaultCallAdapterFactory.INSTANCE);
-      }
+      factories.add(new DefaultCallAdapterFactory(callbackExecutor));
       return unmodifiableList(factories);
     }
 
@@ -148,7 +141,7 @@ class Platform {
     @Override List<? extends CallAdapter.Factory> defaultCallAdapterFactories(
         @Nullable Executor callbackExecutor) {
       if (callbackExecutor == null) throw new AssertionError();
-      ExecutorCallAdapterFactory executorFactory = new ExecutorCallAdapterFactory(callbackExecutor);
+      DefaultCallAdapterFactory executorFactory = new DefaultCallAdapterFactory(callbackExecutor);
       return Build.VERSION.SDK_INT >= 24
         ? asList(CompletableFutureCallAdapterFactory.INSTANCE, executorFactory)
         : singletonList(executorFactory);

retrofit/src/main/java/retrofit2/SkipCallbackExecutor.java

+/*
+ * Copyright (C) 2019 Square, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package retrofit2;
+
+import java.lang.annotation.Annotation;
+import java.lang.annotation.Documented;
+import java.lang.annotation.Retention;
+import java.lang.annotation.Target;
+import java.lang.reflect.Type;
+
+import static java.lang.annotation.ElementType.METHOD;
+import static java.lang.annotation.RetentionPolicy.RUNTIME;
+
+/**
+ * Change the behavior of a {@code Call<BodyType>} return type to not use the
+ * {@linkplain Retrofit#callbackExecutor() callback executor} for invoking the
+ * {@link Callback#onResponse(Call, Response) onResponse} or
+ * {@link Callback#onFailure(Call, Throwable) onFailure} methods.
+ *
+ * <pre>{@code
+ * @SkipCallbackExecutor
+ * @Get("user/{id}/token")
+ * Call<String> getToken(@Path("id") long id);
+ * }</pre>
+ *
+ * This annotation can also be used when a {@link CallAdapter.Factory} <em>explicitly</em> delegates
+ * to the built-in factory for {@link Call} via
+ * {@link Retrofit#nextCallAdapter(CallAdapter.Factory, Type, Annotation[])} in order for the
+ * returned {@link Call} to skip the executor. (Note: by default, a {@link Call} supplied directly
+ * to a {@link CallAdapter} will already skip the callback executor. The annotation is only useful
+ * when looking up the built-in adapter.)
+ */
+@Documented
+@Target(METHOD)
+@Retention(RUNTIME)
+public @interface SkipCallbackExecutor {
+}

retrofit/src/main/java/retrofit2/Utils.java

@@ -383,14 +383,6 @@ final class Utils {
         + "GenericArrayType, but <" + type + "> is of type " + className);
   }
 
-  static Type getCallResponseType(Type returnType) {
-    if (!(returnType instanceof ParameterizedType)) {
-      throw new IllegalArgumentException(
-          "Call return type must be parameterized as Call<Foo> or Call<? extends Foo>");
-    }
-    return getParameterUpperBound(0, (ParameterizedType) returnType);
-  }
-
   private static final class ParameterizedTypeImpl implements ParameterizedType {
     private final Type ownerType;
     private final Type rawType;

retrofit/src/test/java/retrofit2/ExecutorCallAdapterFactoryTest.java → retrofit/src/test/java/retrofit2/DefaultCallAdapterFactoryTest.java

@@ -33,7 +33,7 @@ import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.verifyNoMoreInteractions;
 
 @SuppressWarnings("unchecked")
-public final class ExecutorCallAdapterFactoryTest {
+public final class DefaultCallAdapterFactoryTest {
   private static final Annotation[] NO_ANNOTATIONS = new Annotation[0];
 
   private final Retrofit retrofit = new Retrofit.Builder()
@@ -45,7 +45,7 @@ public final class ExecutorCallAdapterFactoryTest {
       runnable.run();
     }
   });
-  private final CallAdapter.Factory factory = new ExecutorCallAdapterFactory(callbackExecutor);
+  private final CallAdapter.Factory factory = new DefaultCallAdapterFactory(callbackExecutor);
 
   @Test public void rawTypeThrows() {
     try {

retrofit/src/test/java/retrofit2/RetrofitTest.java

@@ -77,6 +77,9 @@ public final class RetrofitTest {
     @GET("/") Call<okhttp3.Response> badType2();
 
     @GET("/") Call<ResponseBody> getResponseBody();
+    @SkipCallbackExecutor
+    @GET("/") Call<ResponseBody> getResponseBodySkippedExecutor();
+
     @GET("/") Call<Void> getVoid();
     @POST("/") Call<ResponseBody> postRequestBody(@Body RequestBody body);
     @GET("/") Call<ResponseBody> queryString(@Query("foo") String foo);
@@ -1341,6 +1344,62 @@ public final class RetrofitTest {
     verifyNoMoreInteractions(executor);
   }
 
+  @Test public void skippedCallbackExecutorNotUsedForSuccess() throws InterruptedException {
+    Executor executor = new Executor() {
+      @Override public void execute(Runnable command) {
+        fail();
+      }
+    };
+    Retrofit retrofit = new Retrofit.Builder()
+        .baseUrl(server.url("/"))
+        .callbackExecutor(executor)
+        .build();
+    CallMethod service = retrofit.create(CallMethod.class);
+    Call<ResponseBody> call = service.getResponseBodySkippedExecutor();
+
+    server.enqueue(new MockResponse());
+
+    final CountDownLatch latch = new CountDownLatch(1);
+    call.enqueue(new Callback<ResponseBody>() {
+      @Override public void onResponse(Call<ResponseBody> call, Response<ResponseBody> response) {
+        latch.countDown();
+      }
+
+      @Override public void onFailure(Call<ResponseBody> call, Throwable t) {
+        t.printStackTrace();
+      }
+    });
+    assertTrue(latch.await(2, TimeUnit.SECONDS));
+  }
+
+  @Test public void skippedCallbackExecutorNotUsedForFailure() throws InterruptedException {
+    Executor executor = new Executor() {
+      @Override public void execute(Runnable command) {
+        fail();
+      }
+    };
+    Retrofit retrofit = new Retrofit.Builder()
+        .baseUrl(server.url("/"))
+        .callbackExecutor(executor)
+        .build();
+    CallMethod service = retrofit.create(CallMethod.class);
+    Call<ResponseBody> call = service.getResponseBodySkippedExecutor();
+
+    server.enqueue(new MockResponse().setSocketPolicy(DISCONNECT_AT_START));
+
+    final CountDownLatch latch = new CountDownLatch(1);
+    call.enqueue(new Callback<ResponseBody>() {
+      @Override public void onResponse(Call<ResponseBody> call, Response<ResponseBody> response) {
+        throw new AssertionError();
+      }
+
+      @Override public void onFailure(Call<ResponseBody> call, Throwable t) {
+        latch.countDown();
+      }
+    });
+    assertTrue(latch.await(2, TimeUnit.SECONDS));
+  }
+
   /** Confirm that Retrofit encodes parameters when the call is executed, and not earlier. */
   @Test public void argumentCapture() throws Exception {
     AtomicInteger i = new AtomicInteger();