Proposal: Lift Operator (was 'bind')

[benjchristensen]

Over the past couple months several sources of inspiration have trigged the idea of adding a bind operator to Observable. I have played with an implementation and propose we adopt it for RxJava but want to get feedback on signatures and find out if there are any use cases I'm completely missing that would negate the idea.

tldr; Prototype code => https://gist.github.com/benjchristensen/8367765

Jan 29th: Updated to lift instead of bind based on discussions in #775

Benefits

1) Synchronous Unsubscribe

The primary benefit of this proposal is that synchronous Observables can be unsubscribed. This means an infinite sequence of integers without a separate thread could now be used. A large Iterable followed by take(20) would actually only emit the first 20.

This is achieved by injecting Subscription into what we today call Observable.OnSubscribeFunc instead of it being returned by the function.

An Observable of this nature would check the Subscription inside its loop such as this:

        Obsurvable<Integer> OBSERVABLE_OF_INFINITE_INTEGERS = Obsurvable.create(new Action2<Observer<Integer>, OperatorSubscription>() {

            @Override
            public void call(Observer<Integer> o, OperatorSubscription s) {
                int i = 1;
                while (!s.isUnsubscribed()) {
                    o.onNext(i++);
                }
                o.onCompleted();
            }
        });

2) Custom Operator Chaining

Because Java doesn't support extension methods, the only approach to applying custom operators without getting them added to rx.Observable is using static methods. This has meant code like this:

MyCustomerOperators.operate(observable.map(...).filter(...).take(5)).map(...).subscribe()

In reality we want:

observable.map(...).filter(...).take(5).myCustomOperator().map(...).subscribe()

Using bind we can get quite close to this:

observable.map(...).filter(...).take(5).bind(MyCustomOperator.operate()).map(...).subscribe()

Here is how the proposed bind method looks if all operators were applied with it:

Obsurvable<String> os = OBSERVABLE_OF_INFINITE_INTEGERS.bind(TAKE_5).bind(MAP_INTEGER_TO_STRING);

3) Simpler Operator Implementations

The bind operator injects the necessary Observer and Subscription instances and eliminates (for most use cases) the need for manual subscription management. Because the Subscription is available in-scope there are no awkward coding patterns needed for creating a Subscription, closing over it and returning and taking into account synchronous vs asynchronous.

For example, fromIterable is simply:

    public static <T> Obsurvable<T> from(final Iterable<T> is) {
        return Obsurvable.create(new Action2<Observer<T>, OperatorSubscription>() {

            @Override
            public void call(Observer<T> o, OperatorSubscription s) {
                for (T i : is) {
                    if (s.isUnsubscribed()) {
                        break;
                    }
                    o.onNext(i);
                }
                o.onCompleted();
                s.unsubscribe();
            }

        });
    }

The take operator is:

    public static class TakeOperator<T> implements Func2<Observer<T>, OperatorSubscription, Observer<T>> {

        final int limit;

        TakeOperator(int limit) {
            this.limit = limit;
        }

        @Override
        public Observer<T> call(final Observer<T> o, final OperatorSubscription s) {
            if (limit == 0) {
                o.onCompleted();
                s.unsubscribe();
            }
            return new Observer<T>() {

                int count = 0;

                @Override
                public void onCompleted() {
                    o.onCompleted();
                }

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

                @Override
                public void onNext(T i) {
                    if (!s.isUnsubscribed()) {
                        o.onNext(i);
                        if (++count >= limit) {
                            o.onCompleted();
                            s.unsubscribe();
                        }
                    }
                }

            };
        }

    }

4) Eliminate Need for CurrentThreadScheduler

A side-effect of injecting the Subscription and handling synchronous execution is we can now use loops instead of trampolining for operators like repeat. This means we should not need the CurrentThreadScheduler (as far as I can tell) and the ugliness it takes to make synchronous unsubscribes work.

Here is the repeat operator without need for scheduling:

    public static class RepeatOperator<T> implements Func2<Observer<T>, OperatorSubscription, Observer<Obsurvable<T>>> {

        RepeatOperator() {
        }

        @Override
        public Observer<Obsurvable<T>> call(final Observer<T> o, final OperatorSubscription s) {
            return new Observer<Obsurvable<T>>() {

                @Override
                public void onCompleted() {
                    o.onCompleted();
                }

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

                @Override
                public void onNext(Obsurvable<T> ot) {
                    while (!s.isUnsubscribed()) {
                        ot.f.call(o, s);
                    }
                }

            };
        }

    }

5) Recursion/Loop Performance

The fromIterable use case is 20x faster when implemented as a loop instead of recursive scheduler (see a18b8c1).

Many places recursive scheduling were used to avoid stack overflow can be done as a loop instead.

Drawbacks

1) Observable.create Signature Change

Currently the Observable.create method is defined as:

 public static <T> Observable<T> create(OnSubscribeFunc<T> func)

This is effectively the same as:

 public static <T> Observable<T> create(Func1<Observer<? super T>, Subscription> func)

