Version 0.17.0 Release Notes [Preview]

[benjchristensen]

#0.17.0 Release Notes

Version 0.17.0 contains some significant signature changes that allow us to significantly improve handling of synchronous Observables and simplify Schedulers. Many of the changes have backwards compatible deprecated methods to ease the migration while some are breaking.

The new signatures related to Observable in this release are:

// A new create method takes `OnSubscribe` instead of `OnSubscribeFunc`
public final static <T> Observable<T> create(OnSubscribe<T> f)

// The new OnSubscribe type accepts a Subscriber instead of Observer and does not return a Subscription
public static interface OnSubscribe<T> extends Action1<Subscriber<? super T>>

// Subscriber is an Observer + Subscription
public abstract class Subscriber<T> implements Observer<T>, Subscription

// The main `subscribe` behavior receives a Subscriber instead of Observer
public final Subscription subscribe(Subscriber<? super T> observer)

// Subscribing with an Observer however is still appropriate
// and the Observer is automatically converted into a Subscriber
public final Subscription subscribe(Observer<? super T> observer)

// A new 'lift' function allows composing Operator implementations together
public <R> Observable<R> lift(Func1<Subscriber<? super R>, Subscriber<? super T>> lift)

Also changed is the Scheduler interface which is much simpler:

public abstract class Scheduler {
    public Subscription schedule(Action1<Scheduler.Inner> action);
    public Subscription schedule(Action1<Scheduler.Inner> action, long delayTime, TimeUnit unit);
    public Subscription schedulePeriodically(Action1<Scheduler.Inner> action, long initialDelay, long period, TimeUnit unit);
    public long now();
    public int degreeOfParallelism();

    public static class Inner implements Subscription {
        public abstract void schedule(Action1<Scheduler.Inner> action, long delayTime, TimeUnit unit);
        public abstract void schedule(Action1<Scheduler.Inner> action);
        public long now();
    }
}

This release applies many lessons learned over the past year and seeks to streamline the API before we hit 1.0.

As shown in the code above the changes fall into 2 major sections:

1) Lift/OnSubscribe/Subscriber

Changes that allow unsubscribing from synchronous Observables without needing to add concurrency.

2) Schedulers

Simplification of the Scheduler interface and make clearer the concept of "outer" and "inner" Schedulers for recursion.

Lift/OnSubscribe/Subscriber

New types Subscriber and OnSubscribe along with the new lift operator have been added. The reasons and benefits are as follows:

1) Synchronous Unsubscribe

RxJava versions up until 0.16.x are unable to unsubscribe from a synchronous Observable such as this:

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

    @Override
    public void call(Observer<? super Integer> Observer) {
        for (int i = 1; i < 1000000; i++) {
            subscriber.onNext(i);
        }
        subscriber.onCompleted();
    }
});

Subscribing to this Observable will always emit all 1,000,000 values even if unsubscribed such as via oi.take(10).

Version 0.17.0 fixes this issue by injecting the Subscription into the OnSubscribe function to allow code like this:

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

    @Override
    public void call(Subscriber<? super Integer> subscriber) {
        // we now receive a Subscriber instead of Observer
        for (int i = 1; i < 1000000; i++) {
            // the OnSubscribe can now check for isUnsubscribed
            if (subscriber.isUnsubscribed()) {
                return;
            }
            subscriber.onNext(i);
        }
        subscriber.onCompleted();
    }

});

Subscribing to this will now correctly only emit 10 onNext and unsubscribe:

// subscribe with an Observer
oi.take(10).subscribe(new Observer<Integer>() {

    @Override
    public void onCompleted() {

    }

    @Override
    public void onError(Throwable e) {

    }

    @Override
    public void onNext(Integer t) {
        println("Received: " + t);
    }

})

Or the new Subscriber type can be used and the Subscriber itself can unsubscribe:

