OperatorObserveFromAndroidComponent to support Fragment.isVisible()

[tehmou]

I have a use case on Android in which I would want a fragment to stop updating its data while it's hidden. It would seem natural to add a support for this in OperatorObserveFromAndroidComponent. Maybe there could be a switch for whether the observable should be inactive only when the fragment is not added or also when it's hidden?

[zsxwing]

So basically, a callback to indicate if the fragment is active can improve the API. What's your opinion, @mttkay

public static <T> Observable<T> from(Func0<Boolean> isActive, Observable<T> sourceObservable) 

[zsxwing]

Here is my thought that will solve this issue and another issue I mentioned in #880 (comment)

Essentially, we need some api to make an Observable become inactive and keep silent (we can not use Subscription returned by subscribe)

At first, we need the following OperatorAttach to attach an Observable with a Func1.

public final class OperatorAttach<T> implements Observable.OnSubscribe<T> {

    private final Observable<T> source;
    private final Func1<Notification<T>, Boolean> isAttached;

    public OperatorAttach(Observable<T> source, final Subscription detach) {
        this(source, new Func1<Notification<T>, Boolean>() {
            @Override
            public Boolean call(Notification<T> t1) {
                return !detach.isUnsubscribed();
            }
        });
    }

    /**
     * Generate a new Observable that mirrors the source but will swallow messages once isAttached return false.
     * 
     * @param source
     * @param isAttached
     */
    public OperatorAttach(Observable<T> source, Func1<Notification<T>, Boolean> isAttached) {
        this.source = source;
        this.isAttached = isAttached;
    }

    @Override
    public void call(final Subscriber<? super T> subscriber) {
        source.materialize().takeWhile(isAttached).subscribe(new Subscriber<Notification<T>>(subscriber) {
            @Override
            public void onCompleted() {
                // ignore
            }

            @Override
            public void onError(Throwable e) {
                // ignore
            }

            @Override
            public void onNext(Notification<T> notification) {
                if (notification.isOnNext()) {
                    subscriber.onNext(notification.getValue());
                } else if (notification.isOnError()) {
                    subscriber.onError(notification.getThrowable());
                } else {
                    subscriber.onCompleted();
                }
            }
        });
    }

}

Then, observeFromAndroidComponent is like this.

    public static <T> Observable<T> observeFromAndroidComponent(Observable<T> source,
            android.app.Fragment fragment, final Func1<android.app.Fragment, Boolean> isAttached) {
        final WeakReference<android.app.Fragment> fragmentRef;
        Object memoryBarrier = new Object();
        synchronized (memoryBarrier) { // force a memory barrier
            fragmentRef = new WeakReference<android.app.Fragment>(fragment);
        }
        return Observable.create(new OperatorAttach<T>(source.observeOn(AndroidSchedulers.mainThread()),
                new Func1<Notification<T>, Boolean>() {
                    @Override
                    public Boolean call(Notification<T> t1) {
                        android.app.Fragment fragment = fragmentRef.get();
                        return fragment != null && isAttached.call(fragment);
                    }
                }));
    }

    public static <T> Observable<T> observeFromAndroidComponent(Observable<T> source, Subscription detach) {
        return Observable
                .create(new OperatorAttach<T>(source.observeOn(AndroidSchedulers.mainThread()), detach));
    }

Use case:

• Activity

public class TestActivity extends Activity {

    private volatile Subscription detach;

    private Observable<Integer> o;

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        detach = new BooleanSubscription();
        o = OperatorObserveFromAndroidComponent.observeFromAndroidComponent(
                Observable.from(Arrays.asList(1, 2, 3)), detach);
    }

    @Override
    protected void onDestroy() {
        detach.unsubscribe();
        super.onDestroy();
    }

}

• Fragment

        android.app.Fragment fragment = ...;
        OperatorObserveFromAndroidComponent.observeFromAndroidComponent(
                Observable.from(Arrays.asList(1, 2, 3)), fragment, new Func1<android.app.Fragment, Boolean>() {

                    @Override
                    public Boolean call(Fragment t1) {
                        return t1.isAdded(); // or t1.isVisible();
                    }

                });

@mttkay, any idea?

[nsk-mironov]

@zsxwing, my guess is that OperatorObserveFromAndroidComponent assumes that Fragment can become active and inactive more than once, so takeWhile isn't appropriate here... I like your implementation more, though.

[zsxwing]

@mironov-nsk , I think creating two operator (one use takeWhile, one use filter) is better. That would meet your requirements.

[zsxwing]