To inject the Subscription we need it to instead be:

public static <T> Obsurvable<T> create(Action2<Observer<? super T>, Subscription> f)

Note that there is not return type any longer, the Subscription is injected instead of returned.

Then to support operators conditionally attaching themselves to the injected Subscription the signature is actually:

public static <T> Obsurvable<T> create(final Action2<Observer<? super T>, OperatorSubscription> f)

The OperatorSubscription simply adds void add(Subscription s) so each operator can register with it as needed.

Prototype Code

The prototype code can be found at https://gist.github.com/benjchristensen/8367765

The name of the class is Obsurvable and this code can be dropped into the rx package as Obsurvable.java alongside Observable.java for testing.

Design Decisions

If we agree that this proposal make sense, there are some design decisions to make:

1) Action2/Func2 or New Type?

Right now the prototype code has this:

public static <T> Obsurvable<T> create(final Action2<Observer<? super T>, OperatorSubscription> f)
public <R> Obsurvable<R> bind(final Func2<Observer<R>, OperatorSubscription, Observer<T>> bind) 

Should we create types that represent those? I think the create one probably should, for the same reason as before, to simplify the pain of dealing with generics. Thus it would be OnSubscribeFunc, OnSubscribe or something similar.

For the bind method I'm less opinionated, though it may be cleaner to provide a type that represents the Func2.

I did play with a variant of this that combined Observer and OperatorSubscription into a single type Operator to make the function simpler but it conflated things and was quite troublesome with composition as the Observer and Subscription had different lifecycles and intents.

2) Private Subscribe for Nested Observables

If you look at the MergeOperator code you'll see inside the onNext(Obsurvable<T> innerObservable) that it has this code:

innerObsurvable.f.call(observer, s)

This is reaching inside the Obsurvable to the private field f and invoking it.

This is equivalent to calling Obsurvable.subscribe except that it inject the Subscription rather than returning it.

I still want the normal Subscription Observable subscribe(Observer o) signature for public consumption, but am considering whether we should have:

a) an alternate public signature for subscription that looks like void Observable otherSubscribe(Observer o, Subscription s)

b) a private mechnism that operator implementations can somehow subscribe (not sure yet how without these only being implemented inside Observable itself)

I tend to prefer (b) so we don't leak implementation details, but (a) would allow anyone to create even the more complicated nested operators such as merge outside of Observable.java.

3) OperatorSubscription Name

Is this name okay? Where should it live? Is its signature correct?

4) Observable.create

Should we deprecate or keep the current signature?

If it exists alongside the new one and both have the create name accepting a single function it will break Groovy and Clojure (and probably JRuby) as they won't be able to distinguish between overloads. Thus it likely is best to just do the breaking change cleanly and have a single create method with the new signature.

Is there any reason to have the old signature that returns Subscription instead of accepting it as an argument?

Operator Refactor

If we adopt this pattern it will take some time to retrofit all operators to use bind.

In Observable.java they would be coded something like this using bind:

    public <R> Obsurvable<R> flatMap(Func1<T, Obsurvable<R>> f) {
        return bind(new MapOperator<T, Obsurvable<R>>(f)).bind(new MergeOperator<R>());
    }

    public Obsurvable<T> take(int num) {
        return bind(new TakeOperator<T>(num));
    }

Each operator would need to be changed to this new model.

I suggest we leverage this to do other cleanup such as:

• eliminate unnecessary synchronization (use of volatile, Atomic_, Concurrent_, locking, etc)
• eliminate unnecessary object allocation (such as Notification or use of materialize)
• add a new /src/perf folder and add a performance test for each operator implementation similar to https://github.com/Netflix/RxJava/blob/master/rxjava-core/src/test/java/rx/schedulers/SchedulerPerformanceTests.java and https://github.com/Netflix/RxJava/blob/master/rxjava-core/src/test/java/rx/subjects/SubjectPerformanceTests.java
• rename each operator from Operation* to Operator* as a marker of which have been migrated (and to read better and match the package name)
• come up with standard approaches for concurrency when it is needed, such as state machines recently applied to Schedulers and Subscriptions.

I'd appreciate feedback and improvement. Thanks.

[benjchristensen]

For feedback /cc @headinthebox @abersnaze @samuelgruetter @jmhofer @mairbek @zsxwing @akarnokd @michaeldejong @johnhmarks @duarten @rickbw

[akarnokd]

After the first look, it definitely solves the take(n) problem as the downstream take can now access the cancellation token for the upstream (btw, loops should exit with return instead of break in case of isUnsubscribed).

However, external users are still unable to cancel the long running operation as subscribe won't return the subscription until it has finished. One would need to have an overload where the cancellation token is passed in from the outside as well (or a binding which leaks the OperatorSubscription):

OperatorSubscription os = new OperatorSubscription();
Schedulers.computation().schedule(() -> os.unsubscribe(), 1, TimeUnit.SECONDS);
Obsurvable.from(1).repeat().subscribe(os, new Observer<T>() { ... }); // void