// or subscribe with a Subscriber which supports unsubscribe
oi.subscribe(new Subscriber<Integer>() {

    @Override
    public void onCompleted() {

    }

    @Override
    public void onError(Throwable e) {

    }

    @Override
    public void onNext(Integer t) {
        println("Received: " + t);
        if(t >= 10) {
            // a Subscriber can unsubscribe
            this.unsubscribe();
        }
    }

})

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 the newly added lift we can get quite close to this:

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

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

Observable<String> os = OBSERVABLE_OF_INTEGERS.lift(TAKE_5).lift(MAP_INTEGER_TO_STRING);

Along with the lift function comes a new Operator signature:

public interface Operator<R, T> extends Func1<Subscriber<? super R>, Subscriber<? super T>>

All operator implementations in the rx.operators package will over time be migrated to this new signature.

3) Simpler Operator Implementations

The lift operator injects the necessary Observer and Subscription instances (via the new Subscriber type) 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, the body of fromIterable is simply:

public void call(Subscriber<? super T> o) {
    for (T i : is) {
        if (o.isUnsubscribed()) {
            return;
        }
        o.onNext(i);
    }
    o.onCompleted();
}

The take operator is:

public final class OperatorTake<T> implements Operator<T, T> {

    final int limit;

    public OperatorTake(int limit) {
        this.limit = limit;
    }

    @Override
    public Subscriber<? super T> call(final Subscriber<? super T> o) {
        CompositeSubscription parent = new CompositeSubscription();
        if (limit == 0) {
            o.onCompleted();
            parent.unsubscribe();
        }
        return new Subscriber<T>(parent) {

            int count = 0;
            boolean completed = false;

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

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

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

        };
    }

}

4) Recursion/Loop Performance

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

Several places we can remove recursive scheduling used originally for unsubscribe support and use a loop instead.

Schedulers

Schedulers were greatly simplified to a design based around Action1<Inner>.

public abstract class Scheduler {
    public Subscription schedule(Action1<Scheduler.Inner> action);
    public Subscription schedule(Action1<Scheduler.Inner> action, long delayTime, TimeUnit unit);
    public Subscription schedulePeriodically(Action1<Scheduler.Inner> action, long initialDelay, long period, TimeUnit unit);
    public long now();
    public int degreeOfParallelism();

    public static class Inner implements Subscription {
        public abstract void schedule(Action1<Scheduler.Inner> action, long delayTime, TimeUnit unit);
        public abstract void schedule(Action1<Scheduler.Inner> action);
        public long now();
    }
}

This design change originated from three findings:

1. It was very easy to cause memory leaks or inadvertent parallel execution since the distinction between outer and inner scheduling was not obvious.

To solve this the new design explicitly has the outer Scheduler and then Scheduler.Inner for recursion.

2. The passing of state is not useful since scheduling over network boundaries with this model does not work.

In this new design all state passing signatures have been removed. This was determined while implementing a RemoteScheduler that attempted to use observeOn to transition execution from one machine to another. This does not work because of the requirement for serializing/deserializing the state of the entire execution stack. Migration of work over the network has been bound to be better suited to explicit boundaries established by Subjects. Thus, the complications within the Schedulers are unnecessary.

3. The number of overloads with different ways of doing the same things were confusing.

This new design removes all but the essential and simplest methods.

4. A scheduled task could not do work in a loop and easily be unsubscribed which generally meant less efficient recursive scheduling.

This new design applies similar principles as done with lift/create/OnSubscribe/Subscriber and injects the Subscription via the Inner interface so a running task can check isUnsubscribed().

WIth this new design, the simplest execution of a single task is:

Schedulers.newThread().schedule(new Action1<Inner>() {

    @Override
    public void call(Inner inner) {
        doWork();
    }

});

Recursion is easily invoked:

Schedulers.newThread().schedule(new Action1<Inner>() {

    @Override
    public void call(Inner inner) {
        doWork();
        // recurse until unsubscribed (the schedule will do nothing if unsubscribed)
        inner.schedule(this);
    }

});

