Repository: square/retrofit
Issue: Facilitate trace propagation with RxJava2

What kind of issue is this?
 - [x] Feature Request. Start by telling us what problem you’re trying to solve. Often a solution
       already exists! Don’t send pull requests to implement new features without first getting our
       support. Sometimes we leave features out on purpose to keep the project small.

I'm trying to help get tracing working for retrofit. For example, continuing an incoming trace during an retrofit outbound request (including but not limited to SLF4J propagation).

I looked at similar tech https://github.com/akaita/RxJava2Debug and assembly hooks look quite good for this. I started testing and noticed everything works nicely, except when the result type is Observable. That's because observable types are returned as-is:

```java
final class RxJava2CallAdapter<R> implements CallAdapter<R, Object> {
 --snip--
    if (isCompletable) {
      return observable.ignoreElements(); // .ignoreElements indirectly calls RxJavaPlugins.onAssembly
    }
    return observable; // returned with no hooks
  }
```

I understand that a custom adapter could do this, but wondering if we could make it default to use the hook on plain observable.

ex.
```java
final class RxJava2CallAdapter<R> implements CallAdapter<R, Object> {
 --snip--
    if (isCompletable) {
      return observable.ignoreElements();
    }
    return RxJavaPlugins.onAssembly(observable);
  }
```

Open to other ideas to make this drop-in

Issue comments:
Seems fine to me. PR?
👍 

Whenever custom Observable type is instantiated, it should go through `RxJavaPlugins.onAssembly()`
https://github.com/square/retrofit/pull/2721 thx for the quick consideration folks!

Pull request review comments:
Hm, I thought only custom observable instantiation will be wrapped:

```java
Observable<Response<R>> responseObservable = RxJavaPlugins.onAssembly(isAsync
        ? new CallEnqueueObservable<>(call)
        : new CallExecuteObservable<>(call));
```

```java
if (isResult) {
  observable = RxJavaPlugins.onAssembly(new ResultObservable<>(responseObservable));
} else if (isBody) {
  observable = RxJavaPlugins.onAssembly(new BodyObservable<>(responseObservable));
} else {
  observable = responseObservable;
}
```

Current implementation is not bad, but:

- As far as I see stactrace won't point to specific lines where `CallEnqueueObservable`/`CallExecuteObservable`/`ResultObservable`/`BodyObservable` got intantiated but will only point to line 87 where it can be hard to figure which branch of code was executed and what is the original observable type if `subscribeOn` was applied
- Wrapping happens after potential `subscribeOn`. Retrofit Observable types are package-private which means that reflecting on their internals is not part of public api provided by Retrofit, however it seems legal to check for `package` of the Observable type to apply some custom behavior. Since wrapping happens after `subscribeOn`, type information gets hidden.
I've made a branch with more fine-grained wrapping that we can turn into PR if you want 

https://github.com/square/retrofit/compare/master...artem-zinnatullin:az/rxjava-adapter-onAssembly-fix?expand=1

Patch that resolved the issue:
diff --git a/retrofit-adapters/rxjava2/src/main/java/retrofit2/adapter/rxjava2/RxJava2CallAdapter.java b/retrofit-adapters/rxjava2/src/main/java/retrofit2/adapter/rxjava2/RxJava2CallAdapter.java
index eccd30da36..254f921ad5 100644
--- a/retrofit-adapters/rxjava2/src/main/java/retrofit2/adapter/rxjava2/RxJava2CallAdapter.java
+++ b/retrofit-adapters/rxjava2/src/main/java/retrofit2/adapter/rxjava2/RxJava2CallAdapter.java
@@ -18,6 +18,7 @@
 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);
   }
 }


Test patch:
diff --git a/retrofit-adapters/rxjava2/src/test/java/retrofit2/adapter/rxjava2/ObservableTest.java b/retrofit-adapters/rxjava2/src/test/java/retrofit2/adapter/rxjava2/ObservableTest.java
index e421c16526..23c20f77e3 100644
--- a/retrofit-adapters/rxjava2/src/test/java/retrofit2/adapter/rxjava2/ObservableTest.java
+++ b/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 @@ interface Service {
     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();
+    }
+  }
 }