OperatorWeakBinding to not use WeakReferences anymore

rxjava-contrib/rxjava-android-samples/samples/src/main/AndroidManifest.xml

@@ -29,6 +29,15 @@
         <activity
             android:name=".ListFragmentActivity">
 
+            <intent-filter>
+                <category android:name="android.intent.category.LAUNCHER"/>
+                <category android:name="android.intent.category.DEFAULT"/>
+                <action android:name="android.intent.action.MAIN"/>
+            </intent-filter>
+        </activity>
+        <activity
+            android:name=".ListenInOutActivity">
+
             <intent-filter>
                 <category android:name="android.intent.category.LAUNCHER"/>
                 <category android:name="android.intent.category.DEFAULT"/>

rxjava-contrib/rxjava-android-samples/samples/src/main/java/com/netflix/rxjava/android/samples/ListenInOutActivity.java

@@ -0,0 +1,87 @@
+package com.netflix.rxjava.android.samples;
+
+import android.app.Activity;
+import android.os.Bundle;
+import android.view.View;
+import android.widget.TextView;
+import android.widget.Toast;
+import android.widget.ToggleButton;
+
+import rx.Observable;
+import rx.Observer;
+import rx.Subscription;
+import rx.observables.ConnectableObservable;
+
+import static rx.android.observables.AndroidObservable.bindActivity;
+
+/**
+ * Activity that binds to a counting sequence and is able to listen in and out to that
+ * sequence by pressing a toggle button. The button disables itself once the sequence
+ * finishes.
+ */
+public class ListenInOutActivity extends Activity implements Observer<String> {
+
+    private Observable<String> source;
+    private Subscription subscription;
+    private TextView textView;
+
+    @Override
+    protected void onCreate(Bundle savedInstanceState) {
+        super.onCreate(savedInstanceState);
+
+        setContentView(R.layout.listen_in_out_activity);
+
+        textView = (TextView) findViewById(android.R.id.text1);
+
+        // in a production app, you would use dependency injection, fragments, or other
+        // means to preserve the observable, but this will suffice here
+        source = (Observable<String>) getLastNonConfigurationInstance();
+        if (source == null) {
+            source = SampleObservables.numberStrings(1, 100, 200).publish();
+            ((ConnectableObservable) source).connect();
+        }
+
+        subscribeToSequence();
+    }
+
+    private void subscribeToSequence() {
+        subscription = bindActivity(this, source).subscribe(this);
+    }
+
+    @Override
+    public Object onRetainNonConfigurationInstance() {
+        return source;
+    }
+
+    @Override
+    protected void onDestroy() {
+        subscription.unsubscribe();
+        super.onDestroy();
+    }
+
+    @Override
+    public void onCompleted() {
+        TextView button = (TextView) findViewById(R.id.toggle_button);
+        button.setText("Completed");
+        button.setEnabled(false);
+    }
+
+    @Override
+    public void onError(Throwable e) {
+        e.printStackTrace();
+        Toast.makeText(this, "Error: " + e, Toast.LENGTH_SHORT).show();
+    }
+
+    @Override
+    public void onNext(String s) {
+        textView.setText(s);
+    }
+
+    public void onSequenceToggleClicked(View view) {
+        if (((ToggleButton) view).isChecked()) {
+            subscription.unsubscribe();
+        } else {
+            subscribeToSequence();
+        }
+    }
+}

rxjava-contrib/rxjava-android-samples/samples/src/main/java/com/netflix/rxjava/android/samples/ListeningFragmentActivity.java

@@ -14,7 +14,7 @@
 import rx.observables.ConnectableObservable;
 import rx.subscriptions.Subscriptions;
 
