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