Merge pull request #880 from zsxwing/android-unsubscribe-thread-safe

Force ViewObservable be subscribed and unsubscribed in the UI thread

Author: benjchristensen

rxjava-contrib/rxjava-android/src/main/java/rx/android/observables/ViewObservable.java

@@ -15,6 +15,7 @@
  */
 package rx.android.observables;
 
+import android.os.Looper;
 import android.view.View;
 import android.widget.CompoundButton;
 import android.widget.EditText;
@@ -37,5 +38,10 @@ public static Observable<Boolean> input(final CompoundButton button, final boole
         return Observable.create(new OperatorCompoundButtonInput(button, emitInitialValue));
     }
 
+    public static void assertUiThread() {
+        if (Looper.getMainLooper() != Looper.myLooper()) {
+            throw new IllegalStateException("Observers must subscribe from the main UI thread, but was " + Thread.currentThread());
+        }
+    }
 }
 

rxjava-contrib/rxjava-android/src/main/java/rx/operators/OperationObserveFromAndroidComponent.java

@@ -18,9 +18,11 @@
 import rx.Observable;
 import rx.Observer;
 import rx.Subscription;
+import rx.Scheduler.Inner;
 import rx.android.schedulers.AndroidSchedulers;
 import rx.subscriptions.Subscriptions;
 import rx.util.functions.Action0;
+import rx.util.functions.Action1;
 import android.app.Activity;
 import android.os.Looper;
 import android.util.Log;
@@ -100,8 +102,15 @@ public void onNext(T args) {
                 @Override
                 public void call() {
                     log("unsubscribing from source sequence");
-                    releaseReferences();
-                    sourceSub.unsubscribe();
+                    AndroidSchedulers.mainThread().schedule(new Action1<Inner>() {
+
+                        @Override
+                        public void call(Inner t1) {
+                            releaseReferences();
+                            sourceSub.unsubscribe();
+                        }
+
+                    });
                 }
             });
         }

rxjava-contrib/rxjava-android/src/main/java/rx/operators/OperatorCompoundButtonInput.java

@@ -23,8 +23,12 @@
 import rx.Observable;
 import rx.Subscriber;
 import rx.Subscription;
+import rx.Scheduler.Inner;
+import rx.android.observables.ViewObservable;
+import rx.android.schedulers.AndroidSchedulers;
 import rx.subscriptions.Subscriptions;
 import rx.util.functions.Action0;
+import rx.util.functions.Action1;
 import android.view.View;
 import android.widget.CompoundButton;
 
@@ -39,6 +43,7 @@ public OperatorCompoundButtonInput(final CompoundButton button, final boolean em
 
     @Override
     public void call(final Subscriber<? super Boolean> observer) {
+        ViewObservable.assertUiThread();
         final CompositeOnCheckedChangeListener composite = CachedListeners.getFromViewOrCreate(button);
 
         final CompoundButton.OnCheckedChangeListener listener = new CompoundButton.OnCheckedChangeListener() {
@@ -51,7 +56,14 @@ public void onCheckedChanged(final CompoundButton button, final boolean checked)
         final Subscription subscription = Subscriptions.create(new Action0() {
             @Override
             public void call() {
-                composite.removeOnCheckedChangeListener(listener);
+                AndroidSchedulers.mainThread().schedule(new Action1<Inner>() {
+
+                    @Override
+                    public void call(Inner t1) {
+                        composite.removeOnCheckedChangeListener(listener);
+                    }
+
+                });
             }
         });
 

rxjava-contrib/rxjava-android/src/main/java/rx/operators/OperatorEditTextInput.java

@@ -18,8 +18,12 @@
 import rx.Observable;
 import rx.Subscriber;
 import rx.Subscription;
+import rx.Scheduler.Inner;
+import rx.android.observables.ViewObservable;
+import rx.android.schedulers.AndroidSchedulers;
 import rx.subscriptions.Subscriptions;
 import rx.util.functions.Action0;
+import rx.util.functions.Action1;
 import android.text.Editable;
 import android.text.TextWatcher;
 import android.widget.EditText;
@@ -35,6 +39,7 @@ public OperatorEditTextInput(final EditText input, final boolean emitInitialValu
 
     @Override
     public void call(final Subscriber<? super String> observer) {
+        ViewObservable.assertUiThread();
         final TextWatcher watcher = new SimpleTextWatcher() {
             @Override
             public void afterTextChanged(final Editable editable) {
@@ -45,7 +50,14 @@ public void afterTextChanged(final Editable editable) {
         final Subscription subscription = Subscriptions.create(new Action0() {
             @Override
             public void call() {
-                input.removeTextChangedListener(watcher);
+                AndroidSchedulers.mainThread().schedule(new Action1<Inner>() {
+
+                    @Override
+                    public void call(Inner t1) {
+                        input.removeTextChangedListener(watcher);
+                    }
+
+                });
             }
         });
 

rxjava-contrib/rxjava-android/src/main/java/rx/operators/OperatorViewClick.java

@@ -21,10 +21,14 @@
 import java.util.WeakHashMap;
 
 import rx.Observable;
+import rx.Scheduler.Inner;
 import rx.Subscriber;
 import rx.Subscription;
+import rx.android.observables.ViewObservable;
+import rx.android.schedulers.AndroidSchedulers;
 import rx.subscriptions.Subscriptions;
 import rx.util.functions.Action0;
+import rx.util.functions.Action1;
 import android.view.View;
 
 public final class OperatorViewClick implements Observable.OnSubscribe<View> {
@@ -38,6 +42,7 @@ public OperatorViewClick(final View view, final boolean emitInitialValue) {
 
     @Override
     public void call(final Subscriber<? super View> observer) {
+        ViewObservable.assertUiThread();
         final CompositeOnClickListener composite = CachedListeners.getFromViewOrCreate(view);
 
         final View.OnClickListener listener = new View.OnClickListener() {
@@ -50,7 +55,14 @@ public void onClick(final View clicked) {
         final Subscription subscription = Subscriptions.create(new Action0() {
             @Override
             public void call() {
-                composite.removeOnClickListener(listener);
+                AndroidSchedulers.mainThread().schedule(new Action1<Inner>() {
+
+                    @Override
+                    public void call(Inner t1) {
+                        composite.removeOnClickListener(listener);
+                    }
+
+                });
             }
         });