The use of Action1<Inner> on both the outer and inner levels makes it so recursion that refer to this and it works easily.

Similar to the new lift/create pattern with Subscriber the Inner is also a Subscription so it allows efficient loops with unsubscribe support:

Schedulers.newThread().schedule(new Action1<Inner>() {

    @Override
    public void call(Inner inner) {
        while(!inner.isUnsubscribed()) {
            doWork();
        }
    }

});

An action can now unsubscribe the Scheduler.Inner:

Schedulers.newThread().schedule(new Action1<Inner>() {

    @Override
    public void call(Inner inner) {
        while(!inner.isUnsubscribed()) {
            int i = doOtherWork();
            if(i > 100) {
                // an Action can cause the Scheduler to unsubscribe and stop
                inner.unsubscribe();
            }
        }
    }

});

Typically just stopping is sufficient:

Schedulers.newThread().schedule(new Action1<Inner>() {

    @Override
    public void call(Inner inner) {
        int i = doOtherWork();
        if (i < 10) {
            // recurse until done 10
            inner.schedule(this);
        }
    }

});

but if other work in other tasks is being done and you want to unsubscribe conditionally you could:

Schedulers.newThread().schedule(new Action1<Inner>() {

    @Override
    public void call(Inner inner) {
        int i = doOtherWork();
        if (i < 10) {
            // recurse until done 10
            inner.schedule(this);
        } else {
            inner.unsubscribe();
        }
    }

});

and the recursion can be delayed:

Schedulers.newThread().schedule(new Action1<Inner>() {

    @Override
    public void call(Inner inner) {
        doWork();
        // recurse until unsubscribed ... but delay the recursion
        inner.schedule(this, 500, TimeUnit.MILLISECONDS);
    }

});

The methods on the Inner never return a Subscription because they are always a single thread/event-loop/actor/etc and controlled by the Subscription returned by the initial Scheduler.schedule method. This is part of clarifying the contract.

Thus an unsubscribe controlled from the outside would be done like this:

Subscription s = Schedulers.newThread().schedule(new Action1<Inner>() {

    @Override
    public void call(Inner inner) {
        while(!inner.isUnsubscribed()) {
            doWork();
        }
    }

});

// unsubscribe from outside
s.unsubscribe();

Migration Path

1) Lift/OnSubscribe/Subscriber

The lift function will not be used by most and is additive so will not affect backwards compatibility. The Subscriber type is also additive and for most use cases does not need to be used directly, the Observer interface can continue being used.

The previous create(OnSubscribeFunc f) signature has been deprecated so code will work but now have warnings. Please begin migrating code as this will be deleted prior to the 1.0 release.

Code such as this:

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

    @Override
    public Subscription onSubscribe(Observer<? super Integer> o) {
        o.onNext(1);
        o.onNext(2);
        o.onCompleted();
        return Subscriptions.empty();
    }
});

should change to this:

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

    @Override
    public void call(Subscriber<? super Integer> subscriber) {
        subscriber.onNext(1);
        subscriber.onNext(2);
        subscriber.onCompleted();
    }
});

If concurrency was being injected:

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

    @Override
    public Subscription onSubscribe(final Observer<? super Integer> o) {
        final BooleanSubscription s = new BooleanSubscription();
        Thread t = new Thread(new Runnable() {

            @Override
            public void run() {
                int i = 0;
                while (s.isUnsubscribed()) {
                    o.onNext(i++);
                }
            }

        });
        t.start();
        return s;
    }
});

you may no longer need it and can implement like this instead:

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

    @Override
    public void call(Subscriber<? super Integer> subscriber) {
        int i = 0;
        while (subscriber.isUnsubscribed()) {
            subscriber.onNext(i++);
        }
    }
});

or can just simplify the Subscription management:

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

    @Override
    public void call(final Subscriber<? super Integer> subscriber) {
        Thread t = new Thread(new Runnable() {

            @Override
            public void run() {
                int i = 0;
                while (subscriber.isUnsubscribed()) {
                    subscriber.onNext(i++);
                }
            }

        });
        t.start();
    }
});

