Merge pull request square#578 from square/jw/mock-scheduling

Simplify scheduling of mock observers.
up2dream · Oct 7, 2014 · 702e005 · 702e005
2 parents 60b4970 + 4cf02b2
commit 702e005
Show file tree

Hide file tree

Showing 2 changed files with 15 additions and 22 deletions.
diff --git 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);
     }
   }
 }
diff --git 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();