Repository: square/retrofit
Issue: Retrofit-Mock chained observable scheduling problem

There seems to be a bug in the MockRestAdapter$MockRxSupport:createMockObservable private class. The way the scheduling is done with respect to subscribing the subscriber to the observable seems wrong. Subscribing to the final observable comes after in the HttpExecutor thread itself is started. I believe the original flow which comes in most cases would come from Schedulers.io() thread is completed and unsubscribed before the mockHandler.invokeSync returned Observable can be subscribed to.

This causes multiple Retrofit Observable chained together to be unable to complete properly.

Using the new Schedulers.from() added in rxjava 0.19.\* seems to solve the issue when passed the httpExecutor in conjunction with using subscribeOn on the mocked observable returned by the mockHandler.invokeSync.

Observable observable =
                (Observable) mockHandler.invokeSync(methodInfo, interceptor, args);
observable.subscribeOn(Schedulers.from(httpExecutor));
observable.subscribe(subscriber);

This Stackoverflow question is where the issue was found from http://stackoverflow.com/questions/24748656/chaining-retrofit-services-w-rxjava-support


Issue comments:
I found a better way to fix this which allows my chained observables to run on different schedulers:

```
  private static class MockRxSupport {
      private final ErrorHandler errorHandler;

      MockRxSupport(RestAdapter restAdapter) {
          errorHandler = restAdapter.errorHandler;
      }

      Observable createMockObservable(final MockHandler mockHandler, final RestMethodInfo methodInfo,
                                  final RequestInterceptor interceptor, final Object[] args) {
          return Observable.create(new Observable.OnSubscribe<Object>() {
              @Override
              public void call(final Subscriber<? super Object> subscriber) {
                  try {
                      if (subscriber.isUnsubscribed()) return;
                      Observable observable =
                              (Observable) mockHandler.invokeSync(methodInfo, interceptor, args);

                      //noinspection unchecked
                      observable.subscribe(new Subscriber<Object>() {
                          @Override public void onCompleted() {
                              subscriber.onCompleted();
                          }
                          @Override public void onError(Throwable e) {
                              subscriber.onError(e);
                          }
                          @Override public void onNext(Object value) {
                              subscriber.onNext(value);
                          }
                      });

                  } catch (RetrofitError e) {
                      subscriber.onError(errorHandler.handleError(e));
                  } catch (Throwable e) {
                      subscriber.onError(e);
                  }
              }
          });
      }
  }
```

Is this what I am running into? Works fine running against the real server, but neither success nor failure gets called on the subscriber running in mock mode.

```
mApi.replyToken(mParentId)
        .flatMap(new Func1<ReplyToken, Observable<Response>>() {
          @Override public Observable<Response> call(ReplyToken replyToken) {
            Timber.d("replyToken = " + replyToken);
            return mApi.reply(mParentId, replyToken.getHmac(), reply);
          }
        })
        .observeOn(AndroidSchedulers.mainThread())
        .subscribeOn(Schedulers.io())
        .subscribe(new Action1<Response>() {
          @Override public void call(Response response) {
            Timber.e("Error: when replying, success is indicated by a 302 with a redirect.");
            onReplyFailed();
          }
        }, new Action1<Throwable>() {
          @Override public void call(Throwable throwable) {
            handleThrowable(throwable);
          }
        });
```

Anyone know of a workaround or a fix? 

Try using a synchronous Executor in mock mode and see if it help.  See the example here http://stackoverflow.com/questions/24748656/chaining-retrofit-services-w-rxjava-support?answertab=votes#tab-top

Ok, that worked. Thanks!


Pull request review comments:
unused?

is this intentional?  seems like the test used to confirm that the http call was made and now it doesn't?

you might be right. the delay still exists without the executor, but i think it happens synchronously now. i was on the wrong end of my balmer peak...

How did this even work before? The test wasn't correct at all.


