ReactiveX · benjchristensen · Feb 18, 2014 · Feb 18, 2014
diff --git a/rxjava-contrib/rxjava-android/src/main/java/rx/android/observables/AndroidObservable.java b/rxjava-contrib/rxjava-android/src/main/java/rx/android/observables/AndroidObservable.java
@@ -16,7 +16,7 @@
 package rx.android.observables;
 
 import rx.Observable;
-import rx.operators.OperationObserveFromAndroidComponent;
+import rx.operators.OperatorObserveFromAndroidComponent;
 import android.app.Activity;
 import android.app.Fragment;
 import android.os.Build;
@@ -59,7 +59,7 @@ private AndroidObservable() {}
      * @return a new observable sequence that will emit notifications on the main UI thread
      */
     public static <T> Observable<T> fromActivity(Activity activity, Observable<T> sourceObservable) {
-        return OperationObserveFromAndroidComponent.observeFromAndroidComponent(sourceObservable, activity);
+        return OperatorObserveFromAndroidComponent.observeFromAndroidComponent(sourceObservable, activity);
     }
 
     /**
@@ -88,9 +88,9 @@ public static <T> Observable<T> fromActivity(Activity activity, Observable<T> so
      */
     public static <T> Observable<T> fromFragment(Object fragment, Observable<T> sourceObservable) {
         if (USES_SUPPORT_FRAGMENTS && fragment instanceof android.support.v4.app.Fragment) {
-            return OperationObserveFromAndroidComponent.observeFromAndroidComponent(sourceObservable, (android.support.v4.app.Fragment) fragment);
+            return OperatorObserveFromAndroidComponent.observeFromAndroidComponent(sourceObservable, (android.support.v4.app.Fragment) fragment);
         } else if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.HONEYCOMB && fragment instanceof Fragment) {
-            return OperationObserveFromAndroidComponent.observeFromAndroidComponent(sourceObservable, (Fragment) fragment);
+            return OperatorObserveFromAndroidComponent.observeFromAndroidComponent(sourceObservable, (Fragment) fragment);
         } else {
             throw new IllegalArgumentException("Target fragment is neither a native nor support library Fragment");
         }

diff --git a/...a-contrib/rxjava-android/src/main/java/rx/android/subscriptions/AndroidSubscriptions.java b/...a-contrib/rxjava-android/src/main/java/rx/android/subscriptions/AndroidSubscriptions.java
@@ -0,0 +1,55 @@
+/**
+ * Copyright 2014 Netflix, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package rx.android.subscriptions;
+
+import rx.Scheduler.Inner;
+import rx.Subscription;
+import rx.android.schedulers.AndroidSchedulers;
+import rx.functions.Action0;
+import rx.functions.Action1;
+import rx.subscriptions.Subscriptions;
+import android.os.Looper;
+
+public final class AndroidSubscriptions {
+
+    private AndroidSubscriptions() {
+        // no instance
+    }
+
+    /**
+     * Create an Subscription that always runs <code>unsubscribe</code> in the UI thread.
+     * 
+     * @param unsubscribe
+     * @return an Subscription that always runs <code>unsubscribe</code> in the UI thread.
+     */
+    public static Subscription unsubscribeInUiThread(final Action0 unsubscribe) {
+        return Subscriptions.create(new Action0() {
+            @Override
+            public void call() {
+                if (Looper.getMainLooper() == Looper.myLooper()) {
+                    unsubscribe.call();
+                } else {
+                    AndroidSchedulers.mainThread().schedule(new Action1<Inner>() {
+                        @Override
+                        public void call(Inner inner) {
+                            unsubscribe.call();
+                        }
+                    });
+                }
+            }
+        });
+    }
+}
diff --git a/rxjava-contrib/rxjava-android/src/main/java/rx/operators/OperatorCompoundButtonInput.java b/rxjava-contrib/rxjava-android/src/main/java/rx/operators/OperatorCompoundButtonInput.java
@@ -23,12 +23,9 @@
 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.android.subscriptions.AndroidSubscriptions;
 import rx.functions.Action0;
-import rx.functions.Action1;
-import rx.subscriptions.Subscriptions;
 import android.view.View;
 import android.widget.CompoundButton;
 
@@ -53,17 +50,10 @@ public void onCheckedChanged(final CompoundButton button, final boolean checked)
             }
         };
 