@benjchristensen is it worth to put such OperatorAttach in the rxjava-core? Does anyone need this operator out of rxjava-android?

[tehmou]

@zsxwing I came to a similar conclusion with my initial forked implementation - I actually ended up needing a slightly custom isAttached function. My nested fragment checks if its parent is visible in addition to itself because FragmentTransaction.hide does not cascade the hidden property.

OperatorAttach does indeed seem a lot more generic than just this rxjava-android use case, I would like seeing it in the core myself.

[zsxwing]

I just realized I did the same thing as dematerialize. The following codes are simpler.

public static Observable<T> attach(Observable<T> source, Func1<Notification<T>, Boolean> isAttached) {
    return source.materialize().takeWhile(isAttached).dematerialize();
}

public static Observable<T> filter(Observable<T> source, Func1<Notification<T>, Boolean> isAttached) {
    return source.materialize().filter(isAttached).dematerialize();
}

[benjchristensen]

I don't fully understand the use case as I don't work with UIs or Android so can't help very much, but are you trying to unsubscribe from the event source when it is hidden, or just ignore them until it becomes visible?

In other words, must you keep listening to the events so as to know when it becomes visible again?

[benjchristensen]

If I understand correctly, if the visibility is not taken into account we receive events when hidden?

e: touch event [visible]
e: touch event [visible]
e: hidden [hidden]
e: touch event [hidden]
e: touch event [hidden]
e: touch event [hidden]
e: visible [visible]
e: hidden [hidden]
e: touch event [hidden]
e: visible [visible]
e: touch event [visible]

I imagine you want this?

e: touch event [visible]
e: touch event [visible]
e: touch event [visible]
e: touch event [visible]
e: hidden [hidden]
e: visible [visible]
e: touch event [visible]
e: hidden [hidden]
e: visible [visible]
e: touch event [visible]
e: touch event [visible]
e: touch event [visible]

    public static Observable<FragmentEvent> listenToFragment(final Fragment fragment) {
        // simulate registering with the listener for a fragment
        return Observable.create(new OnSubscribe<FragmentEvent>() {

            @Override
            public void call(final Subscriber<? super FragmentEvent> s) {
                s.add(Schedulers.newThread().scheduleRecursive(new Action1<Recurse>() {

                    boolean hidden = false;

                    @Override
                    public void call(Recurse r) {
                        if (Math.random() < 0.2) {
                            hidden = !hidden;
                            if (hidden) {
                                s.onNext(new FragmentEvent(fragment, "hidden"));
                            } else {
                                s.onNext(new FragmentEvent(fragment, "visible"));
                            }
                        } else {
                            s.onNext(new FragmentEvent(fragment, "touch event"));
                        }
                        r.schedule(500, TimeUnit.MILLISECONDS);

                    }

                }));
            }
        });
    }

    public static void main(String[] args) {
        listenToFragment(new Fragment("f1")).toBlockingObservable().forEach(new Action1<FragmentEvent>() {

            boolean visible = true;

            @Override
            public void call(FragmentEvent e) {
                if (e.event.equals("hidden")) {
                    visible = false;
                } else if (e.event.equals("visible")) {
                    visible = true;
                }
                System.out.println("e: " + e.event + " [" + (visible ? "visible" : "hidden") + "]");
            }

        });
    }

[tehmou]

I want to ignore the events of the observable while the fragment is hidden. However, as I mentioned, actually whether or not it is hidden is not a property but a custom function. Therefore, as @zsxwing suggested, there would need to be a boolean function isAttached to determine the state.

In the OperatorObserveFromAndroidComponent there is already such functionality for the fragment state isAdded. Explicitly subscribing/unsubscribing according to a temporary state of the fragment is inconvenient because it requires subclassing and overriding methods.

