Filter breaks backpressure

[boris-petrov]

The following code:

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

import rx.Observable;
import rx.Subscription;

public class Test {
    public static void main(String[] arguments) throws InterruptedException {
        new Test().test();
    }

    void test() throws InterruptedException {
        CountDownLatch latch = new CountDownLatch(1);

        Subscription subscription = createBackpressuredObservable()
            .filter(item -> true)
            .mergeWith(createBackpressuredObservable())
            .subscribe(
                item -> {},
                e -> {
                    e.printStackTrace();
                    latch.countDown();
                },
                latch::countDown);
        latch.await(7, TimeUnit.SECONDS);
        subscription.unsubscribe();
    }

    private Observable<Integer> createBackpressuredObservable() {
        return Observable
            .<Integer> create(subscriber -> {
                new Thread(() -> {
                    while (!subscriber.isUnsubscribed()) {
                        subscriber.onNext(0);
                    }
                    if (!subscriber.isUnsubscribed()) {
                        subscriber.onCompleted();
                    }
                }).start();
            })
            .onBackpressureBuffer();
    }
}

Works with RxJava version 1.1.3, but breaks in 1.1.4 with a rx.exceptions.MissingBackpressureException. If you remove the filter on line 16, it works fine on both versions. I'm not sure I understand everything about how backpressure works, but I think this is a bug. Please correct me if I'm wrong.

As a side note, the documentation on backpressure, how it is propagated and which operators use reactive pull backpressure internally is really scarce and difficult to follow.

[akarnokd]

Thanks for reporting. Fix in #3912.

[akarnokd]

Fix released with 1.1.5. Should be available from maven within a few hours.