Fix observeOn() invocation order.

It should be called before lift() so that the operator can observe it on the main thread.
This change also adds some test for other bindXx() variants to get better coverage.

Author: omo

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);