But it doesn't work at the moment as from(1) unsubscribes the os and repeat can't really repeat it (it prints 1 and two onCompleted). If I remove the unsubscribe and change repeat to not emit onCompleted, it works as expected and honors the RxJava contract better (see 8415495 and 8415517).

After the modification it appears to me that we replace one problem with another: now one would need to carefully isUnsubscribed almost everywhere and think twice when to call unsubscribe.

[duarten]

I like this approach. It keeps the user facing behaviour the same while allowing for a simpler implementation.

After a first look, it seems that upstream operators shouldn't be able to unsubscribe downstream operators. So, bind would change to:

public <R> Obsurvable<R> bind(final Func2<Observer<R>, OperatorSubscription, Observer<T>> bind) {
    return new Obsurvable<R>(new Action2<Observer<R>, OperatorSubscription>() {

        @Override
        public void call(Observer<R> o, OperatorSubscription s) {
            OperatorSubscription s2 = new OperatorSubscription();
            s.add(s2)
            f.call(bind.call(o, s), s2);
        }
    });
}

Take can unsubscribe from all the upstream operators, but these can't unsubscribe Take (which would be notified through onComplete).

[duarten]

We can write merge like so:

@Override
public void onNext(Obsurvable<T> innerObsurvable) {
    Subscription innerSubscription = innerObsurvable.subscribe(new Observer<T>() {
      // ...
    });

    outerSubscription.add(innerSubscription);
}

We already have to protect against onNext calls arriving after a call to onCompleted and unsubscribe, so first subscribing to the inner observable and then adding the inner subscription to the composite won't introduce any new interleaving.

This seems to allow for option 2) b) while also allowing people to create arbitrary nested observables, or am I missing something?

[benjchristensen]

We can write merge like so

Each of the inner Observable instances can also be synchronous which would make the call to subscribe block and thus we would never receive the Subscription back.

I think we may just want to provide a subscribe overload or new name such as:

public final void subscribe(Observer<? super T> observer, OperatorSubscription s)
// or
public final void observe(Observer<? super T> observer, OperatorSubscription s)

I think I'd prefer the different name observe. I'm not super concerned with discoverability as it's incredibly rare that someone would need to be building an operator of this type and in that case they're reading the docs and complying with the create/bind/observe semantics.

UPDATE

These signatures only work for "one shot" unless they create a new OperatorSubscription each time. Which is why it's beneficial for it to be hidden inside the normal subscribe. If someone is going to use this though they generally should have a reason (creating a nested operator) so that may be okay, or we could change the signature to protect against that by doing it as:

public final void observe(Observer<? super T> observer, Func0<OperatorSubscription> subscriptionFactory)

[benjchristensen]

external users are still unable to cancel the long running operation as subscribe won't return the subscription until it has finished

I have yet to see actual use code where they use the Subscription received from subscribe. It's more-or-less a sign they are doing it wrong if they need to as they should instead be using things like take and takeUntil (particularly takeUntil for conditional cases) for ending Observables.

The only time I'm aware of a Subscription from subscribe being used for valid reasons is when building operators to combine Observables together, and in that case we are saying to use bind instead. See the previous comment for thoughts on exposing the "private subscribe" publicly for the rare case of building a new operator beyond merge/concat/zip/switch/etc for nested behavior.

[benjchristensen]

After the modification it appears to me that we replace one problem with another: now one would need to carefully isUnsubscribed almost everywhere and think twice when to call unsubscribe.

How is this any different than now? We currently have to very carefully consider where to place a Subscription, what to put within it and what manual cancellation tokens (usually a boolean) we have to check while in loops that will be set by that Subscription we return. I see the consideration exactly the same.

I also don't see it as any more sensitive about when to call unsubscribe. It has the same impact as our current approach does. As soon as you call it everything is supposed to shut down. So how do we have to think twice in one approach but not in the other?

On top of that, with the current implementation we have to deal with the fact that the approach doesn't work for synchronous use cases and we have to keep answering questions from people as to why their code doesn't behave as they expect and we keep telling them "use Schedulers" which in effect means Rx is only useful if all data sources is async (or small enough to not be a problem ... which kind of defeats the purpose).

However, this is actually only half the answer, as in Rx.Net those synchronous cases work by putting everything on a recursive CurrentThreadScheduler and interweaving through the code base "hacks" to modify tokens similar or equal to ThreadLocal state that then stop the recursion. See my benefits (4) and (5) for why this new approach to create/bind seems far better than that. Thus, if we don't use this bind approach we have to accept slower behavior for all recursion and figure out how to make CurrentThreadScheduler work. I spent time implementing it and it was basically "whack-a-mole" with hacks in multiple places in the codebase to capture an unsubscribe and set it on a ThreadLocal so CurrentThreadScheduler could see that it had been unsubscribed and stop recursion. This is fairly straight-forward when on a single thread. It all becomes far more difficult when crossing over thread boundaries or when nesting Observables and different Schedulers in a single Observable sequence.

