Repository: square/retrofit
Issue: Support rx.Single from RxJava?

With `rx.Single` released as experimented type of `Observable` do you see the value in supporting it as a return type for network requests?
https://github.com/ReactiveX/RxJava/blob/1.x/CHANGES.md#rxsingle
https://github.com/ReactiveX/RxJava/issues/1594#issuecomment-101300655

Quote from there:

```
A Single will always behave in one of 3 ways:

1) respond with an error
2) never respond
3) respond with a success

An Observable on the other hand may:

1) response with an error
2) never respond
3) respond successfully with no data and terminate
4) respond successfully with a single value and terminate
5) respond successfully with multiple values and terminate
7) respond successfully with one or more values and never terminate (waiting for more data)
```


Issue comments:
We are aware and it may be added. It's experimental nature adds a level of
volatility that if we choose to add it as the API may change in a
backwards-incompatible way.

On Tue, Jul 21, 2015, 3:12 AM Sergii Pechenizkyi notifications@github.com
wrote:

> With rx.Single released as experimented type of Observable do you see the
> value in supporting it as a return type for network requests?
> https://github.com/ReactiveX/RxJava/blob/1.x/CHANGES.md#rxsingle
> ReactiveX/RxJava#1594 (comment)
> https://github.com/ReactiveX/RxJava/issues/1594#issuecomment-101300655
> 
> Quote from there:
> 
> A Single will always behave in one of 3 ways:
> 
> 1) respond with an error
> 2) never respond
> 3) respond with a success
> 
> An Observable on the other hand may:
> 
> 1) response with an error
> 2) never respond
> 3) respond successfully with no data and terminate
> 4) respond successfully with a single value and terminate
> 5) respond successfully with multiple values and terminate
> 7) respond successfully with one or more values and never terminate (waiting for more data)
> 
> —
> Reply to this email directly or view it on GitHub
> https://github.com/square/retrofit/issues/969.

+1 for `Single` :)


Pull request review comments:

Patch that resolved the issue:
diff --git a/pom.xml b/pom.xml
index 51f8203296..439fd2ff6f 100644
--- a/pom.xml
+++ b/pom.xml
@@ -53,7 +53,7 @@
     <okhttp.version>2.5.0-SNAPSHOT</okhttp.version>
 
     <!-- Adapter Dependencies -->
-    <rxjava.version>1.0.10</rxjava.version>
+    <rxjava.version>1.0.13</rxjava.version>
 
     <!-- Converter Dependencies -->
     <gson.version>2.3.1</gson.version>
diff --git a/retrofit-adapters/rxjava/src/main/java/retrofit/ObservableCallAdapterFactory.java b/retrofit-adapters/rxjava/src/main/java/retrofit/ObservableCallAdapterFactory.java
index 65f6da878f..a621805879 100644
--- a/retrofit-adapters/rxjava/src/main/java/retrofit/ObservableCallAdapterFactory.java
+++ b/retrofit-adapters/rxjava/src/main/java/retrofit/ObservableCallAdapterFactory.java
@@ -18,12 +18,13 @@
 import java.io.IOException;
 import java.lang.reflect.ParameterizedType;
 import java.lang.reflect.Type;
-import rx.Observable;
-import rx.Subscriber;
+import rx.Single;
+import rx.SingleSubscriber;
 import rx.functions.Action0;
 import rx.functions.Func1;
 import rx.subscriptions.Subscriptions;
 
+
 /**
  * TODO docs
  */
