Update RxJava to 0.17.1.

* Note that we now do not return the response if the subscriber is unsubscribed.

pom.xml

@@ -51,7 +51,7 @@
     <android.platform>16</android.platform>
     <gson.version>2.2.4</gson.version>
     <okhttp.version>1.3.0</okhttp.version>
-    <rxjava.version>0.16.1</rxjava.version>
+    <rxjava.version>0.17.1</rxjava.version>
     <appengine.version>1.8.9</appengine.version>
 
     <!-- Converter Dependencies -->

retrofit-mock/src/main/java/retrofit/MockRestAdapter.java

@@ -12,9 +12,8 @@ import java.util.concurrent.TimeUnit;
 import retrofit.client.Request;
 import retrofit.client.Response;
 import rx.Observable;
-import rx.Observer;
 import rx.Scheduler;
-import rx.Subscription;
+import rx.Subscriber;
 import rx.schedulers.Schedulers;
 
 import static retrofit.RestAdapter.LogLevel;
@@ -519,15 +518,15 @@ public final class MockRestAdapter {
 
     Observable createMockObservable(final MockHandler mockHandler, final RestMethodInfo methodInfo,
         final RequestInterceptor interceptor, final Object[] args) {
-      return Observable.create(new Observable.OnSubscribeFunc<Object>() {
-        @Override public Subscription onSubscribe(Observer<? super Object> observer) {
+      return Observable.create(new Observable.OnSubscribe<Object>() {
+        @Override public void call(Subscriber<? super Object> subscriber) {
           try {
             Observable observable =
                 (Observable) mockHandler.invokeSync(methodInfo, interceptor, args);
             //noinspection unchecked
-            return observable.subscribe(observer);
+            observable.subscribe(subscriber);
           } catch (Throwable throwable) {
-            return Observable.error(throwable).subscribe(observer);
+            Observable.error(throwable).subscribe(subscriber);
           }
         }
       }).subscribeOn(scheduler);

retrofit-mock/src/test/java/retrofit/MockRestAdapterTest.java

@@ -14,7 +14,7 @@ import retrofit.client.Request;
 import retrofit.client.Response;
 import retrofit.http.GET;
 import rx.Observable;
-import rx.util.functions.Action1;
+import rx.functions.Action1;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.Assert.fail;

retrofit/src/main/java/retrofit/RestAdapter.java

@@ -39,11 +39,9 @@ import retrofit.mime.TypedByteArray;
 import retrofit.mime.TypedInput;
 import retrofit.mime.TypedOutput;
 import rx.Observable;
-import rx.Observer;
 import rx.Scheduler;
-import rx.Subscription;
+import rx.Subscriber;
 import rx.schedulers.Schedulers;
-import rx.subscriptions.Subscriptions;
 
 /**
  * Adapts a Java interface to a REST API.
@@ -237,19 +235,24 @@ public class RestAdapter {
     }
 
     Observable createRequestObservable(final Callable<ResponseWrapper> request) {
-      return Observable.create(new Observable.OnSubscribeFunc<Object>() {
-        @Override public Subscription onSubscribe(Observer<? super Object> observer) {
+      return Observable.create(new Observable.OnSubscribe<Object>() {
+        @Override public void call(Subscriber<? super Object> subscriber) {
+          if (subscriber.isUnsubscribed()) {
+            return;
+          }
           try {
             ResponseWrapper wrapper = request.call();
-            observer.onNext(wrapper.responseBody);
-            observer.onCompleted();
+            if (subscriber.isUnsubscribed()) {
+              return;
+            }
+            subscriber.onNext(wrapper.responseBody);
+            subscriber.onCompleted();
           } catch (RetrofitError e) {
-            observer.onError(e);
+            subscriber.onError(e);
           } catch (Exception e) {
             // This is from the Callable.  It shouldn't actually throw.
             throw new RuntimeException(e);
           }
-          return Subscriptions.empty();
         }
       }).subscribeOn(scheduler);
     }

retrofit/src/test/java/retrofit/RestAdapterTest.java

@@ -28,7 +28,7 @@ import retrofit.mime.TypedInput;
 import retrofit.mime.TypedOutput;
 import retrofit.mime.TypedString;
 import rx.Observable;
-import rx.util.functions.Action1;
+import rx.functions.Action1;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.fail;