[benjchristensen]

After a first look, it seems that upstream operators shouldn't be able to unsubscribe downstream operators. So, bind would change to:

I need to play with this a bit to better understand and will then respond...

[duarten]

Each of the inner Observable instances can also be synchronous which would make the call to subscribe block and thus we would never receive the Subscription back.

Right! I forgot about that case. The overload taking a subscription factory seems like the better approach then.

[benjchristensen]

The MergeOperator would end up like this:

                public void onNext(Obsurvable<T> innerObsurvable) {
                    innerObsurvable.subscribe(new Observer<T>() {

                        @Override
                        public void onCompleted() {
                            synchronized (o) {
                                o.onCompleted();
                            }
                        }

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

                        @Override
                        public void onNext(T a) {
                            synchronized (o) {
                                o.onNext(a);
                            }
                        }

                    }, new Func0<OperatorSubscription>() {

                        @Override
                        public OperatorSubscription call() {
                            OperatorSubscription innerSubscription = new OperatorSubscription();
                            outerSubscription.add(innerSubscription);
                            return innerSubscription;
                        }

                    });

                }

Using a subscribe method like this:

    public void subscribe(Observer<T> o, Func0<OperatorSubscription> sf) {
        f.call(o, sf.call());
    }

It's a little unnecessary for a private operator "doing the right thing" and involves closing over the outerSubscription but it does provide a safe public signature.

[benjchristensen]

Updated prototype code with observe method: https://gist.github.com/benjchristensen/8367765

This would make the create/bind/observe methods siblings for creating operators.

Thus, an Observable is observed by an Observer and the OperatorSubscription is always injected.

The "public" signature of Observable.subscribe is unchanged.

[akarnokd]

I've implemented zip with this new paradigm and it felt a bit odd:

public <U, R> Obsurvable<R> zip(Obsurvable<U> other, Func2<T, U, R> resultSelector) {
        Obsurvable<T> t = this;
        return Obsurvable.create((o, s) -> {
            Object guard = new Object();
            Queue<T> leftQueue = new LinkedList<>();
            Queue<U> rightQueue = new LinkedList<>();
            AtomicBoolean leftDone = new AtomicBoolean();
            AtomicBoolean rightDone = new AtomicBoolean();
            OperatorSubscription tsub = new OperatorSubscription();
            OperatorSubscription usub = new OperatorSubscription();
            s.add(tsub);
            s.add(usub);
            Action0 stride = () -> {
                boolean done = false;
                synchronized (guard) {
                    while (!leftQueue.isEmpty() 
                            && !rightQueue.isEmpty() 
                            && !s.isUnsubscribed()) {
                        try {
                            o.onNext(resultSelector.call(leftQueue.poll(), rightQueue.poll()));
                        } catch (Throwable e) {
                            o.onError(e);
                            s.unsubscribe();
                            return;
                        }
                    }
                    if (!s.isUnsubscribed()) {
                        if ((leftQueue.isEmpty() && leftDone.get())
                                || (rightQueue.isEmpty() && rightDone.get())) {
                            done = true;
                            o.onCompleted();
                        }
                    }
                }
                if (done) {
                    s.unsubscribe();
                }
            };
            t.observe(new Observer<T>() {
                @Override
                public void onNext(T args) {
                    synchronized (guard) {
                        leftQueue.offer(args);
                    }
                    stride.call();
                }
                @Override
                public void onError(Throwable e) {
                    synchronized (guard) {
                        o.onError(e);
                    }
                    s.unsubscribe();
                }
                @Override
                public void onCompleted() {
                    synchronized (guard) {
                        leftDone.set(true);
                    }
                    stride.call();
                }
            }, () -> tsub);

            other.observe(new Observer<U>() {

                @Override
                public void onNext(U args) {
                    synchronized (guard) {
                        rightQueue.offer(args);
                    }
                    stride.call();
                }

                @Override
                public void onError(Throwable e) {
                    synchronized (guard) {
                        o.onError(e);
                    }
                }

                @Override
                public void onCompleted() {
                    synchronized (guard) {
                        rightDone.set(true);
                    }
                    stride.call();
                }
            }, () -> usub);
        });
    }

And a test method:

Obsurvable.from(1).repeat().take(10000).zip(
Obsurvable.from(2).repeat().take(5), (a, b) -> a + b).take(2)
.subscribe(System.out::println);

Which again doesn't work with the reference repeat() as it should not call onCompleted once its upstream calls it because it will continue with subscribing to it again. Once fixed the above example works.

The oddity comes from where and what to call unsubscribe on; is this okay?

OperatorSubscription tsub = new OperatorSubscription();
OperatorSubscription usub = new OperatorSubscription();
s.add(tsub);
s.add(usub);

And should I call tsub.unsubscribe() in the first observer.onCompleted() or s.unsubscribe()?

[abersnaze]