or use a Scheduler:

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

    @Override
    public void call(final Subscriber<? super Integer> subscriber) {
        Schedulers.io().schedule(new Action1<Inner>() {

            @Override
            public void call(Inner inner) {
                int i = 0;
                while (subscriber.isUnsubscribed()) {
                    subscriber.onNext(i++);
                }
            }

        });
    }
});

or use subscribeOn which now works to make synchronous Observables async while supporting unsubscribe (this didn't work before):

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

    @Override
    public void call(Subscriber<? super Integer> subscriber) {
        int i = 0;
        while (subscriber.isUnsubscribed()) {
            subscriber.onNext(i++);
        }
    }
}).subscribeOn(Schedulers.newThread());

2) Schedulers

Custom Scheduler implementations will need to be re-implemented and any direct use of the Scheduler interface will also need to be updated.

3) Subscription

If you have custom Subscription implementations you will see they now need an isUnsubscribed() method.

You can either add this method, or wrap your function using Subscriptions.create and it will handle the isUnsubscribed behavior and execute your function when unsubscribe() is called.

The Future...

This is hopefully the last of the major refactors to rxjava-core and we're approaching version 1.0. We have most if not all operators from Rx.Net that we want or intend to port. We think we have got the create/subscribe signatures as we want and the Subscription and Scheduler interfaces are now clean.

We need to improve on some of the Subject implementations still, particularly ReplaySubject. We are beginning to focus after this release on cleaning up all of the operator implementations, stabilizing, fixing bugs and performance tuning.

We appreciate your usage, feedback and contributions and hope the library is creating value for you!

[benjchristensen]

Sometime this or next week I hope to release 0.17. These release notes try to communicate what changed, why and how to migrate code that is impacted by the changes. Please let me know what you feel needs to be added/changed for it to be clearer or more useful.

Also, now is the time for any final feedback on the changes in 0.17 before they are released and become the published API.

The major changes in this release are:

• Proposal: Lift Operator (was 'bind') #746 lift and create(OnSubscribe)
• Observer and/or Subscriber? #792 Subscriber
• Scheduler Outer/Inner [Preview] #797 Scheduler Redesign

[zsxwing]

I have a question about Operator.

Now Operator is only a Func1<Subscriber<? super R>, Subscriber<? super T>>, Observable can not be touched in Operator. So if some Operators like repeat need to subscribe the original Observable more than once, how to do it?

Of course, the Operator can get the Observable through the constructor. But lift will not be able to chain it with other operators. For example, we need to break the chain.

o = observable.map(...).filter(...).take(5)
o.lift(new MyCustomOperator(o)).map(...).subscribe()

[abersnaze]

Just because it can't be written in a single line of code doesn't mean that it isn't chained like the others.

[benjchristensen]

So if some Operators like repeat need to subscribe the original Observable more than once, how to do it?

By converting to Observable<Observable>> so lift has access to the inner Observable.

See a prototype example here: https://gist.github.com/benjchristensen/8367765#file-obsurvable-java-L132

    public Obsurvable<T> repeat() {
        return from(this).bind(new RepeatOperator<T>());
    }

Here is a pull request implementing zip using lift: https://github.com/benjchristensen/RxJava/blob/100983d29b5890fa604ae76f2c3dd29114936621/rxjava-core/src/main/java/rx/Observable.java#L3126

    public final static <T1, T2, R> Observable<R> zip(Observable<? extends T1> o1, Observable<? extends T2> o2, final Func2<? super T1, ? super T2, ? extends R> zipFunction) {
        return just(new Observable<?>[] { o1, o2 }).lift(new OperatorZip<R>(zipFunction));
    }

[benjchristensen]

Just because it can't be written in a single line of code doesn't mean that it isn't chained like the others.