Patch that resolved the issue:
diff --git a/retrofit-mock/src/main/java/retrofit/MockRestAdapter.java b/retrofit-mock/src/main/java/retrofit/MockRestAdapter.java
index 75026d8834..20e7a639c3 100644
--- a/retrofit-mock/src/main/java/retrofit/MockRestAdapter.java
+++ b/retrofit-mock/src/main/java/retrofit/MockRestAdapter.java
@@ -8,12 +8,13 @@
 import java.lang.reflect.Proxy;
 import java.util.Map;
 import java.util.Random;
-import java.util.concurrent.Executor;
 import java.util.concurrent.TimeUnit;
 import retrofit.client.Request;
 import retrofit.client.Response;
 import rx.Observable;
-import rx.Subscriber;
+import rx.Scheduler;
+import rx.functions.Func1;
+import rx.schedulers.Schedulers;
 
 import static retrofit.RestAdapter.LogLevel;
 import static retrofit.RetrofitError.unexpectedError;
@@ -468,36 +469,28 @@ private static long uptimeMillis() {
 
   /** Indirection to avoid VerifyError if RxJava isn't present. */
   private static class MockRxSupport {
-    private final Executor httpExecutor;
+    private final Scheduler httpScheduler;
     private final ErrorHandler errorHandler;
 
     MockRxSupport(RestAdapter restAdapter) {
-      httpExecutor = restAdapter.httpExecutor;
+      httpScheduler = Schedulers.from(restAdapter.httpExecutor);
       errorHandler = restAdapter.errorHandler;
     }
 
     Observable createMockObservable(final MockHandler mockHandler, final RestMethodInfo methodInfo,
         final RequestInterceptor interceptor, final Object[] args) {
-      return Observable.create(new Observable.OnSubscribe<Object>() {
-        @Override public void call(final Subscriber<? super Object> subscriber) {
-          if (subscriber.isUnsubscribed()) return;
-          httpExecutor.execute(new Runnable() {
-            @Override public void run() {
+      return Observable.just("nothing") //
+          .flatMap(new Func1<String, Observable<?>>() {
+            @Override public Observable<?> call(String s) {
               try {
-                if (subscriber.isUnsubscribed()) return;
-                Observable observable =
-                        (Observable) mockHandler.invokeSync(methodInfo, interceptor, args);
-                //noinspection unchecked
-                observable.subscribe(subscriber);
+                return (Observable) mockHandler.invokeSync(methodInfo, interceptor, args);
               } catch (RetrofitError e) {
-                subscriber.onError(errorHandler.handleError(e));
-              } catch (Throwable e) {
-                subscriber.onError(e);
+                return Observable.error(errorHandler.handleError(e));
+              } catch (Throwable throwable) {
+                return Observable.error(throwable);
               }
             }
-          });
-        }
-      });
+          }).subscribeOn(httpScheduler);
     }
   }
 }


Test patch:
diff --git a/retrofit-mock/src/test/java/retrofit/MockRestAdapterTest.java b/retrofit-mock/src/test/java/retrofit/MockRestAdapterTest.java
index 412cf7d5ba..680ffdfffc 100644
--- a/retrofit-mock/src/test/java/retrofit/MockRestAdapterTest.java
+++ b/retrofit-mock/src/test/java/retrofit/MockRestAdapterTest.java
@@ -477,7 +477,7 @@ class MockObservableExample implements ObservableExample {
       }
     });
 
-    verify(httpExecutor, atLeastOnce()).execute(any(Runnable.class));
+    verify(httpExecutor).execute(any(Runnable.class));
     verifyZeroInteractions(callbackExecutor);
 
     RetrofitError error = errorRef.get();
@@ -518,7 +518,7 @@ class MockObservableExample implements ObservableExample {
       }
     });
 
-    verify(httpExecutor, atLeastOnce()).execute(any(Runnable.class));
+    verify(httpExecutor).execute(any(Runnable.class));
     verifyZeroInteractions(callbackExecutor);
 
     RetrofitError error = errorRef.get();