It's probably unrealistic to assume that all of the operations can be rewritten in one pull request. This method will allow existing operators that use OnSubscribeFunc to still work.

    public static <X> Obsurvable<X> create(final OnSubscribeFunc<X> onSub) {
        return create(new Action2<Observer<X>, OperatorSubscription>() {
            @Override
            public void call(Observer<X> in, OperatorSubscription s) {
                Subscription sub = onSub.onSubscribe(in);
                s.add(sub);
            }
        });
    }

[benjchristensen]

@akarnokd Here is another implementation of zip as comparison: https://gist.github.com/benjchristensen/8367765#file-obsurvable-java-L340

The only place an unsubscribe is invoked is here: https://gist.github.com/benjchristensen/8367765#file-obsurvable-java-L427

There are no places where it must be checked.

The most interesting unit test for this is testZipInfiniteAndFinite() as it must unsubscribe the children: https://gist.github.com/benjchristensen/8367765#file-obsurvable-java-L549

The thing that makes this work is that whenever a subscription occurs via observe the Subscription is injected into it and applies to the entire lifecycle of that subscription. This can be seen here:

os[i].observe((InnerObserver) observers[i], new Func0<OperatorSubscription>() {

    @Override
    public OperatorSubscription call() {
        return childSubscription;
    }
});

This could be simplified to the following if we want to make a user have to "do the right thing" when using it:

os[i].observe((InnerObserver) observers[i], childSubscription);

The same childSubscription is injected into all Obsurvables being zipped together so a single unsubscribe will cause them all to be unsubscribed as each of them will be looking at the same Subscription wherever applicable (such as a ObsurvableFromIterable loop).

[abersnaze]

@akarnokd I think the root of your zip problems comes from the use of Observable.create(...) and not this.bind(...).

[abersnaze]

For last couple days I've been working on building on this to build a visual debugger of sorts for Rx. Because of the OperatorSubscription.add() method it allows the association between inner and outer observers to be discovered (aka merge). Here is a sample the raw data that I was able to get with four hooks (on create, on bind, on add and on observe) for the code

from(1,3,5).flatMap({i -> from(i, i+1)}).take(3).subscribe({ print it })

[rickbw]

In the current API, the fundamental thing you do with an observable object is subscribe to it by passing an observer and getting a subscription back. You can invoke that behavior on Observable itself with subscribe(Observer), or you can inject that behavior into Observable with Observable.create and OnSubscribeFunc.onSubscribe(Observer). Either way, the signature and the contract are exactly the same.

In the proposed new API, the fundamental thing you do is observe by passing in both an observer and a subscription. You can invoke that behavior with observe(Observer<T>, Func0<OperatorSubscription>) or inject it with Action2<Observer<T>, OperatorSubscription>.call. Except that you've lost the symmetry of the current API! I would either go with Action2<Observer<T>, Func0<OperatorSubscription>> and observe(Observer<T>, Func0<OperatorSubscription>) or with Action2<Observer<T>, OperatorSubscription> and observe(Observer<T>, OperatorSubscription). But don't mismatch the signatures. Otherwise, we'll find ourselves constantly writing closures to convert OperatorSubscription into Func0<OperatorSubscription>.

[abersnaze]

I think you might be right about the Func0 being unnecessary. My current workspace is too far removed from the base line its hard to say for sure but I was able to remove the Func0 implementations and it still work for my use case.

[benjchristensen]

While implementing various operators I ran into groupBy which required a signature change on bind to compose the Subscriptions. It's working, though I'm not convinced the signature is as elegant as it can be.

The prototype code is still here: https://gist.github.com/benjchristensen/8367765 It has grown to include bind, map, flatMap, merge, take, zip, groupBy, and repeat. There are unit tests for these asserting behavior including handling infinite synchronous streams, composed subscriptions and general functionality.

I have also split out a simpler example (ObsurvableBind.java) with just the bind operator and a small number of unit tests for necessary functionality to simplify discussion of and iterating on the bind signature: https://gist.github.com/benjchristensen/8486461

The essential parts of the code are currently:

public static <T> ObsurvableBind<T> create(final Action2<Observer<T>, OperatorSubscription> f) {
    return new ObsurvableBind<T>(f);
}

public void observe(Observer<T> o, OperatorSubscription sf) {
    f.call(o, sf);
}

public Subscription subscribe(Observer<T> o) {
    final OperatorSubscription os = new OperatorSubscription();
    observe(o, os);
    return os;
}

public static interface Operator<T> extends Observer<T> {

    public OperatorSubscription getSubscription();
}

public <R> ObsurvableBind<R> bind(final Func2<Observer<R>, OperatorSubscription, Operator<T>> bind) {
    return new ObsurvableBind<R>(new Action2<Observer<R>, OperatorSubscription>() {

        @Override
        public void call(Observer<R> o, final OperatorSubscription s) {
            Operator<T> ot = bind.call(o, s);
            observe(ot, ot.getSubscription());
        }
    });
}

This achieves the generally desired traits but I'd like to see if we can come up with a better way of representing the Func definition of bind.

The public void testBindUnsubscribeNested() unit test represents the groupBy type functionality that requires nested composition of Subscriptions.