The usage of the code in question would end up something like this:

    Observable<String> src = ...;
    android.app.Fragment fragment = ...;

    Observable<String> filteredSource =
        OperatorObserveFromAndroidComponent.observeFromAndroidComponent(
             src, fragment, new Func1<android.app.Fragment, Boolean>() {
                 @Override
                 public Boolean call(Fragment t1) {
                     android.app.Fragment parent = t1.getParentFragment();
                     return t1.isVisible() && parent != null && parent.isVisible();
                 }
             });

    filteredSource.subscribe(new Action1<String>() {
        @Override
        public void call(String str) {
            // Update the UI with the new data
        });

[benjchristensen]

Should it actually subscribe/unsubscribe, or just ignore events?

If it keeps the subscription open here are two approaches, one with filter and one with switchOnNext:

    public static Observable<FragmentEvent> fragmentWithVisibilityCheckUsingSwitch(final Fragment fragment) {

        return Observable.switchOnNext(listenToFragment(fragment).map(new Func1<FragmentEvent, Observable<FragmentEvent>>() {

            boolean visible = true;

            @Override
            public Observable<FragmentEvent> call(FragmentEvent e) {
                if (e.event.equals("hidden")) {
                    visible = false;
                    return Observable.from(e); // could be Observable.never() to ignore
                } else if (e.event.equals("visible")) {
                    visible = true;
                    return Observable.from(e); // could be Observable.never() to ignore
                } else {
                    if (visible) {
                        return Observable.from(e);
                    } else {
                        return Observable.never();
                    }
                }

            }

        }));
    }

    public static Observable<FragmentEvent> fragmentWithVisibilityCheckUsingFilter(final Fragment fragment) {

        return listenToFragment(fragment).filter(new Func1<FragmentEvent, Boolean>() {

            boolean visible = true;

            @Override
            public Boolean call(FragmentEvent e) {
                if (e.event.equals("hidden")) {
                    visible = false;
                    return true; // could be false to ignore
                } else if (e.event.equals("visible")) {
                    visible = true;
                    return true; // could be false to ignore
                } else {
                    if (visible) {
                        return true;
                    } else {
                        return false;
                    }
                }

            }

        });
    }

Of course the function for determining whether it is visible can be injected as you state.

is it worth to put such OperatorAttach in the rxjava-core?

What is different from the attach operator and using filter?

[mttkay]

Maybe I'm missing something, but doesn't the operator already do that? It will skip all notifications unless the fragment isAdded.

[zsxwing]

What is different from the attach operator and using filter?

Now I think we don't need to add this operator.

Here I think there are two use cases:

• disable an Observable forever once the isAttached return false. For example, once an Activity is destroyed, we should disable the Observable forever.
• disable an Observable once the isAttached return false, and enable an Observable once the isAttached return true. For example, according the visible property of the fragment to enable or disable the Observable .

[zsxwing]

Maybe I'm missing something, but doesn't the operator already do that? It will skip all notifications unless the fragment isAdded.

I'm trying to support that the Observable can always be subscribed by many Observers if the Activity is alive.

[mttkay]

OK. From the the initial discussion it sounded as if this was merely about not emitting notifications if a fragment is in the background.

As for the use cases you mentioned:

disable an Observable forever once the isAttached return false. For example, once an Activity is destroyed, we should disable the Observable forever.

We can't do that; going through a configuration change means the fragment will get re-attached, and the observable must continue emitting. We use that pattern frequently with cache and replay

disable an Observable once the isAttached return false, and enable an Observable once the isAttached return true. For example, according the visible property of the fragment to enable or disable the Observable .

What do you mean by disable? Could this also be solved by using cache and replay? You just keep emitting items into the cached observable, and once your fragment/activity is back to life, you re-subscribe

[benjchristensen]

I'm trying to support that the Observable can always be subscribed by many Observers if the Activity is alive.

That suggests use of a Subject.

cache and replay

This sounds correct, but what are the lifecycle implications as far as memory is concerned? The cache() operator is replay() with no limits so if an app is alive for hours or days it will keep retaining memory. Perhaps it is just the last x events or minutes of events?

Curious, why would there be events when an activity is not alive? or a fragment is not visible? And if there are events while hidden or not alive, should they be retained or just ignored completely?

[zsxwing]

Curious, why would there be events when an activity is not alive? or a fragment is not visible? And if there are events while hidden or not alive, should they be retained or just ignored completely?

Here the events are not the UI events. It's some events emitted from a background thread. For example, a thread may read some information from a web server in a background thread, and send it to the UI thread. In the UI thread, the UI elements will be updated according to the information.

In Android, AndroidSchedulers.mainThread() is used to dispatch such events to the UI thread. There is an event queue for the UI thread. Sometimes, when the UI thread fetches an event (sent from a background thread) from the queue, maybe now the Activity is gone, or a fragment is not visible. So such events need to be dropped.

I think it's a common pattern that creating an Observable that has the same life-cycle as the Activity.

In addition, I don't want to store a strong reference to an Activity or a Fragment in the Observable, or it will not be cleaned by GC. So I use a WeakReference.

[mttkay]

The operator already takes care of this? If you unsubscribe in onDestroy,
it will drop all outstanding messages.

Android guarantees that no messages will be delivered between onDestroy and
the next call to onCreate, so there's only two cases:

• the Activity doesn't come back to life, because the user backed out for
  instance
• the Activity goes through onCreate again because it was a configuration
  change like change in rotation. In that case, use cache or replay and keep
  the Observable alive, for instance using a fragment that retains instance
  state

That said, I'm still not sure what the problem is we're trying to solve or
how it's not accounted for already?

[zsxwing]

That said, I'm still not sure what the problem is we're trying to solve or
how it's not accounted for already?

• In current rxjava-android, I think the following behavior is wrong. This is the first problem I want to fix.

public class TestActivity extends Activity {

    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        Observable<Integer> o = OperatorObserveFromAndroidComponent.observeFromAndroidComponent(
                Observable.from(1), this);
        // the following codes output "1"
        o.subscribe(new Observer<Integer>() {

            @Override
            public void onCompleted() {
            }

            @Override
            public void onError(Throwable e) {
                e.printStackTrace();
            }

            @Override
            public void onNext(Integer t) {
                System.out.println(t);
            }
        });
        // the following codes output nothing
        // this is a wrong behavior, should output "1"
        o.subscribe(new Observer<Integer>() {

            @Override
            public void onCompleted() {
            }

            @Override
            public void onError(Throwable e) {
                e.printStackTrace();
            }

            @Override
            public void onNext(Integer t) {
                System.out.println(t);
            }
        });
    }
}