It wouldn't take much to allow it being written in a single line. All we need is a simple function/operator that goes Observable<T> -> Observable<Observable<T>>.

For example, if we called it toNestedObservable() we could then do this:

o = observable.map(...).filter(...).take(5).toNestedObservable().lift(new MyCustomOperator()).map(...).subscribe()

The implementation of the function would simply be:

public Observable<Observable<T>> toNestedObservable() {
      return from(this);
}

[zsxwing]

Thanks for your clarification. I feel Observable<Observable<T>> is not very intuitive, but I think it's the best solution in current design.

[benjchristensen]

We deal with Observable<Observable<T>> on any combinatorial operator such as zip, merge, concat and others like groupBy, switch, etc thus it is a very common type to deal with. What about it do you feel is not intuitive?

Also, it should be very rare that someone will need to use lift directly and worry about this for nested behavior as generally the existing operators would be leveraged to groupBy, merge, concat, zip etc and then custom operators applied to the Observable<T> or Observable<Observable<T>> that comes out of those.

[zsxwing]

I feel it's not intuitive because, for example, repeat is repeating the values in Observable<T>, but lift it on Observable<Observable<T>> looks like emitting Observable<T>s. But I'm convinced by Operator is rarely used by the users.

[benjchristensen]

If you consider what lift is doing then I see it as intuitive. The function being "lifted" is acting on T, thus when T is an Observable<T> it is acting on the Observable<T> that is emitted.

That's the elegance of lift. The operators always act on type T and emit type R and are statically typed to whatever type T they accept. In the case of repeat it accepts an Observable<T> so that means it can be "lifted" against an Observable<Observable<T>> since that will emit an Observable<T> to it.

[zsxwing]

Thanks for your explanation. Really cleared my confusion.

[benjchristensen]

Glad it was helpful, and thank you for your involvement and being willing to ask and share your opinion!

[zsxwing]

I'm trying to implement the OperatorSubscribeOn to fix #813. Here is the prototype code:

public class OperatorSubscribeOn<T> implements Operator<T, Observable<T>> {

    private final Scheduler scheduler;

    public OperatorSubscribeOn(Scheduler scheduler) {
        this.scheduler = scheduler;
    }

    @Override
    public Subscriber<? super Observable<T>> call(final Subscriber<? super T> subscriber) {
        return new Subscriber<Observable<T>>() {

            @Override
            public void onCompleted() {
            }

            @Override
            public void onError(Throwable e) {
            }

            @Override
            public void onNext(final Observable<T> o) {
                scheduler.schedule(new Action1<Inner>() {

                    @Override
                    public void call(final Inner inner) {
                        if(!inner.isUnsubscribed()) {
                            o.subscribe(subscriber);
                        }
                    }

                });
            }

        };
    }
}

However, I encounter a problem. I can not dispatch the unsubscribe action to the scheduler. One way to implement it is overriding Subscriber.unsubscribe, but it is final.

Is there any way that an Operator can change the unsubscribe behavior?

[zsxwing]

Just find I can create a Subscriber with a CompositeSubscription to change the unsubscribe behavior. Like all of the ideas in this release. Really powerful.

[benjchristensen]

Yes, a Subscriber allows passing in a CompositeSubscription or Subscriber (to get its Subscription), or you can add any other Subscription to a Subscriber to create the chains in whatever way is needed.

[benjchristensen]

Outstanding items I want resolved or completed before releasing include:

• GroupBy functional bugs
  Some of the unit tests are actually asserting incorrect behavior (for example, unsubscribing of all current children should not cause the parent to unsubscribe).
• ObserveOn performance [Will work on in 0.17.x]
  Despite using a ring buffer in the current implementation, something seems wrong that the performance was hit as hard during the recent changes. Despite the changes being necessary to provide correct behavior for back pressure, it should be possible to improve behavior in that case and regain full performance if someone chooses to allow buffering.