[benjchristensen]

Here is another possible variant of the signature: https://gist.github.com/benjchristensen/8497234

Instead of Func2<Observer<R>, OperatorSubscription, Operator<T>> it uses Func1<Operator<R>, Operator<T>>:

public static <T> ObsurvableBind2<T> create(final Action1<Operator<T>> f) {
    return new ObsurvableBind2<T>(f);
}

public <R> ObsurvableBind2<R> bind(final Func1<Operator<R>, Operator<T>> bind) {
    return new ObsurvableBind2<R>(new Action1<Operator<R>>() {

        @Override
        public void call(Operator<R> o) {
            observe(bind.call(o));
        }
    });
}

public void observe(Operator<T> o) {
    f.call(o);
}

public Subscription subscribe(final Observer<T> o) {
    final OperatorSubscription os = new OperatorSubscription();
    observe(createOperator(o, os));
    return os;
}

public static interface Operator<T> extends Observer<T> {

    /**
     * Get the Subscription intended to pass up to the source being bound to.
     * <p>
     * In other words, the subscription from Operator -> Source
     */
    public OperatorSubscription getSubscription();
}

[benjchristensen]

Here is another variant: https://gist.github.com/benjchristensen/8506432

This eliminates the OperatorSubscription and the Operator becomes an abstract class instead of interface:

public static abstract class Operator<T> implements Observer<T>, Subscription {

        private final CompositeSubscription cs;

        Operator(CompositeSubscription cs) {
            this.cs = cs;
        }

        Operator() {
            this.cs = new CompositeSubscription();
        }

        /**
         * Used to register an unsubscribe callback.
         */
        public final void add(Subscription s) {
            cs.add(s);
        }

        @Override
        public final void unsubscribe() {
            cs.unsubscribe();
        }

        public final boolean isUnsubscribed() {
            return cs.isUnsubscribed();
        }
    }

The create/bind/observe signatures stay the same:

    public static <T> ObsurvableBind3<T> create(final Action1<Operator<T>> f) {
        return new ObsurvableBind3<T>(f);
    }

    public <R> ObsurvableBind3<R> bind(final Func1<Operator<R>, Operator<T>> bind) {
        return new ObsurvableBind3<R>(new Action1<Operator<R>>() {

            @Override
            public void call(Operator<R> o) {
                observe(bind.call(o));
            }
        });
    }

    public void observe(Operator<T> o) {
        f.call(o);
    }

@rickbw Are these different variants solving your concerns? What improvements would you make to their signatures and which do you prefer?

@duarten I believe these new signatures address the problem you brought up regarding the passing of Subscription. The unit tests combining groupBy and take assert the particular use case.

[headinthebox]

public static abstract class Operator implements Observer, Subscription {

}

This I like.

On Sun, Jan 19, 2014 at 4:44 PM, Ben Christensen
[email protected]:

Here is another variant: https://gist.github.com/benjchristensen/8506432

This eliminates the OperatorSubscription and the Operator becomes an
abstract class instead of interface:

public static abstract class Operator implements Observer, Subscription {

    private final CompositeSubscription cs;

    Operator(CompositeSubscription cs) {
        this.cs = cs;
    }

    Operator() {
        this.cs = new CompositeSubscription();
    }

    /**         * Used to register an unsubscribe callback.         */
    public final void add(Subscription s) {
        cs.add(s);
    }

    @Override
    public final void unsubscribe() {
        cs.unsubscribe();
    }

    public final boolean isUnsubscribed() {
        return cs.isUnsubscribed();
    }
}

The create/bind/observe signatures stay the same:

public static <T> ObsurvableBind3<T> create(final Action1<Operator<T>> f) {
    return new ObsurvableBind3<T>(f);
}

public <R> ObsurvableBind3<R> bind(final Func1<Operator<R>, Operator<T>> bind) {
    return new ObsurvableBind3<R>(new Action1<Operator<R>>() {

        @Override
        public void call(Operator<R> o) {
            observe(bind.call(o));
        }
    });
}

public void observe(Operator<T> o) {
    f.call(o);
}

@rickbw https://github.com/rickbw Are these different variants solving
your concerns? What improvements would you make to their signatures and
which do you prefer?

@duarten https://github.com/duarten I believe these new signatures
address the problem you brought up regarding the passing of Subscription.
The unit tests combining groupBy and take assert the particular use case.

—
Reply to this email directly or view it on GitHubhttps://github.com//issues/746#issuecomment-32711145
.

[samuelgruetter]

This thread blessed me with the following insight:

If we look at the pull-oriented Iterator

interface Iterator<T> {
    boolean hasNext();
    T next();
}

and try to find its push-oriented dual, we get something like

interface PushIterator<T> {
    boolean wantsNext();
    void onNext(T);
    void onError(Throwable);
    void onComplete();
}

which is basically the same as @benjchristensen's

public static abstract class Operator<T> implements Observer<T>, Subscription

where wantsNext() = !isUnsubscribed().

