First implementation of OperatorWeakBinding

Author: mttkay

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

@@ -15,8 +15,12 @@
  */
 package rx.android.observables;
 
+import static rx.android.schedulers.AndroidSchedulers.mainThread;
+
 import rx.Observable;
 import rx.operators.OperatorObserveFromAndroidComponent;
+import rx.operators.OperatorWeakBinding;
+
 import android.app.Activity;
 import android.app.Fragment;
 import android.os.Build;
@@ -58,6 +62,7 @@ private AndroidObservable() {}
      * @param <T>
      * @return a new observable sequence that will emit notifications on the main UI thread
      */
+    @Deprecated
     public static <T> Observable<T> fromActivity(Activity activity, Observable<T> sourceObservable) {
         return OperatorObserveFromAndroidComponent.observeFromAndroidComponent(sourceObservable, activity);
     }
@@ -86,6 +91,7 @@ public static <T> Observable<T> fromActivity(Activity activity, Observable<T> so
      * @param <T>
      * @return a new observable sequence that will emit notifications on the main UI thread
      */
+    @Deprecated
     public static <T> Observable<T> fromFragment(Object fragment, Observable<T> sourceObservable) {
         if (USES_SUPPORT_FRAGMENTS && fragment instanceof android.support.v4.app.Fragment) {
             return OperatorObserveFromAndroidComponent.observeFromAndroidComponent(sourceObservable, (android.support.v4.app.Fragment) fragment);
@@ -95,4 +101,21 @@ public static <T> Observable<T> fromFragment(Object fragment, Observable<T> sour
             throw new IllegalArgumentException("Target fragment is neither a native nor support library Fragment");
         }
     }
+
+    public static <T> Observable<T> bindActivity(Activity activity, Observable<T> cachedSequence) {
+        return cachedSequence.observeOn(mainThread()).lift(new OperatorWeakBinding<T, Activity>(activity));
+    }
+
+    public static <T> Observable<T> bindFragment(Object fragment, Observable<T> cachedSequence) {
+        Observable<T> source = cachedSequence.observeOn(mainThread());
+        if (USES_SUPPORT_FRAGMENTS && fragment instanceof android.support.v4.app.Fragment) {
+            android.support.v4.app.Fragment f = (android.support.v4.app.Fragment) fragment;
+            return source.lift(new OperatorWeakBinding<T, android.support.v4.app.Fragment>(f));
+        } else if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.HONEYCOMB && fragment instanceof Fragment) {
+            Fragment f = (Fragment) fragment;
+            return source.lift(new OperatorWeakBinding<T, Fragment>(f));
+        } else {
+            throw new IllegalArgumentException("Target fragment is neither a native nor support library Fragment");
+        }
+    }
 }

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

@@ -25,6 +25,7 @@
 import android.os.Looper;
 import android.util.Log;
 
+@Deprecated
 public class OperatorObserveFromAndroidComponent {
 
     public static <T> Observable<T> observeFromAndroidComponent(Observable<T> source, android.app.Fragment fragment) {

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

@@ -0,0 +1,75 @@
+package rx.operators;
+
+import rx.Observable;
+import rx.Subscriber;
+
+import android.util.Log;
+
+import java.lang.ref.WeakReference;
+
+public final class OperatorWeakBinding<T, R> implements Observable.Operator<T, T> {
+
+    private static final String LOG_TAG = "WeakBinding";
+
+    private final WeakReference<R> boundRef;
+
+    public OperatorWeakBinding(R bound) {
+        boundRef = new WeakReference<R>(bound);
+    }
+
+    @Override
+    public Subscriber<? super T> call(final Subscriber<? super T> child) {
+        return new WeakSubscriber<T, R>(child, boundRef);
+    }
+
+    private static final class WeakSubscriber<T, R> extends Subscriber<T> {
+
+        private final WeakReference<Subscriber<? super T>> subscriberRef;
+        private final WeakReference<R> boundRef;
+
+        private WeakSubscriber(Subscriber<? super T> op, WeakReference<R> boundRef) {
+            subscriberRef = new WeakReference<Subscriber<? super T>>(op);
+            this.boundRef = boundRef;
+        }
+
+        @Override
+        public void onCompleted() {
+            Subscriber<? super T> sub = subscriberRef.get();
+            if (sub != null && boundRef.get() != null) {
+                sub.onCompleted();
+            } else {
+                handleLostBinding(sub, "onCompleted");
+            }
+        }
+
+        @Override
+        public void onError(Throwable e) {
+            Subscriber<? super T> sub = subscriberRef.get();
+            if (sub != null && boundRef.get() != null) {
+                sub.onError(e);
+            } else {
+                handleLostBinding(sub, "onError");
+            }
+        }
+
+        @Override
+        public void onNext(T t) {
+            Subscriber<? super T> sub = subscriberRef.get();
+            if (sub != null && boundRef.get() != null) {
+                sub.onNext(t);
+            } else {
+                handleLostBinding(sub, "onNext");
+            }
+        }
+
+        private void handleLostBinding(Subscriber<? super T> sub, String context) {
+            if (sub == null) {
+                Log.d(LOG_TAG, "subscriber gone; skipping " + context);
+            } else {
+                Log.d(LOG_TAG, "bound component gone; skipping " + context);
+            }
+            unsubscribe();
+        }
+
+    }
+}