-import static rx.android.schedulers.AndroidSchedulers.mainThread;
+import static rx.android.observables.AndroidObservable.bindFragment;
 
 /**
  * Problem:
@@ -52,7 +52,7 @@ public ListeningFragment() {
         public void onCreate(Bundle savedInstanceState) {
             super.onCreate(savedInstanceState);
 
-            strings = SampleObservables.numberStrings(1, 50, 250).observeOn(mainThread()).publish();
+            strings = SampleObservables.numberStrings(1, 50, 250).publish();
             strings.connect(); // trigger the sequence
         }
 
@@ -74,7 +74,7 @@ public void onViewCreated(final View view, Bundle savedInstanceState) {
             final TextView textView = (TextView) view.findViewById(android.R.id.text1);
 
             // re-connect to sequence
-            subscription = strings.subscribe(new Subscriber<String>() {
+            subscription = bindFragment(this, strings).subscribe(new Subscriber<String>() {
 
                 @Override
                 public void onCompleted() {

rxjava-contrib/rxjava-android-samples/samples/src/main/java/com/netflix/rxjava/android/samples/RetainedFragmentActivity.java

@@ -14,11 +14,12 @@
 
 import rx.Observable;
 import rx.Subscription;
-import rx.android.schedulers.AndroidSchedulers;
 import rx.functions.Action1;
 import rx.functions.Func1;
 import rx.subscriptions.Subscriptions;
 
+import static rx.android.observables.AndroidObservable.bindFragment;
+
 /**
  * Problem:
  * You have a data source (where that data is potentially expensive to obtain), and you want to
@@ -68,9 +69,10 @@ public void onCreate(Bundle savedInstanceState) {
             super.onCreate(savedInstanceState);
 
             // simulate fetching a JSON document with a latency of 2 seconds
-            strings = SampleObservables.fakeApiCall(2000).map(PARSE_JSON)
-                    .observeOn(AndroidSchedulers.mainThread())
-                    .cache();
+            // in retained fragments, it's sufficient to bind the fragment in onCreate, since
+            // Android takes care of detaching the Activity for us, and holding a reference for
+            // the duration of the observable does not harm.
+            strings = bindFragment(this, SampleObservables.fakeApiCall(2000).map(PARSE_JSON).cache());
         }
 
         @Override

rxjava-contrib/rxjava-android-samples/samples/src/main/res/layout/listen_in_out_activity.xml

@@ -0,0 +1,28 @@
+<RelativeLayout xmlns:android="http://schemas.android.com/apk/res/android"
+    xmlns:tools="http://schemas.android.com/tools"
+    android:layout_width="match_parent"
+    android:layout_height="match_parent"
+    android:paddingLeft="@dimen/activity_horizontal_margin"
+    android:paddingRight="@dimen/activity_horizontal_margin"
+    android:paddingTop="@dimen/activity_vertical_margin"
+    android:paddingBottom="@dimen/activity_vertical_margin"
+    tools:context="com.netflix.rxjava.android.samples.ListenInOutActivity">
+
+    <TextView
+        android:id="@android:id/text1"
+        android:layout_centerInParent="true"
+        android:layout_width="wrap_content"
+        android:layout_height="wrap_content" />
+
+    <ToggleButton
+        android:id="@+id/toggle_button"
+        android:layout_width="wrap_content"
+        android:layout_height="wrap_content"
+        android:layout_below="@android:id/text1"
+        android:layout_centerHorizontal="true"
+        android:textOff="Pause"
+        android:textOn="Resume"
+        android:onClick="onSequenceToggleClicked" />
+
+
+</RelativeLayout>

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

@@ -20,7 +20,7 @@
 import rx.Observable;
 import rx.functions.Func1;
 import rx.operators.OperatorObserveFromAndroidComponent;
-import rx.operators.OperatorWeakBinding;
+import rx.operators.OperatorConditionalBinding;
 
 import android.app.Activity;
 import android.app.Fragment;
@@ -51,15 +51,15 @@ public Boolean call(Activity activity) {
     private static final Func1<Fragment, Boolean> FRAGMENT_VALIDATOR = new Func1<Fragment, Boolean>() {
         @Override
         public Boolean call(Fragment fragment) {
-            return fragment.isAdded();
+            return fragment.isAdded() && !fragment.getActivity().isFinishing();
         }
     };
 
     private static final Func1<android.support.v4.app.Fragment, Boolean> FRAGMENTV4_VALIDATOR =
             new Func1<android.support.v4.app.Fragment, Boolean>() {
                 @Override
                 public Boolean call(android.support.v4.app.Fragment fragment) {
-                    return fragment.isAdded();
+                    return fragment.isAdded() && !fragment.getActivity().isFinishing();
                 }
             };
 
@@ -131,39 +131,47 @@ public static <T> Observable<T> fromFragment(Object fragment, Observable<T> sour
     }
 
     /**
-     * Binds the given source sequence to the life-cycle of an activity.
+     * Binds the given source sequence to an activity.
      * <p/>
      * This helper will schedule the given sequence to be observed on the main UI thread and ensure
-     * that no notifications will be forwarded to the activity in case it gets destroyed by the Android runtime
-     * or garbage collected by the VM.
+     * that no notifications will be forwarded to the activity in case it is scheduled to finish.
+     * <p/>
+     * You should unsubscribe from the returned Observable in onDestroy at the latest, in order to not
+     * leak the activity or an inner subscriber. Conversely, when the source sequence can outlive the activity,
+     * make sure to bind to new instances of the activity again, e.g. after going through configuration changes.
+     * Refer to the samples project for actual examples.
      *
      * @param activity the activity to bind the source sequence to
      * @param source   the source sequence
      */
     public static <T> Observable<T> bindActivity(Activity activity, Observable<T> source) {
         Assertions.assertUiThread();
-        return source.observeOn(mainThread()).lift(new OperatorWeakBinding<T, Activity>(activity, ACTIVITY_VALIDATOR));
+        return source.observeOn(mainThread()).lift(new OperatorConditionalBinding<T, Activity>(activity, ACTIVITY_VALIDATOR));
     }
 
     /**
-     * Binds the given source sequence to the life-cycle of a fragment (native or support-v4).
+     * Binds the given source sequence to a fragment (native or support-v4).
      * <p/>
      * This helper will schedule the given sequence to be observed on the main UI thread and ensure
      * that no notifications will be forwarded to the fragment in case it gets detached from its
-     * activity or garbage collected by the VM.
+     * activity or the activity is scheduled to finish.
+     * <p/>
+     * You should unsubscribe from the returned Observable in onDestroy for normal fragments, or in onDestroyView
+     * for retained fragments, in order to not leak any references to the host activity or the fragment.
+     * Refer to the samples project for actual examples.
      *
      * @param fragment the fragment to bind the source sequence to
      * @param source   the source sequence
      */