-        final Subscription subscription = Subscriptions.create(new Action0() {
+        final Subscription subscription = AndroidSubscriptions.unsubscribeInUiThread(new Action0() {
             @Override
             public void call() {
-                AndroidSchedulers.mainThread().schedule(new Action1<Inner>() {
-
-                    @Override
-                    public void call(Inner t1) {
-                        composite.removeOnCheckedChangeListener(listener);
-                    }
-
-                });
+                composite.removeOnCheckedChangeListener(listener);
             }
         });
 

diff --git a/rxjava-contrib/rxjava-android/src/main/java/rx/operators/OperatorEditTextInput.java b/rxjava-contrib/rxjava-android/src/main/java/rx/operators/OperatorEditTextInput.java
@@ -18,12 +18,9 @@
 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.android.subscriptions.AndroidSubscriptions;
 import rx.functions.Action0;
-import rx.functions.Action1;
-import rx.subscriptions.Subscriptions;
 import android.text.Editable;
 import android.text.TextWatcher;
 import android.widget.EditText;
@@ -47,17 +44,10 @@ public void afterTextChanged(final Editable editable) {
             }
         };
 
-        final Subscription subscription = Subscriptions.create(new Action0() {
+        final Subscription subscription = AndroidSubscriptions.unsubscribeInUiThread(new Action0() {
             @Override
             public void call() {
-                AndroidSchedulers.mainThread().schedule(new Action1<Inner>() {
-
-                    @Override
-                    public void call(Inner t1) {
-                        input.removeTextChangedListener(watcher);
-                    }
-
-                });
+                input.removeTextChangedListener(watcher);
             }
         });
 

diff --git a/...OperationObserveFromAndroidComponent.java → .../OperatorObserveFromAndroidComponent.java b/...OperationObserveFromAndroidComponent.java → .../OperatorObserveFromAndroidComponent.java
@@ -17,17 +17,15 @@
 
 import rx.Observable;
 import rx.Observer;
-import rx.Subscription;
-import rx.Scheduler.Inner;
+import rx.Subscriber;
 import rx.android.schedulers.AndroidSchedulers;
+import rx.android.subscriptions.AndroidSubscriptions;
 import rx.functions.Action0;
-import rx.functions.Action1;
-import rx.subscriptions.Subscriptions;
 import android.app.Activity;
 import android.os.Looper;
 import android.util.Log;
 
-public class OperationObserveFromAndroidComponent {
+public class OperatorObserveFromAndroidComponent {
 
     public static <T> Observable<T> observeFromAndroidComponent(Observable<T> source, android.app.Fragment fragment) {
         return Observable.create(new OnSubscribeFragment<T>(source, fragment));
@@ -41,7 +39,7 @@ public static <T> Observable<T> observeFromAndroidComponent(Observable<T> source
         return Observable.create(new OnSubscribeBase<T, Activity>(source, activity));
     }
 
-    private static class OnSubscribeBase<T, AndroidComponent> implements Observable.OnSubscribeFunc<T> {
+    private static class OnSubscribeBase<T, AndroidComponent> implements Observable.OnSubscribe<T> {
 
         private static final String LOG_TAG = "AndroidObserver";
 
@@ -67,10 +65,10 @@ protected boolean isComponentValid(AndroidComponent component) {
         }
 
         @Override
-        public Subscription onSubscribe(Observer<? super T> observer) {
+        public void call(Subscriber<? super T> subscriber) {
             assertUiThread();
-            observerRef = observer;
-            final Subscription sourceSub = source.observeOn(AndroidSchedulers.mainThread()).subscribe(new Observer<T>() {
+            observerRef = subscriber;
+            source.observeOn(AndroidSchedulers.mainThread()).subscribe(new Subscriber<T>(subscriber) {
                 @Override
                 public void onCompleted() {
                     if (componentRef != null && isComponentValid(componentRef)) {
@@ -98,21 +96,13 @@ public void onNext(T args) {
                     }
                 }
             });
-            return Subscriptions.create(new Action0() {
+            subscriber.add(AndroidSubscriptions.unsubscribeInUiThread(new Action0() {
                 @Override
                 public void call() {
                     log("unsubscribing from source sequence");
-                    AndroidSchedulers.mainThread().schedule(new Action1<Inner>() {
-
-                        @Override
-                        public void call(Inner t1) {
-                            releaseReferences();
-                            sourceSub.unsubscribe();
-                        }
-
-                    });
+                    releaseReferences();
                 }
-            });
+            }));
         }
 
         private void releaseReferences() {

diff --git a/rxjava-contrib/rxjava-android/src/main/java/rx/operators/OperatorViewClick.java b/rxjava-contrib/rxjava-android/src/main/java/rx/operators/OperatorViewClick.java
@@ -21,14 +21,11 @@
 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.android.subscriptions.AndroidSubscriptions;
 import rx.functions.Action0;
-import rx.functions.Action1;
-import rx.subscriptions.Subscriptions;
 import android.view.View;
 
 public final class OperatorViewClick implements Observable.OnSubscribe<View> {
@@ -52,17 +49,10 @@ public void onClick(final View clicked) {
             }
         };
 
-        final Subscription subscription = Subscriptions.create(new Action0() {
+        final Subscription subscription = AndroidSubscriptions.unsubscribeInUiThread(new Action0() {
             @Override
             public void call() {
-                AndroidSchedulers.mainThread().schedule(new Action1<Inner>() {
-
-                    @Override
-                    public void call(Inner t1) {
-                        composite.removeOnClickListener(listener);
-                    }
-
-                });
+                composite.removeOnClickListener(listener);
             }
         });