• Parallel operator
  It appears there is some non-determism in when the onComplete/unsubscribe occurs and that it is related to groupBy. The parallel operator uses both groupBy and observeOn so it is impacted by changes in 0.17.
• Zip with back pressure [Not Doing in 0.17.0]
  The zip operator is typically the biggest issue (other than observeOn) for back pressure and we want a solution. I want it in 0.17 as it will likely be a big enough change to warrant a 0.x version change. Removing the buffers in zip fix back pressure but result in deadlocks when zipping synchronous Observables thus the obvious change can not be done. Likely it is appropriate in this case to inject concurrency so that each stream can synchronously push data while the zip operator itself can subscribe to multiple streams. We may be able to conditionally not inject extra concurrency if we find a source already on a different thread.
• Scheduler.periodicDelay
  Change signature from:

public Subscription schedulePeriodically(final Action1<Scheduler.Inner> action, long initialDelay, long period, TimeUnit unit) 

to

public Subscription schedulePeriodically(final Action1<Scheduler.Inner> action, long initialDelay, , TimeUnit unit, long period, TimeUnit unit) 

• Debug plugin functional (all hooks proven)
  Make sure that the execution hook plugin supports the desired debug capabilities along with the new lift, Subscriber and OnSubscribe behavior. See Setting up the new subproject for debugging observable chains. #836 for first attempts.
• Production canary testing
  I intend on running 0.17 on Netflix production canaries before releasing. I can do a release candidate if there is interest, but a final release needs to be proven to work at scale. This does not mean all operators get equally battle-tested as Netflix code does not use all of them, but it does prove out the core operators and main code flows (such as lift, OnSubscribe, Subscriber, etc).

Please let me know if you have things you feel are needed before 0.17 are released. I'd also appreciate insight or help on achieving the above goals.

[benjchristensen]

@abersnaze After the merge of #836 are you comfortable with "Debug plugin functional (all hooks proven)"?

[benjchristensen]

Marked "Debug plugin functional (all hooks proven)" as completed as the changes from @abersnaze are merged and we reviewed together and they look good. They resulted in #857 changing the signature to public <R> Observable<R> lift(final Operator<R, T> bind).

[benjchristensen]

"Scheduler.periodicDelay" is done in #856 ... open for debate on whether it should be merged or not.

[benjchristensen]

"GroupBy functional bugs" is more-or-less done I believe, except for determining whether the subscribeOn "time gap" #844 issue needs a solution. I'm leaving it unfinished until that is figured out.

[benjchristensen]

Other than the final things being worked on, does anyone else have things you feel are needed in 0.17, or that have happened in 0.17 that you disagree with (for objective reasons)?

[akarnokd]

Can we have an issue for "Zip with back pressure" with all the requirements so discussions won't detract this issue?

[benjchristensen]

Here you go: Zip Back-pressure #867

[benjchristensen]

I'm not going to hold up 0.17.0 for zip. It's massive enough as it is. I want to get a release candidate out. I'd like to just wrap up the groupBy/subscribeOn stuff and start doing canary testing.

[benjchristensen]

The groupBy and parallel operators are working well enough to move forward I believe: #869

My only remaining concern is performance for observeOn, but that can be worked on going forward.

