Rxjava 1.1.1 and implement Completable support in rxjava adapter

pom.xml

@@ -53,7 +53,7 @@
     <animal.sniffer.version>1.14</animal.sniffer.version>
 
     <!-- Adapter Dependencies -->
-    <rxjava.version>1.1.0</rxjava.version>
+    <rxjava.version>1.1.1</rxjava.version>
 
     <!-- Converter Dependencies -->
     <gson.version>2.4</gson.version>

retrofit-adapters/rxjava/src/main/java/retrofit2/adapter/rxjava/CompletableHelper.java

+/*
+ * Copyright (C) 2016 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.adapter.rxjava;
+
+import java.lang.reflect.Type;
+import retrofit2.Call;
+import retrofit2.CallAdapter;
+import retrofit2.Response;
+import rx.Completable;
+import rx.Completable.CompletableOnSubscribe;
+import rx.Completable.CompletableSubscriber;
+import rx.Subscription;
+import rx.exceptions.Exceptions;
+import rx.functions.Action0;
+import rx.subscriptions.Subscriptions;
+
+final class CompletableHelper {
+  static final CallAdapter<Completable> INSTANCE =
+      new CallAdapter<Completable>() {
+        @Override public Type responseType() {
+          return Void.class;
+        }
+
+        @Override public Completable adapt(Call call) {
+          return Completable.create(new CompletableCallOnSubscribe(call));
+        }
+      };
+
+  private static final class CompletableCallOnSubscribe implements CompletableOnSubscribe {
+    private final Call originalCall;
+
+    CompletableCallOnSubscribe(Call originalCall) {
+      this.originalCall = originalCall;
+    }
+
+    @Override public void call(CompletableSubscriber subscriber) {
+      // Since Call is a one-shot type, clone it for each new subscriber.
+      final Call call = originalCall.clone();
+
+      // Attempt to cancel the call if it is still in-flight on unsubscription.
+      Subscription subscription = Subscriptions.create(new Action0() {
+        @Override public void call() {
+          call.cancel();
+        }
+      });
+      subscriber.onSubscribe(subscription);
+
+      try {
+        Response response = call.execute();
+        if (!subscription.isUnsubscribed()) {
+          if (response.isSuccess()) {
+            subscriber.onCompleted();
+          } else {
+            subscriber.onError(new HttpException(response));
+          }
+        }
+      } catch (Throwable t) {
+        Exceptions.throwIfFatal(t);
+        if (!subscription.isUnsubscribed()) {
+          subscriber.onError(t);
+        }
+      }
+    }
+  }
+}

retrofit-adapters/rxjava/src/main/java/retrofit2/adapter/rxjava/RxJavaCallAdapterFactory.java

@@ -46,16 +46,26 @@ public final class RxJavaCallAdapterFactory extends CallAdapter.Factory {
   @Override
   public CallAdapter<?> get(Type returnType, Annotation[] annotations, Retrofit retrofit) {
     Class<?> rawType = getRawType(returnType);
-    boolean isSingle = "rx.Single".equals(rawType.getCanonicalName());
-    if (rawType != Observable.class && !isSingle) {
+    String canonicalName = rawType.getCanonicalName();
+    boolean isSingle = "rx.Single".equals(canonicalName);
+    boolean isCompletable = "rx.Completable".equals(canonicalName);
+    if (rawType != Observable.class && !isSingle && !isCompletable) {
       return null;
     }
-    if (!(returnType instanceof ParameterizedType)) {
+    if (!isCompletable && !(returnType instanceof ParameterizedType)) {
       String name = isSingle ? "Single" : "Observable";
       throw new IllegalStateException(name + " return type must be parameterized"
           + " as " + name + "<Foo> or " + name + "<? extends Foo>");
     }
 
+    if (isCompletable) {
+      // Add Completable-converter wrapper from a separate class. This defers classloading such that
+      // regular Observable operation can be leveraged without relying on this unstable RxJava API.
+      // Note that this has to be done separately since Completable doesn't have a parametrized
+      // type.
+      return CompletableHelper.INSTANCE;
+    }
+
     CallAdapter<Observable<?>> callAdapter = getCallAdapter(returnType);
     if (isSingle) {
       // Add Single-converter wrapper from a separate class. This defers classloading such that

retrofit-adapters/rxjava/src/test/java/retrofit2/adapter/rxjava/CompletableTest.java

+/*
+ * Copyright (C) 2016 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.adapter.rxjava;
+
+import okhttp3.mockwebserver.MockResponse;
+import okhttp3.mockwebserver.MockWebServer;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import retrofit2.Retrofit;
+import retrofit2.http.GET;
+import rx.Completable;
+
+import java.io.IOException;
+
+import static okhttp3.mockwebserver.SocketPolicy.DISCONNECT_AFTER_REQUEST;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.Assert.fail;
+
+public final class CompletableTest {
+  @Rule
+  public final MockWebServer server = new MockWebServer();
+
+  interface Service {
+    @GET("/") Completable completable();
+  }
+
+  private Service service;
+
+  @Before public void setUp() {
+    Retrofit retrofit = new Retrofit.Builder()
+        .baseUrl(server.url("/"))
+        .addConverterFactory(new StringConverterFactory())
+        .addCallAdapterFactory(RxJavaCallAdapterFactory.create())
+        .build();
+    service = retrofit.create(Service.class);
+  }
+
+  @Test public void completableSuccess200() {
+    server.enqueue(new MockResponse().setBody("Hi"));
+
+    service.completable().await();
+  }
+
+  @Test public void completableSuccess404() {
+    server.enqueue(new MockResponse().setResponseCode(404));
+
+    try {
+      service.completable().await();
+      fail();
+    } catch (RuntimeException e) {
+      Throwable cause = e.getCause();
+      assertThat(cause).isInstanceOf(HttpException.class).hasMessage("HTTP 404 Client Error");
+    }
+  }
+
+  @Test public void completableFailure() {
+    server.enqueue(new MockResponse().setSocketPolicy(DISCONNECT_AFTER_REQUEST));
+
+    try {
+      service.completable().await();
+      fail();
+    } catch (RuntimeException e) {
+      assertThat(e.getCause()).isInstanceOf(IOException.class);
+    }
+  }
+}