flatMap with maxConcurrent value of 1 results in NPE if Observable emitting null is returned.

[sfitts]

The easiest way to reproduce this is as follows:

    @Test
    public void testNullFromRange() {
        TestSubscriber<Object> subscriber = new TestSubscriber<>();
        Observable.range(0, 2).flatMap(i -> {
            if (i == 0) {
                return Observable.interval(1, TimeUnit.SECONDS).take(1);
            } else {
                return Observable.just(null);
            }
        }, 1).subscribe(subscriber);
        subscriber.awaitTerminalEvent();
        subscriber.assertNoErrors();
    }

In this case the value from Observable.just(null) cannot be immediate emitted so it is queued. Since the observable is a scalar we queue the underlying value (in this case null) which results in a call to offer on the class SpscArrayQueue. This class does not accept null element values so the result is an NPE.

Removing the maxConcurrent argument fixes this, but unfortuantely in our case that's not possible as we need have a range that we need to run sequentially.

[akarnokd]

This has been fixed in 1.1.3. Please make sure you use the latest version of RxJava.

[sfitts]

Thanks - I'm using this via RxVertx so let me see if I can upgrade the underlying RxJava version without breaking something else. Good to know it is fixed however.

[sfitts]

The other issue I have is that I'm using RxGroovy -- the most recent release there appears to be 1.0.3 -- is this compatible with the 1.1.x releases of RxJava (it's build.gradle lists a dependency of 1.0.+ which might imply no). If not are there plans to update RxGroovy to the same level as RxJava?

[akarnokd]

They were some removals but I don't know much about RxGroovy. I suggest you talk to them about upgrading.

[sfitts]

Do you know who they are? There is an issue for the upgrade which has been around since Feb and the contact info is the same as the main RxJava project. I'm happy to do the update and submit a PR, but I'm not sure if there is anyone to review/merge it if I do.

[akarnokd]

I have the rights to merge and @zsxwing is there to, but can't give any promise on releasing anything. You could also fork RxGroovy and release on your own terms to maven.

[akarnokd]

Continued at ReactiveX/RxGroovy#13