Merge pull request ReactiveX#67 from omo/observeon-first

Fix observeOn() invocation order.

Author: mttkay

rxandroid/src/main/java/rx/android/content/ContentObservable.java

@@ -65,7 +65,7 @@ public Boolean call(android.support.v4.app.Fragment fragment) {
      */
     public static <T> Observable<T> bindActivity(Activity activity, Observable<T> source) {
         Assertions.assertUiThread();
-        return source.lift(new OperatorConditionalBinding<T, Activity>(activity, ACTIVITY_VALIDATOR)).observeOn(mainThread());
+        return source.observeOn(mainThread()).lift(new OperatorConditionalBinding<T, Activity>(activity, ACTIVITY_VALIDATOR));
     }
 
     /**

rxandroid/src/main/java/rx/android/content/OperatorConditionalBinding.java

@@ -15,6 +15,7 @@
 
 import rx.Observable;
 import rx.Subscriber;
+import rx.android.internal.Assertions;
 import rx.functions.Func1;
 import rx.internal.util.UtilityFunctions;
 
@@ -53,6 +54,7 @@ public Subscriber<? super T> call(final Subscriber<? super T> child) {
 
             @Override
             public void onCompleted() {
+                Assertions.assertUiThread();
                 if (shouldForwardNotification()) {
                     child.onCompleted();
                 } else {
@@ -62,6 +64,7 @@ public void onCompleted() {
 
             @Override
             public void onError(Throwable e) {
+                Assertions.assertUiThread();
                 if (shouldForwardNotification()) {
                     child.onError(e);
                 } else {
@@ -71,6 +74,7 @@ public void onError(Throwable e) {
 
             @Override
             public void onNext(T t) {
+                Assertions.assertUiThread();
                 if (shouldForwardNotification()) {
                     child.onNext(t);
                 } else {

rxandroid/src/test/java/rx/android/TestUtil.java

@@ -16,9 +16,18 @@
 import android.view.View;
 
 import org.robolectric.Robolectric;
+import org.robolectric.util.Scheduler;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executors;
+
+import rx.Observable;
+import rx.Subscriber;
 
 public class TestUtil {
 
+    static public final String STRING_EXPECTATION = "Hello";
+
     private TestUtil() {
         throw new AssertionError("Utility class");
     }
@@ -27,4 +36,20 @@ public static View createView() {
         return new View(Robolectric.application);
     }
 
+    public static Observable<String> atBackgroundThread(final CountDownLatch done) {
+        return Observable.create(new Observable.OnSubscribe<String>() {
+            @Override
+            public void call(final Subscriber<? super String> subscriber) {
+                Executors.newSingleThreadExecutor().submit(new Runnable() {
+                    @Override
+                    public void run() {
+                        subscriber.onNext(STRING_EXPECTATION);
+                        subscriber.onCompleted();
+                        done.countDown();
+                    }
+                });
+            }
+        });
+    }
+
 }

rxandroid/src/test/java/rx/android/content/ContentObservableTest.java

@@ -23,16 +23,19 @@
 import org.robolectric.Robolectric;
 import org.robolectric.RobolectricTestRunner;
 import org.robolectric.annotation.Config;
+
 import rx.Observable;
 import rx.Observer;
 import rx.android.content.ContentObservable;
+import rx.android.TestUtil;
 import rx.observers.TestObserver;
 
 import android.app.Activity;
 import android.app.Fragment;
 import android.support.v4.app.FragmentActivity;
 
 import java.util.concurrent.Callable;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
@@ -101,6 +104,30 @@ public Object call() throws Exception {
         }
     }
 
+    @Test
+    public void bindFragmentToSourceFromDifferentThread() throws InterruptedException {
+        CountDownLatch done = new CountDownLatch(1);
+        ContentObservable.bindFragment(fragment, TestUtil.atBackgroundThread(done)).subscribe(new TestObserver<String>(observer));
+        done.await();
+
+        Robolectric.runUiThreadTasksIncludingDelayedTasks();
+
+        verify(observer).onNext(TestUtil.STRING_EXPECTATION);
+        verify(observer).onCompleted();
+    }
+
+    @Test
+    public void bindSupportFragmentToSourceFromDifferentThread() throws InterruptedException {
+        CountDownLatch done = new CountDownLatch(1);
+        ContentObservable.bindFragment(supportFragment, TestUtil.atBackgroundThread(done)).subscribe(new TestObserver<String>(observer));
+        done.await();
+
+        Robolectric.runUiThreadTasksIncludingDelayedTasks();
+
+        verify(observer).onNext(TestUtil.STRING_EXPECTATION);
+        verify(observer).onCompleted();
+    }
+
     @Test(expected = IllegalStateException.class)
     public void itThrowsIfObserverCallsFromActivityFromBackgroundThread() throws Throwable {
         final Future<Object> future = Executors.newSingleThreadExecutor().submit(new Callable<Object>() {
@@ -116,4 +143,18 @@ public Object call() throws Exception {
             throw e.getCause();
         }
     }
+
+    @Test
+    public void bindActivityToSourceFromDifferentThread() throws InterruptedException {
+        CountDownLatch done = new CountDownLatch(1);
+        ContentObservable.bindActivity(activity, TestUtil.atBackgroundThread(done)).subscribe(new TestObserver<String>(observer));
+        done.await();
+
+        Robolectric.runUiThreadTasksIncludingDelayedTasks();
+
+        verify(observer).onNext(TestUtil.STRING_EXPECTATION);
+        verify(observer).onCompleted();
+    }
+
+
 }

rxandroid/src/test/java/rx/android/view/BindViewTest.java

@@ -16,8 +16,11 @@
 import org.robolectric.RobolectricTestRunner;
 import org.robolectric.annotation.Config;
 
+import java.util.concurrent.CountDownLatch;
+
 import rx.Observer;
 import rx.Subscription;
+import rx.android.TestUtil;
 import rx.subjects.PublishSubject;
 
 @RunWith(RobolectricTestRunner.class)
@@ -119,6 +122,18 @@ public void earlyUnsubscribeStopsNotifications() {
         verifyNoMoreInteractions(observer);
     }
 
+    @Test
+    public void bindViewInDifferentThread() throws InterruptedException {
+        CountDownLatch done = new CountDownLatch(1);
+        ViewObservable.bindView(target, TestUtil.atBackgroundThread(done)).subscribe(observer);
+        done.await();
+
+        Robolectric.runUiThreadTasksIncludingDelayedTasks();
+
+        verify(observer).onNext(TestUtil.STRING_EXPECTATION);
+        verify(observer).onCompleted();
+    }
+
     @Test(expected = IllegalArgumentException.class)
     public void nullViewIsNotAllowed() {
         ViewObservable.bindView(null, subject);