@@ -43,7 +44,7 @@ private ObservableCallAdapterFactory() {
   }
 
   @Override public CallAdapter<?> get(Type returnType) {
-    if (Utils.getRawType(returnType) != Observable.class) {
+    if (Utils.getRawType(returnType) != Single.class) {
       return null;
     }
     if (!(returnType instanceof ParameterizedType)) {
@@ -75,14 +76,14 @@ private ObservableCallAdapterFactory() {
     return new SimpleCallAdapter(observableType);
   }
 
-  static final class CallOnSubscribe<T> implements Observable.OnSubscribe<Response<T>> {
+  static final class CallOnSubscribe<T> implements Single.OnSubscribe<Response<T>> {
     private final Call<T> originalCall;
 
     private CallOnSubscribe(Call<T> originalCall) {
       this.originalCall = originalCall;
     }
 
-    @Override public void call(final Subscriber<? super Response<T>> subscriber) {
+    @Override public void call(final SingleSubscriber<? super Response<T>> subscriber) {
       // Since Call is a one-shot type, clone it for each new subscriber.
       final Call<T> call = originalCall.clone();
 
@@ -99,12 +100,10 @@ private CallOnSubscribe(Call<T> originalCall) {
             return;
           }
           try {
-            subscriber.onNext(response);
+            subscriber.onSuccess(response);
           } catch (Throwable t) {
             subscriber.onError(t);
-            return;
           }
-          subscriber.onCompleted();
         }
 
         @Override public void onFailure(Throwable t) {
@@ -128,8 +127,8 @@ static final class ResponseCallAdapter<T> implements CallAdapter<T> {
       return responseType;
     }
 
-    @Override public Observable<Response<T>> adapt(Call<T> call) {
-      return Observable.create(new CallOnSubscribe<>(call));
+    @Override public Single<Response<T>> adapt(Call<T> call) {
+      return Single.create(new CallOnSubscribe<>(call));
     }
   }
 
@@ -144,14 +143,14 @@ static final class SimpleCallAdapter<T> implements CallAdapter<T> {
       return responseType;
     }
 
-    @Override public Observable<T> adapt(Call<T> call) {
-      return Observable.create(new CallOnSubscribe<>(call)) //
-          .flatMap(new Func1<Response<T>, Observable<T>>() {
-            @Override public Observable<T> call(Response<T> response) {
+    @Override public Single<T> adapt(Call<T> call) {
+      return Single.create(new CallOnSubscribe<>(call)) //
+          .flatMap(new Func1<Response<T>, Single<T>>() {
+            @Override public Single<T> call(Response<T> response) {
               if (response.isSuccess()) {
-                return Observable.just(response.body());
+                return Single.just(response.body());
               }
-              return Observable.error(new IOException()); // TODO non-suck message.
+              return Single.error(new IOException()); // TODO non-suck message.
             }
           });
     }
@@ -168,8 +167,8 @@ static final class ResultCallAdapter<T> implements CallAdapter<T> {
       return responseType;
     }
 
-    @Override public Observable<Result<T>> adapt(Call<T> call) {
-      return Observable.create(new CallOnSubscribe<>(call)) //
+    @Override public Single<Result<T>> adapt(Call<T> call) {
+      return Single.create(new CallOnSubscribe<>(call)) //
           .map(new Func1<Response<T>, Result<T>>() {
             @Override public Result<T> call(Response<T> response) {
               return Result.fromResponse(response);


Test patch:
diff --git a/retrofit-adapters/rxjava/src/test/java/retrofit/ObservableCallAdapterFactoryTest.java b/retrofit-adapters/rxjava/src/test/java/retrofit/ObservableCallAdapterFactoryTest.java
index 87eedb1192..03b78b708f 100644
--- a/retrofit-adapters/rxjava/src/test/java/retrofit/ObservableCallAdapterFactoryTest.java
+++ b/retrofit-adapters/rxjava/src/test/java/retrofit/ObservableCallAdapterFactoryTest.java
@@ -28,7 +28,7 @@
 import org.junit.Rule;
 import org.junit.Test;
 import retrofit.http.GET;
-import rx.Observable;
+import rx.Single;
 import rx.observables.BlockingObservable;
 
 import static com.squareup.okhttp.mockwebserver.SocketPolicy.DISCONNECT_AFTER_REQUEST;
@@ -39,9 +39,9 @@ public final class ObservableCallAdapterFactoryTest {
   @Rule public final MockWebServer server = new MockWebServer();
 
   interface Service {
-    @GET("/") Observable<String> body();
-    @GET("/") Observable<Response<String>> response();
-    @GET("/") Observable<Result<String>> result();
+    @GET("/") Single<String> body();
+    @GET("/") Single<Response<String>> response();
+    @GET("/") Single<Result<String>> result();
   }
 
   private Service service;
@@ -58,14 +58,14 @@ interface Service {
   @Test public void bodySuccess200() {
     server.enqueue(new MockResponse().setBody("Hi"));
 
-    BlockingObservable<String> o = service.body().toBlocking();
+    BlockingObservable<String> o = service.body().toObservable().toBlocking();
     assertThat(o.first()).isEqualTo("Hi");
   }
 
   @Test public void bodySuccess404() {
     server.enqueue(new MockResponse().setResponseCode(404));
 
-    BlockingObservable<String> o = service.body().toBlocking();
+    BlockingObservable<String> o = service.body().toObservable().toBlocking();
     try {
       o.first();
       fail();
@@ -77,7 +77,7 @@ interface Service {
   @Test public void bodyFailure() {
     server.enqueue(new MockResponse().setSocketPolicy(DISCONNECT_AFTER_REQUEST));
 
-    BlockingObservable<String> o = service.body().toBlocking();
+    BlockingObservable<String> o = service.body().toObservable().toBlocking();
     try {
       o.first();
       fail();
@@ -89,7 +89,7 @@ interface Service {
   @Test public void responseSuccess200() {
     server.enqueue(new MockResponse().setBody("Hi"));
 
-    BlockingObservable<Response<String>> o = service.response().toBlocking();
+    BlockingObservable<Response<String>> o = service.response().toObservable().toBlocking();
     Response<String> response = o.first();
     assertThat(response.isSuccess()).isTrue();
     assertThat(response.body()).isEqualTo("Hi");
@@ -98,7 +98,7 @@ interface Service {
   @Test public void responseSuccess404() throws IOException {
     server.enqueue(new MockResponse().setResponseCode(404).setBody("Hi"));
 
-    BlockingObservable<Response<String>> o = service.response().toBlocking();
+    BlockingObservable<Response<String>> o = service.response().toObservable().toBlocking();
     Response<String> response = o.first();
     assertThat(response.isSuccess()).isFalse();
     assertThat(response.errorBody().string()).isEqualTo("Hi");
@@ -107,7 +107,7 @@ interface Service {
   @Test public void responseFailure() {
     server.enqueue(new MockResponse().setSocketPolicy(DISCONNECT_AFTER_REQUEST));
 
-    BlockingObservable<Response<String>> o = service.response().toBlocking();
+    BlockingObservable<Response<String>> o = service.response().toObservable().toBlocking();
     try {
       o.first();
       fail();
@@ -119,7 +119,7 @@ interface Service {
   @Test public void resultSuccess200() {
     server.enqueue(new MockResponse().setBody("Hi"));
 
-    BlockingObservable<Result<String>> o = service.result().toBlocking();
+    BlockingObservable<Result<String>> o = service.result().toObservable().toBlocking();
     Result<String> result = o.first();
     assertThat(result.isError()).isFalse();
     Response<String> response = result.response();
@@ -130,7 +130,7 @@ interface Service {
   @Test public void resultSuccess404() throws IOException {
     server.enqueue(new MockResponse().setResponseCode(404).setBody("Hi"));
 
-    BlockingObservable<Result<String>> o = service.result().toBlocking();
+    BlockingObservable<Result<String>> o = service.result().toObservable().toBlocking();
     Result<String> result = o.first();
     assertThat(result.isError()).isFalse();
     Response<String> response = result.response();
@@ -141,7 +141,7 @@ interface Service {
   @Test public void resultFailure() {
     server.enqueue(new MockResponse().setSocketPolicy(DISCONNECT_AFTER_REQUEST));
 
-    BlockingObservable<Result<String>> o = service.result().toBlocking();
+    BlockingObservable<Result<String>> o = service.result().toObservable().toBlocking();
     Result<String> result = o.first();
     assertThat(result.isError()).isTrue();
     assertThat(result.error()).isInstanceOf(IOException.class);
@@ -149,16 +149,16 @@ interface Service {
 
   @Test public void responseType() {
     CallAdapter.Factory factory = ObservableCallAdapterFactory.create();
-    Type classType = new TypeToken<Observable<String>>() {}.getType();
+    Type classType = new TypeToken<Single<String>>() {}.getType();
     assertThat(factory.get(classType).responseType()).isEqualTo(String.class);
-    Type wilcardType = new TypeToken<Observable<? extends String>>() {}.getType();
+    Type wilcardType = new TypeToken<Single<? extends String>>() {}.getType();
     assertThat(factory.get(wilcardType).responseType()).isEqualTo(String.class);
-    Type genericType = new TypeToken<Observable<List<String>>>() {}.getType();
+    Type genericType = new TypeToken<Single<List<String>>>() {}.getType();
     assertThat(factory.get(genericType).responseType()) //
         .isEqualTo(new TypeToken<List<String>>() {}.getType());
-    Type responseType = new TypeToken<Observable<Response<String>>>() {}.getType();
+    Type responseType = new TypeToken<Single<Response<String>>>() {}.getType();
     assertThat(factory.get(responseType).responseType()).isEqualTo(String.class);
-    Type resultType = new TypeToken<Observable<Response<String>>>() {}.getType();
+    Type resultType = new TypeToken<Single<Response<String>>>() {}.getType();
     assertThat(factory.get(resultType).responseType()).isEqualTo(String.class);
   }
 
@@ -169,7 +169,7 @@ interface Service {
   }
 
   @Test public void rawTypeThrows() {
-    Type type = new TypeToken<Observable>() {}.getType();
+    Type type = new TypeToken<Single>() {}.getType();
     CallAdapter.Factory factory = ObservableCallAdapterFactory.create();
     try {
       factory.get(type);
@@ -180,7 +180,7 @@ interface Service {
   }
 
   @Test public void rawResponseTypeThrows() {
-    Type type = new TypeToken<Observable<Response>>() {}.getType();
+    Type type = new TypeToken<Single<Response>>() {}.getType();
     CallAdapter.Factory factory = ObservableCallAdapterFactory.create();
     try {
       factory.get(type);
@@ -191,7 +191,7 @@ interface Service {
   }
 
   @Test public void rawResultTypeThrows() {
-    Type type = new TypeToken<Observable<Result>>() {}.getType();
+    Type type = new TypeToken<Single<Result>>() {}.getType();
     CallAdapter.Factory factory = ObservableCallAdapterFactory.create();
     try {
       factory.get(type);