-    public static <T> Observable<T> bindFragment(Object fragment, Observable<T> cachedSequence) {
+    public static <T> Observable<T> bindFragment(Object fragment, Observable<T> source) {
         Assertions.assertUiThread();
-        final Observable<T> source = cachedSequence.observeOn(mainThread());
+        final Observable<T> o = source.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, FRAGMENTV4_VALIDATOR));
+            return o.lift(new OperatorConditionalBinding<T, android.support.v4.app.Fragment>(f, FRAGMENTV4_VALIDATOR));
         } 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, FRAGMENT_VALIDATOR));
+            return o.lift(new OperatorConditionalBinding<T, Fragment>(f, FRAGMENT_VALIDATOR));
         } else {
             throw new IllegalArgumentException("Target fragment is neither a native nor support library Fragment");
         }

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

@@ -0,0 +1,91 @@
+package rx.operators;
+
+import rx.Observable;
+import rx.Subscriber;
+import rx.functions.Func1;
+import rx.functions.Functions;
+
+import android.util.Log;
+
+/**
+ * Ties a source sequence to the given target object using a predicate. If the predicate fails
+ * to validate, the sequence unsubscribes itself and releases the bound reference.
+ * <p/>
+ * You can also pass in an optional predicate function, which whenever it evaluates to false
+ * on the target object, will also result in the operator unsubscribing from the sequence.
+ *
+ * @param <T> the type of the objects emitted to a subscriber
+ * @param <R> the type of the target object to bind to
+ */
+public final class OperatorConditionalBinding<T, R> implements Observable.Operator<T, T> {
+
+    private static final String LOG_TAG = "ConditionalBinding";
+
+    private R boundRef;
+    private final Func1<? super R, Boolean> predicate;
+
+    public OperatorConditionalBinding(R bound, Func1<? super R, Boolean> predicate) {
+        boundRef = bound;
+        this.predicate = predicate;
+    }
+
+    public OperatorConditionalBinding(R bound) {
+        boundRef = bound;
+        this.predicate = Functions.alwaysTrue();
+    }
+
+    @Override
+    public Subscriber<? super T> call(final Subscriber<? super T> child) {
+        return new  Subscriber<T>(child) {
+
+            @Override
+            public void onCompleted() {
+                if (shouldForwardNotification()) {
+                    child.onCompleted();
+                } else {
+                    handleLostBinding("onCompleted");
+                }
+            }
+
+            @Override
+            public void onError(Throwable e) {
+                if (shouldForwardNotification()) {
+                    child.onError(e);
+                } else {
+                    handleLostBinding("onError");
+                }
+            }
+
+            @Override
+            public void onNext(T t) {
+                if (shouldForwardNotification()) {
+                    child.onNext(t);
+                } else {
+                    handleLostBinding("onNext");
+                }
+            }
+
+            private boolean shouldForwardNotification() {
+                return boundRef != null && predicate.call(boundRef);
+            }
+
+            private void handleLostBinding(String context) {
+                log("bound object has become invalid; skipping " + context);
+                log("unsubscribing...");
+                boundRef = null;
+                unsubscribe();
+            }
+
+            private void log(String message) {
+                if (Log.isLoggable(LOG_TAG, Log.DEBUG)) {
+                    Log.d(LOG_TAG, message);
+                }
+            }
+        };
+    }
+
+    /* Visible for testing */
+    R getBoundRef() {
+        return boundRef;
+    }
+}