• The second one is adding an API to let users can control when the Observable is active. For example, a Button, or a TextView can also be removed, like Fragment. So some Observable may be bind to a Button, or other UI component. I want to provide an API to keep the Observable silence once the UI component is removed. Since the use case may be very complex, I want to add one function like isAttached to let users determine when to keep the Observable silence.

[zsxwing]

The operator already takes care of this? If you unsubscribe in onDestroy,
it will drop all outstanding messages.

Here I'm not sure which one is better, using takeWhile or filter? Maybe filter is better, since Activity or Fragment can come back to live.

[mttkay]

Just to summarize, what we want to fix is this yes:

• make observables wrapped by fromFragment/fromActivity re-subscribable
• pull out decision logic for when to silence the observable into a function that we pass in, rather than having it internal subclasses

[zsxwing]

Yes, thanks for your summary.

[tehmou]

Thanks @zsxwing for elaborating the use case. @mttkay I agree with the summary, it supports my need of having a more complex function to determine the state.

@benjchristensen, as suggested, the source observable I have is sending events whenever it receives them from a push network API. Subscribing/Unsubscribing is done in fragment onCreate/onDestroy, but in between there are moments when the UI is hidden or otherwise inactive and the events can be discarded. The story of keeping the displayed data consistent over the application is of course a lot longer and we have an elaborate system for it, but in this case each notification was only shown in the UI for a short moment.

[mttkay]

Not sure anymore where this should be best discussed, but will leave my comment here:

While working on the samples project for rxjava-android, I actually discovered a few other issues that I was unaware of previously. Currently, fromFragment has to be used with much care and in a very specific, often non-obvious way to actually prevent memory leaks from happening. This is due to a combination of things:

• resubscribing doesn't work, as already pointed out, so one cannot unsubscribe from a cached observable in onDestroyView before a config change, and resubscribe in onViewCreated. This means the fragment will leak. The only way to fix this right now is to always recreate the outer observable by calling fromFragment again immediately before subscribing to the cached inner sequence.
• once the reference is cleared out, the operator doesn't unsubscribe its inner observer, so it will continue receiving messages even though they will all get discarded (this is rather easy to fix but I'm not sure if this has other repercussions, so I'm testing this right now)
• because of the assertUiThread issue, the sequence can fail silently if no onError handler is specified (easy to fix too)

As said, I have fixes for the latter two issues, although they require more testing. However, I agree this operator demands rethinking on a bigger scale I guess. Especially the eager (and final) binding to the observed UI component in the constructor is a culprit here that needs to be addressed.

[mttkay]

So, I gave this some deeper thought over the weekend and came to the conclusion that I would prefer to deprecate and completely rewrite the current operator. Here is is what I came up with:

https://gist.github.com/mttkay/0590979394aec6144a2e

TL;DR:
OperatorWeakBinding takes an arbitrary object reference, binds it weakly, binds any subscriber weakly, and only forwards notifications if both are alive. If not, it unsubscribes itself.

Benefits:

• does not leak context references, even when they're implicit through an inner class subscriber
• no distinction anymore between fragments and activities; everything is considered a weak binding to some UI component (should therefore work with views, too)
• it's not necessary to unsubscribe anymore, due to the use of weak references
• it's not necessary anymore to test for fragment isAdded; since we re-subscribe in onViewCreated, we discard any old observer references after going through a config change
• auto-unsubscribes when subscriber or bound reference are gone
• can be used with lift since it's an Operator now