Once I confirm a test (#869 (comment)) I will merge this last pull request and release a candidate (0.17.0-RC1) for testing and if that goes well release the real thing (0.17.0).

[benjchristensen]

Released 0.17.0-RC1 for testing: https://github.com/Netflix/RxJava/releases/tag/0.17.0-RC1

Starting the canary testing on our end. Once we have proven we can serve our production traffic on this code I will release 0.17.0. I am only merging bug fixes for now before tackling new features and design improvements.

[benjchristensen]

Verison 0.17.0-RC2 has been released for testing: https://github.com/Netflix/RxJava/releases/tag/0.17.0-RC2

/cc @samuelgruetter @jmhofer @mttkay @daveray for feedback on whether all is well with language adaptors, Android and Swing modules.

Testing at Netflix is continuing, I'd appreciate your feedback before calling this version ready for production release due to it's massive changes in subscription and scheduling behavior.

[benjchristensen]

Of interest, while hunting down the observeOn performance issue, it actually seems to be related to the NewThreadScheduler and not observeOn itself.

[benjchristensen]

As of now I am not aware of any further changes needed for 0.17.0 so I'm looking for bug reports and fixes.

[benjchristensen]

Thanks @daveray for the confirmation.

[benjchristensen]

0.17.0 RC6 was just released. Testing in dev is going well though I have not yet been able to canary it in our production environment as we have some yak shaving to do to get it into our production codebase due to the signature changes. They are a side-effect of a hack we did over a year ago when we moved from internal to open source version and that hack has caught up to us with the lift/Subscriber changes so we have a few days more of work before I can validate this release. That is why I have not yet released 0.17.0 as I can't assert it is production worthy.

Is anyone else doing testing to validate this release is working for you? I would particularly be interested in experience from Android developers.

[mttkay]

Ben, I set a day aside today to focus on this. We haven't yet moved to 0.17
due to other priorities but will work on it today.

[mttkay]

Looking good so far. Would like to put out a beta with these changes in place to be sure.

One thing I had issues with:

Observables are not mockable anymore using Mockito, since the f field that holds the OnSubscribe function is always null, and it's final to can't be overridden. I realize observables aren't great candidates to be mocked out to begin with, since you can quickly end up in train wreck mocks, but it can be convenient at times (for instance, to test whether subscribe was called with a specific observer type, verify is quite useful.) I had to build a custom mock class now which traces all subscribers so that I can keep verifying that observables returned by peer objects are actually being subscribed to, as opposed to just calling the method that created the sequence.

More feedback soon.

[mttkay]

Forgot to mention: this might be an IntelliJ issue, but whenever I subscribe a Subscriber to an observable, IntelliJ now complains:

Ambiguous method call: both subscribe Observer and subscribe Subscriber in Observable match

However, it builds fine on the command line. Anyone else getting this? I think the problem is that a Subscriber is an Observer too. Don't remember exactly how Java handles this, I guess one would assume it picks the most specific signature match at runtime? Anyone else getting this warning?

It's a bit annoying, because I actually configured my IDE to display warnings as errors, so the RxJava code is bleeding all over the place.

[benjchristensen]

Thank you for taking the time to play with it.

IntelliJ issue

That seems very wrong for IntelliJ to do that. Eclipse doesn't have the issue.

I'm not sure what we would change since it seems fairly normal to have method overloads such as this.

Observables are not mockable anymore

They were never very good at being mocked anyways. There is a new TestSubscriber that I find very useful for unit tests. Can you look at it and suggest improvements that would allow the assertions you're trying to do?

[benjchristensen]

We could eliminate this method:

public final Subscription subscribe(Subscriber<? super T> observer)

... and do an instanceof check on Observer and then treat it accordingly, but that feels just wrong and doesn't correctly communicate via the API that Subscriber is intended for use with the subscribe() method.

And I don't want to delete the subscribe(Observer<? super T> observer) method, as it's fine to subscribe with an Observer if the added functionality of Subscriber is not needed.

If I had to choose between them, I'd rather kill subscribe(Observer<? super T> observer).

I'd really rather just make IntelliJ not be broken.

[akarnokd]

@mttkay Which version of IntelliJ is that? I had similar trouble with the 12.x series, unrelated to RxJava. I think it is (still) caused by the incomplete and/or bogus Java 8 support.

[benjchristensen]

@akarnokd Do you not see this issue? Are you on 13.x? I'm curious because RxJava (or Android) doesn't involve Java 8 and thus shouldn't be affected by that bug.

[headinthebox]

I get the same, but it really does not matter since it compiles fine. I would not change anything for that.

[headinthebox]

The once place I get the spurious error is in merge:

@Override
            public void onNext(Observable<? extends T> innerObservable) {
                runningCount.incrementAndGet();
                Subscriber<T> i = new InnerObserver();
                childrenSubscriptions.add(i);
                innerObservable.subscribe(i); // <===
            }

[mttkay]

This is on 13.0.2. It's just a warning, not an error, but still. I actually
did install Java 8 recently, let me check whether that's the issue.

[akarnokd]

@mttkay This compiler issue was the reason I turned away from IJ to NetBeans; I'm working on a lambda intensive code and got tired IJ complaining about most of it.

[benjchristensen]

We have a solid proposal from @abersnaze that provides "pause" functionality on Subscription to allow natural back-pressure on things like reading files or Iterables, and since it involves signature changes to Subscription I'm holding off on releasing 0.17 a little longer (and the fact that we still haven't confirmed the release in our production environment). I'd rather have 0.17 with all of those changes since we already have changed Subscription. The prototype he demonstrated to @headinthebox and I today works with zip, take, map and conceptually with merge and thus flatMap.