That said, I like the Operator idea, but I think the naming can be improved: Operator is not general enough, I can also imagine use cases where I'd use an Operator instead of an Observer, without my Operator really being an operator in the sense that it transforms an Observable. Moreover, the name Subscription makes sense if it's returned by a call to subscribe, but if I create a Subscription myself and pass it to the subscribe method, this name is not very intuitive any more. I've not yet any good naming suggestion (I do not think PushIterator is a good name), but I'm thinking...

[headinthebox]

Samuel, the subscription comes from the IDispoable, which is not in the Java Iterable interface.

[duarten]

Yes, it solves that problem. I like it!

[benjchristensen]

Anyone interested in this please take a look at the pull request I just submitted.

@headinthebox and @samuelgruetter I could use your help fixing the Scala module.

[benjchristensen]

Thanks everyone for your discussion on this. I have merged the implementation.

Next steps are:

1. Nail down naming and signatures: Naming of Create/Bind/Operator Functions and Types #775
2. Document (a README or Wiki page) operator implementation standards, guidelines and best practices
3. Update operator implementations over time.

[mttkay]

Sorry for chiming in late. I'm a bit all over the place these days and some of the latest developments in the project slipped past me. I'll comment here w.r.t. #770 as well.

I read the proposal and I like the suggested changes. One concern I have is that observers now have to extend an abstract class. In Java, it's quite intrusive if a library or a framework forces super classes on you, since it lacks multiple inheritance. We had a few components in our app that would behave like observers, but inherited from Android framework classes. This was quite nice, since they would provide the "glue" between the Android framework and our Rx based client code. We now have to convert these to observers first by delegating to an inner class, which makes their public API a bit awkward.

More problematic though is test friendliness. Maybe I'm missing something, but a very common recipe that we adopted in dozens of unit tests is the following:

Observer mockObserver = mock(Observer.class);
obj.getObservable().subscribe(mockObserver);
ArgumentCaptor<Observer> onNextArgs = ArgumentCaptor.of(...)
verify(mockObserver).onNext(onNextArgs.capture())
// do expecations against onNextArgs

This is not possible anymore, since SafeObserver into which all external observers get wrapped requires a valid subscription in a private final field, something that a mock object cannot provide.

FWIW, Mockito @SPY's work, but it might not be wise to rely on spies. Maybe this could at least be mitigated by providing a non-final getter for the subscription which SafeObserver queries? At least then we'd be able to return a subscription for the mock that SafeObserver can accept.

Ben also pointed me to TestObserver, which can be used as a wrapper to query arguments. Neither solution strikes me as very elegant though.

Sorry I'll have to work more with the current code base to get a deeper impression, just wanted to throw this in as a first impression of sorts.

[benjchristensen]

Thanks @mttkay for getting involved and providing feedback.

I too have similar concerns about Observer become abstract and I'm still considering the pros and cons of the change but thus far have come to see abstract Observer implements Subscription as being the best of the proposed designs.

For unit tests your code would need to change like this:

Observer mockObserver = mock(Observer.class);
obj.getObservable().subscribe(new TestObserver(mockObserver));
ArgumentCaptor<Observer> onNextArgs = ArgumentCaptor.of(...)
verify(mockObserver).onNext(onNextArgs.capture())
// do expecations against onNextArgs

Going back to the discussion on signatures in #775 the choice is basically to either:

a) change Observer to an abstract class and have a single type
b) create a new type that combines Observer and Subscription and is used in the create(OnSubscribe) and lift signatures but not subscribe.

At this point the strongest argument has been to keep the public API clean and have a single Observer type that works in all cases. The drawback of this approach though is the impact on multiple-inheritance and unit testing.

The current signatures are:

// Observable.create
public final static <T> Observable<T> create(OnSubscribe<T> f)

// Observable.OnSubscribe typed function interface
public static interface OnSubscribe<T> extends Action1<Observer<? super T>>

// lift function
public <R> Observable<R> lift(final Func1<Observer<? super R>, Observer<? super T>> bind)

// Observer
public abstract class Observer<T> implements Subscription {
     public abstract void onNext(T t);
     public abstract void onError(Throwable e);
     public abstract void onCompleted();
     public final void add(Subscription s)
     public final void unsubscribe()
     public final boolean isUnsubscribed()
}

// Subject
public abstract class Subject<T, R> extends Observer<T> {
    public abstract Observable<R> toObservable();
}

Anyone have a better design than this?

[benjchristensen]

This code demonstrates what the new signatures allow:

        Observable.create(new OnSubscribe<Integer>() {

            @Override
            public void call(Observer<? super Integer> ob) {
                for (int i = 1; i < 100000; i++) {
                    /*
                     * The Observer communicates whether it is unsubscribed
                     * so loops and seqential processing on the same thread
                     * can now unsubscribe.
                     */
                    if (ob.isUnsubscribed()) {
                        System.out.println("--- Unsubscribed at: " + i);
                        return;
                    }
                    ob.onNext(i);
                }
                ob.onCompleted();
            }

        }).subscribe(new Observer<Integer>() {

            @Override
            public void onCompleted() {
                System.out.println("Completed");
            }

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

            @Override
            public void onNext(Integer i) {
                System.out.println("Received: " + i);
                if (i == 10) {
                    // an Observer can now unsubscribe
                    unsubscribe();
                }
            }

        });

    }

