Merge pull request #2721 from adriancole/observableAssembly

Makes sure observable assembly hooks are called

retrofit-adapters/rxjava2/src/main/java/retrofit2/adapter/rxjava2/RxJava2CallAdapter.java

@@ -18,6 +18,7 @@ package retrofit2.adapter.rxjava2;
 import io.reactivex.BackpressureStrategy;
 import io.reactivex.Observable;
 import io.reactivex.Scheduler;
+import io.reactivex.plugins.RxJavaPlugins;
 import java.lang.reflect.Type;
 import javax.annotation.Nullable;
 import retrofit2.Call;
@@ -83,6 +84,6 @@ final class RxJava2CallAdapter<R> implements CallAdapter<R, Object> {
     if (isCompletable) {
       return observable.ignoreElements();
     }
-    return observable;
+    return RxJavaPlugins.onAssembly(observable);
   }
 }

retrofit-adapters/rxjava2/src/test/java/retrofit2/adapter/rxjava2/ObservableTest.java

@@ -16,6 +16,8 @@
 package retrofit2.adapter.rxjava2;
 
 import io.reactivex.Observable;
+import io.reactivex.functions.Function;
+import io.reactivex.plugins.RxJavaPlugins;
 import java.io.IOException;
 import okhttp3.mockwebserver.MockResponse;
 import okhttp3.mockwebserver.MockWebServer;
@@ -133,4 +135,18 @@ public final class ObservableTest {
     assertThat(result.error()).isInstanceOf(IOException.class);
     observer.assertComplete();
   }
+
+  @Test public void observableAssembly() {
+    try {
+      final Observable<String> justMe = Observable.just("me");
+      RxJavaPlugins.setOnObservableAssembly(new Function<Observable, Observable>() {
+        @Override public Observable apply(Observable f) {
+          return justMe;
+        }
+      });
+      assertThat(service.body()).isEqualTo(justMe);
+    } finally {
+      RxJavaPlugins.reset();
+    }
+  }
 }