More to come in the next few days.

[benjchristensen]

Does everyone want to wait for Netflix to finish testing this release in prod before officially releasing, or are enough of you content with RC 6 that I should release it and let further improvements come in 0.17.x?

[benjchristensen]

Any takers on giving a yes/no on whether you feel 0.17.0 RC6 is ready for your production use?

[akarnokd]

My production use currently consists of PublishSubject only, so it is a go for me. However, we should consider reimplementing the most-used operations as operators before releasing.

[mttkay]

I've used it in development builds for about a week now and didn't have any
problems either.

We do not use any operators beyond map, mergeMap, cache, publish and
replay, however.

+1

[samueltardieu]

I've been using it on Android since RC2 for the cgeo development branch and nightly releases. We use AndroidObservable a lot, and operators merge, flatMap, parallel, reduce, toBlockingObservable, single, onErrorResumeNext, from, filter, publish, refcount, lastOrDefault as well as PublishSubject and BehaviourSubject. Also, schedulePeriodically and schedule on Scheduler instances.

So far, everything seems to be working well.

[benjchristensen]

Great, thank you for your feedback.

we should consider reimplementing the most-used operations as operators before releasing.

@akarnokd I feel that will take far too long and introduce more risk of delaying for another month. The full benefits of the lift changes won't occur until we do this, but I'd rather do them in 0.17.x incremental releases. Are you okay with that?

[AlexeyRaga]

We tried to use it, no good so far since there are problems when working with iterables (we need to observe queues and Kafka topics). What makes it worse is that it looks like everything works (observer gets data that seems right) until you start digging.
Of course I prefer it to be fixed.

[benjchristensen]

there are problems when working with iterables

Are you referring to back pressure (buffer bloat) or something that was working in 0.16 and no longer works in 0.17?

[akarnokd]

@benjchristensen Yes, I guess the top 10 operators with overloads would take a few weeks. Regardless, I would like to start working on the reimplementations en-masse.

[benjchristensen]

Regardless, I would like to start working on the reimplementations en-masse.

Yes, it will be happening soon. Let's not quite start until we get 0.17.0 out the door and agree upon the coding patterns. I also want to nail down our answer for back pressure.

[benjchristensen]

Despite not being able to fully validate in our prod environment (for severe yak shaving reasons) I don't want to hold this up any further and believe we have validate enough to move forward. Bugs (which will surely be found) will be fixed in 0.17.x releases along with re-implementaitons of the many operators that are still on the old OnSubscribeFunc mechanism.

I'll revise the release notes and release this soon.

[benjchristensen]

RxJava 0.17.0-RC7 is released. Some small issues arose with debug hooks I want to validate before making this 0.17.0 Release.

[akarnokd]

@benjchristensen in the new release notes, can you fix the while(s.isUnsubscribed()) to while(!s.isUnsubscribed()) among the examples.

[benjchristensen]

Ha, good catch, just fixed.

I wrote the code in an IDE to make sure it compiled, but didn't test that part!