This outputs:

Received: 1
Received: 2
Received: 3
Received: 4
Received: 5
Received: 6
Received: 7
Received: 8
Received: 9
Received: 10
--- Unsubscribed at: 11

Note how the source Observable can check ob.isUnsubscribed() and the Observer can now unsubscribe() all within a single-threaded sequential call.

This is equally beneficial if the Observable is made async by running on a separate thread, but still a sequential for-loop instead of trampolining (which is far slower than a loop and both less obvious and more complicated to implement):

        Observable.create(new OnSubscribe<Integer>() {

            @Override
            public void call(Observer<? super Integer> ob) {
                for (int i = 1; i < 100000; i++) {
                    /*
                     * The Observer communicates whether it is unsubscribed
                     * so loops and seqential processing on the same thread
                     * can now unsubscribe.
                     */
                    if (ob.isUnsubscribed()) {
                        System.out.println("--- Unsubscribed at: " + i);
                        return;
                    }
                    ob.onNext(i);
                }
                ob.onCompleted();
            }
                  // use subscribeOn so it is now async
        }).subscribeOn(Schedulers.newThread()).subscribe(new Observer<Integer>() {

            @Override
            public void onCompleted() {
                System.out.println("Completed");
            }

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

            @Override
            public void onNext(Integer i) {
                System.out.println("Received: " + i);
                if (i == 10) {
                    // an Observer can now unsubscribe
                    unsubscribe();
                }
            }

        });

This means we do want the Observer to be more intelligent and encapsulate the Subscription so we achieve these goals.

Unit testing is doable by using TestObserver and I'm okay with that being slightly less elegant.

The remaining drawback is the inability to implement Observer since it is now abstract.

A relevant question is whether classes need to implement Observer so as to "be" an Observer. We do not support this on Observable; classes are not "an" Observable but generally have getters that return an Observable or a class has a toObservable() method on it.

Similarly classes do not need to be an Observer but can encapsulate that logic. That said, there may still be value in having an interface that represents something that can be turned into an Observer.

Right now Observable has that with the Observable.OnSubscribe interface. Should we have something similar for Observer? If so, what should it be called? Observer.OnObserve?

We either need a new name for Observer+Subscription (the currently modified Observer) and return Observer to just the 3 on* methods and a mechanism for converting from it to the Observer+Subscription. Or we leave Observer as Observer+Subscription and come up with a different name to represent the interface of the 3 on* methods.

[mttkay]

Thanks for the write up Ben! I do agree this is very desirable.

So an Observer's role is an active one now (it can unsubscribe) rather than
just a passive one (it receives notifications). That's fine, and Observer
is still a good name (I'm less happy with the fact that Observer "is a"
Subscription now, just in terms of naming, but anyway, I see how this makes
sense implementation wise.)

I usually name interfaces after the behavior they enable. If we reintroduce
an interface that resembles what Observer used to be, I'd vote for
something along the lines of Notifiable or NotificationReceiver to
reflect that the interface captures the behavior of being able to receive
Rx notifications. OnObserve sounds like a single callback to an observe
function so that might be misleading.

Just my thoughts.

[headinthebox]

Can't we provide a default implementation of Observer like this?

public abstract class Observer<T> implements Subscription {
     public abstract void onNext(T t);
     public abstract void onError(Throwable e);
     public abstract void onCompleted();

     protected Subscription subscription = .... default implementation for Subscription ...
     public final void add(Subscription s){ subscription.add(s); }
     public final void unsubscribe(){ subscription.unsubscribe(); }
     public final boolean isUnsubscribed(){ return subscription.isUnsubscribed(); }
}

[benjchristensen]

Can't we provide a default implementation of Observer like this?

That is how it is implemented: https://github.com/Netflix/RxJava/blob/master/rxjava-core/src/main/java/rx/Observer.java

[benjchristensen]

If we go this path, an interface Notifiable could work, since we already have the Notification type.

public abstract class Observer<T> implements Subscription, Notifiable

Or we could leave Observer as the passive object that receives notifications and use Subscriber to represent the thing that calls subscribe and it would implement Observer and Subscription.

public abstract class Subscriber<T> implements Observer<T>, Subscription

[benjchristensen]

Note that if we went to having Subscriber and Observer we'd then have the question of whether we allow Observable.subscribe(Observer o) because if we did, those implementations could not unsubscribe.

That is probably fine though, as most Observer's don't need to unsubscribe.

Thus we could have methods like this in Observable:

void subscribe(Subscriber s)
Subscription subscribe(Observer o)

[benjchristensen]

I did an implementation using Subscriber to see if it makes sense. Please provide feedback on issue #792.