Drawbacks:

• expects to (and requires) the caller to manually observeOn(mainThread()).cache, since otherwise re-subscribing would restart the sequence (perhaps there are even use cases for this though?)

I only tested this guy quickly in the samples project in both a fragment that retains instance state and an activity which retains the sequence by piping it through onRetainNonConfigurationInstance and I saw the activities being properly garbage collected.

What's missing:
What @tehmou suggested, i.e. making bindings more controllable by passing in a function. I was wondering though whether this should be its own operator, or implemented on top of this using existing operators? When I wrote the initial implementation of OperatorObserveFromAndroidComponent my goals were to make it simple to observe sequences on the main UI thread, without having to manage context references. That backfired a bit I guess, so I see this new implementation to be a superior implementation for this particular use case, not so much something that's more feature rich.

Could you guys test this, review this and give me your thoughts? Thanks!

[mttkay]

I realized this can even be further simplified by providing new helpers through AndroidObservable similar to what fromFragment used to do:

    public static <T> Observable<T> bindActivity(final Activity activity, Observable<T> cachedSequence) {
        return cachedSequence.observeOn(AndroidSchedulers.mainThread()).lift(new OperatorWeakBinding<T, Activity>(activity));
    }

    public static <T> Observable<T> bindFragment(final Fragment fragment, Observable<T> cachedSequence) {
        return cachedSequence.observeOn(AndroidSchedulers.mainThread()).lift(new OperatorWeakBinding<T, Fragment>(fragment));
    }

In the Activity, you can then simplify to:

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_reactive);

        o = (Observable<String>) getLastNonConfigurationInstance();
        if (o == null) {
            o = SampleObservables.numberStrings(1, 100, 200).cache();
        }

        s = bindActivity(this, o).subscribe(new Observer());
    }

and the Fragment even to:

    @Override
    public void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);

        strings = bindFragment(this, SampleObservables.numberStrings(1, 100, 200).cache());
    }

So the only assumption that remains is that the source sequence must always be cached; otherwise going through an unsubscribe/subscribe cycle during a rotation change would restart the sequence instead of continuing it. Any ideas how we could solve this? I guess the problem is that the underlying sequence must continue to receive notifications during the time the fragment/activity is detached.

[mttkay]

Just tested this in one of our apps' core screens, works nicely so far.

I've prepared a branch against netflix/master that deprecates ObserveFromAndroidComponent and introduces this one. I'll push it and send a PR, so that you guys can test it out more easily.

[mttkay]

Maybe it's just getting late, but while working on the RxJava Android samples project, I made an observation: it seems the very problem all the Android operator implementations so far try to address, i.e. binding the sequence to the life time of a fragment or activity without leaking it, seems only to actually be an issue when using cache or replay (which, unfortunately, is a very common use case for us.)

When using e.g. just publish to connect a fragment's inner observer (which in turn holds a reference to the fragment), then as long as I unsubscribe in onDestroyView, no context will leak, even for retained fragments. (Using publish is unfortunately not useful on Android, since the UI component might miss out on an emitted item while being detached.)

I've opened an issue report here since I don't want to drive this issue further off topic, but would be glad if someone could either verify or falsify it:
#939

I'm just wondering whether the whole issue we're trying to fix might simply be down to a bug in ReplaySubject, in which case the whole discussion around OperatorWeakBinding would be null and void...

[mttkay]

So, I invested more time into this today, and the result is in the PR mentioned before:
#938

I have added support to bind a sequence via a predicate function that can be controlled from the outside. bindActivity now by default adds a predicate that tests for isFinishing, and bindFragment tests for isAdded, but this behaviour can overridden now by lifting the operator manually using a custom predicate:

source.lift(new OperatorWeakBinding(this, new Func1<Fragment, T> { ... })).subscribe(observer)

In summary:

bindActivity(this, sourceSequence).subscribe(s)

==> keeps a weak reference to both this and s, schedules notifications on the main UI thread, and drops notifications and unsubscribes whenever:

• this (the activity) is finishing
• this is gone
• s is gone

bindFragment(this, sourceSequence).subscribe(s)

==> same, just that the predicate test is for isAdded

@zsxwing @tehmou let me know whether this addresses all the problems we discussed?

[mttkay]

/cc @samueltardieu

[samueltardieu]

This looks nice indeed!

[mttkay]

I suggest to close this and in case of follow up problems / discussion to open a new issue.