src/test/java/io/reactivex/rxjava4/flowable/FlowableBackpressureTests.java - 4.x - githubmirror/RxJava

src/test/java/io/reactivex/rxjava4/flowable/FlowableBackpressureTests.java
669 строк27 KB
David Karnok
4.x: Convert from JUnit 4 to JUnit 6 (#8202)
30 июн 2026, 12:01
30 июн 2026, 12:01531388f
/*
 * Copyright (c) 2016-present, RxJava Contributors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in
 * compliance with the License. You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software distributed under the License is
 * distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See
 * the License for the specific language governing permissions and limitations under the License.
 */

package io.reactivex.rxjava4.flowable;

import static org.junit.jupiter.api.Assertions.*;

import java.io.Serial;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.Flow.*;
import java.util.concurrent.atomic.*;

import org.junit.jupiter.api.*;

import io.reactivex.rxjava4.core.*;
import io.reactivex.rxjava4.exceptions.QueueOverflowException;
import io.reactivex.rxjava4.functions.Function;
import io.reactivex.rxjava4.internal.subscriptions.SubscriptionHelper;
import io.reactivex.rxjava4.internal.util.BackpressureHelper;
import io.reactivex.rxjava4.schedulers.Schedulers;
import io.reactivex.rxjava4.subscribers.*;

public class FlowableBackpressureTests extends RxJavaTest {

    static final class FirehoseNoBackpressure extends AtomicBoolean implements Subscription {

        @Serial
        private static final long serialVersionUID = -669931580197884015L;
        final Subscriber<? super Integer> downstream;
        final AtomicInteger counter;
        volatile boolean cancelled;

        private FirehoseNoBackpressure(AtomicInteger counter, Subscriber<? super Integer> s) {
            this.counter = counter;
            this.downstream = s;
        }

        @Override
        public void request(long n) {
            if (!SubscriptionHelper.validate(n)) {
                return;
            }
            if (compareAndSet(false, true)) {
                int i = 0;
                final Subscriber<? super Integer> a = downstream;
                final AtomicInteger c = counter;

                while (!cancelled) {
                    a.onNext(i++);
                    c.incrementAndGet();
                }
                System.out.println("unsubscribed after: " + i);
            }
        }

        @Override
        public void cancel() {
            cancelled = true;
        }
    }

    @AfterEach
    public void doAfterTest() {
        // FIXME LATER
//        TestObstructionDetection.checkObstruction();
    }

    @Test
    public void observeOn() {
        int num = (int) (Flowable.bufferSize() * 2.1);
        AtomicInteger c = new AtomicInteger();
        TestSubscriber<Integer> ts = new TestSubscriber<>();
        incrementingIntegers(c).observeOn(Schedulers.computation()).take(num).subscribe(ts);
        ts.awaitDone(5, TimeUnit.SECONDS);
        ts.assertNoErrors();
        System.out.println("testObserveOn => Received: " + ts.values().size() + "  Emitted: " + c.get());
        assertEquals(num, ts.values().size());
        assertTrue(c.get() < Flowable.bufferSize() * 4);
    }

    @Test
    public void observeOnWithSlowConsumer() {
        int num = (int) (Flowable.bufferSize() * 0.2);
        AtomicInteger c = new AtomicInteger();
        TestSubscriber<Integer> ts = new TestSubscriber<>();
        incrementingIntegers(c).observeOn(Schedulers.computation()).map(
            i -> {
                try {
                    Thread.sleep(1);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                return i;
            }
        ).take(num).subscribe(ts);
        ts.awaitDone(5, TimeUnit.SECONDS);
        ts.assertNoErrors();
        System.out.println("testObserveOnWithSlowConsumer => Received: " + ts.values().size() + "  Emitted: " + c.get());
        assertEquals(num, ts.values().size());
        assertTrue(c.get() < Flowable.bufferSize() * 2);
    }

    @Test
    public void mergeSync() {
        int num = (int) (Flowable.bufferSize() * 4.1);
        AtomicInteger c1 = new AtomicInteger();
        AtomicInteger c2 = new AtomicInteger();
        TestSubscriber<Integer> ts = new TestSubscriber<>();
        Flowable<Integer> merged = Flowable.mergeArray(incrementingIntegers(c1), incrementingIntegers(c2));

        merged.take(num).subscribe(ts);
        ts.awaitDone(5, TimeUnit.SECONDS);
        ts.assertNoErrors();
        System.out.println("Expected: " + num + " got: " + ts.values().size());
        System.out.println("testMergeSync => Received: " + ts.values().size() + "  Emitted: " + c1.get() + " / " + c2.get());
        assertEquals(num, ts.values().size());
        // either one can starve the other, but neither should be capable of doing more than 5 batches (taking 4.1)
        // TODO is it possible to make this deterministic rather than one possibly starving the other?
        // benjchristensen => In general I'd say it's not worth trying to make it so, as "fair" algorithms generally take a performance hit
        assertTrue(c1.get() < Flowable.bufferSize() * 5);
        assertTrue(c2.get() < Flowable.bufferSize() * 5);
    }

    @Test
    public void mergeAsync() {
        int num = (int) (Flowable.bufferSize() * 4.1);
        AtomicInteger c1 = new AtomicInteger();
        AtomicInteger c2 = new AtomicInteger();
        TestSubscriber<Integer> ts = new TestSubscriber<>();
        Flowable<Integer> merged = Flowable.mergeArray(
                incrementingIntegers(c1).subscribeOn(Schedulers.computation()),
                incrementingIntegers(c2).subscribeOn(Schedulers.computation()));

        merged.take(num).subscribe(ts);
        ts.awaitDone(5, TimeUnit.SECONDS);
        ts.assertNoErrors();
        System.out.println("testMergeAsync => Received: " + ts.values().size() + "  Emitted: " + c1.get() + " / " + c2.get());
        assertEquals(num, ts.values().size());
        // either one can starve the other, but neither should be capable of doing more than 5 batches (taking 4.1)
        // TODO is it possible to make this deterministic rather than one possibly starving the other?
        // benjchristensen => In general I'd say it's not worth trying to make it so, as "fair" algorithms generally take a performance hit
        int max = Flowable.bufferSize() * 7;
        assertTrue(c1.get() < max, "" + c1.get() + " >= " + max);
        assertTrue(c2.get() < max, "" + c2.get() + " >= " + max);
    }

    @Test
    public void mergeAsyncThenObserveOnLoop() {
        for (int i = 0; i < 500; i++) {
            if (i % 10 == 0) {
                System.out.println("testMergeAsyncThenObserveOnLoop >> " + i);
            }
            // Verify there is no MissingBackpressureException
            int num = (int) (Flowable.bufferSize() * 4.1);
            AtomicInteger c1 = new AtomicInteger();
            AtomicInteger c2 = new AtomicInteger();

            TestSubscriber<Integer> ts = new TestSubscriber<>();
            Flowable<Integer> merged = Flowable.mergeArray(
                    incrementingIntegers(c1).subscribeOn(Schedulers.computation()),
                    incrementingIntegers(c2).subscribeOn(Schedulers.computation()));

            merged
            .observeOn(Schedulers.cached())
            .take(num)
            .subscribe(ts);

            ts.awaitDone(5, TimeUnit.SECONDS);
            ts.assertComplete();
            ts.assertNoErrors();
            System.out.println("testMergeAsyncThenObserveOn => Received: " + ts.values().size() + "  Emitted: " + c1.get() + " / " + c2.get());
            assertEquals(num, ts.values().size());
        }
    }

    @Test
    public void mergeAsyncThenObserveOn() {
        int num = (int) (Flowable.bufferSize() * 4.1);
        AtomicInteger c1 = new AtomicInteger();
        AtomicInteger c2 = new AtomicInteger();
        TestSubscriber<Integer> ts = new TestSubscriber<>();
        Flowable<Integer> merged = Flowable.mergeArray(
                incrementingIntegers(c1).subscribeOn(Schedulers.computation()),
                incrementingIntegers(c2).subscribeOn(Schedulers.computation()));

        merged.observeOn(Schedulers.newThread()).take(num).subscribe(ts);
        ts.awaitDone(5, TimeUnit.SECONDS);
        ts.assertNoErrors();
        System.out.println("testMergeAsyncThenObserveOn => Received: " + ts.values().size() + "  Emitted: " + c1.get() + " / " + c2.get());
        assertEquals(num, ts.values().size());
        // either one can starve the other, but neither should be capable of doing more than 5 batches (taking 4.1)
        // TODO is it possible to make this deterministic rather than one possibly starving the other?
        // benjchristensen => In general I'd say it's not worth trying to make it so, as "fair" algorithms generally take a performance hit
        // akarnokd => run this in a loop over 10k times and never saw values get as high as 7*SIZE, but since observeOn
        // delays the unsubscription non-deterministically, the test will remain unreliable
        assertTrue(c1.get() < Flowable.bufferSize() * 7);
        assertTrue(c2.get() < Flowable.bufferSize() * 7);
    }

    @Test
    public void flatMapSync() {
        int num = (int) (Flowable.bufferSize() * 2.1);
        AtomicInteger c = new AtomicInteger();
        TestSubscriber<Integer> ts = new TestSubscriber<>();

        incrementingIntegers(c)
        .flatMap((Function<Integer, Publisher<Integer>>) _ -> incrementingIntegers(new AtomicInteger()).take(10))
        .take(num).subscribe(ts);

        ts.awaitDone(5, TimeUnit.SECONDS);
        ts.assertNoErrors();
        System.out.println("testFlatMapSync => Received: " + ts.values().size() + "  Emitted: " + c.get());
        assertEquals(num, ts.values().size());
        // expect less than 1 buffer since the flatMap is emitting 10 each time, so it is num/10 that will be taken.
        assertTrue(c.get() < Flowable.bufferSize());
    }

    @Test
    public void zipSync() {
        int num = (int) (Flowable.bufferSize() * 4.1);
        AtomicInteger c1 = new AtomicInteger();
        AtomicInteger c2 = new AtomicInteger();
        TestSubscriber<Integer> ts = new TestSubscriber<>();

        Flowable<Integer> zipped = Flowable.zip(
                incrementingIntegers(c1),
                incrementingIntegers(c2),
                Integer::sum);

        zipped.take(num)
        .subscribe(ts);

        ts.awaitDone(5, TimeUnit.SECONDS);
        ts.assertNoErrors();
        System.out.println("testZipSync => Received: " + ts.values().size() + "  Emitted: " + c1.get() + " / " + c2.get());
        assertEquals(num, ts.values().size());
        assertTrue(c1.get() < Flowable.bufferSize() * 7);
        assertTrue(c2.get() < Flowable.bufferSize() * 7);
    }

    @Test
    public void zipAsync() {
        int num = (int) (Flowable.bufferSize() * 2.1);
        AtomicInteger c1 = new AtomicInteger();
        AtomicInteger c2 = new AtomicInteger();
        TestSubscriber<Integer> ts = new TestSubscriber<>();
        Flowable<Integer> zipped = Flowable.zip(
                incrementingIntegers(c1).subscribeOn(Schedulers.computation()),
                incrementingIntegers(c2).subscribeOn(Schedulers.computation()),
                Integer::sum);

        zipped.take(num).subscribe(ts);
        ts.awaitDone(5, TimeUnit.SECONDS);
        ts.assertNoErrors();
        System.out.println("testZipAsync => Received: " + ts.values().size() + "  Emitted: " + c1.get() + " / " + c2.get());
        assertEquals(num, ts.values().size());
        int max = Flowable.bufferSize() * 5;
        assertTrue(c1.get() < max, "" + c1.get() + " >= " + max);
        assertTrue(c2.get() < max, "" + c2.get() + " >= " + max);
    }

    @Test
    public void subscribeOnScheduling() {
        // in a loop for repeating the concurrency in this to increase chance of failure
        for (int i = 0; i < 100; i++) {
            int num = (int) (Flowable.bufferSize() * 2.1);
            AtomicInteger c = new AtomicInteger();
            ConcurrentLinkedQueue<Thread> threads = new ConcurrentLinkedQueue<>();
            TestSubscriber<Integer> ts = new TestSubscriber<>();
            // observeOn is there to make it async and need backpressure
            incrementingIntegers(c, threads).subscribeOn(Schedulers.computation()).observeOn(Schedulers.computation()).take(num).subscribe(ts);
            ts.awaitDone(5, TimeUnit.SECONDS);
            ts.assertNoErrors();
            System.out.println("testSubscribeOnScheduling => Received: " + ts.values().size() + "  Emitted: " + c.get());
            assertEquals(num, ts.values().size());
            assertTrue(c.get() < Flowable.bufferSize() * 4);
            Thread first = null;
            for (Thread t : threads) {
                System.out.println("testSubscribeOnScheduling => thread: " + t);
                if (first == null) {
                    first = t;
                } else {
                    if (!first.equals(t)) {
                        fail("Expected to see the same thread");
                    }
                }
            }
            System.out.println("testSubscribeOnScheduling => Number of batch requests seen: " + threads.size());
            assertTrue(threads.size() > 1);
            System.out.println("-------------------------------------------------------------------------------------------");
        }
    }

    @Test
    public void takeFilterSkipChainAsync() {
        int num = (int) (Flowable.bufferSize() * 2.1);
        AtomicInteger c = new AtomicInteger();
        TestSubscriber<Integer> ts = new TestSubscriber<>();
        incrementingIntegers(c).observeOn(Schedulers.computation())
                .skip(10000)
                .filter(i -> i > 11000).take(num).subscribe(ts);

        ts.awaitDone(5, TimeUnit.SECONDS);
        ts.assertNoErrors();

        // emit 10000 that are skipped
        // emit next 1000 that are filtered out
        // take num
        // so emitted is at least 10000+1000+num + extra for buffer size/threshold
        int expected = 10000 + 1000 + Flowable.bufferSize() * 3 + Flowable.bufferSize() / 2;

        System.out.println("testTakeFilterSkipChain => Received: " + ts.values().size() + "  Emitted: " + c.get() + " Expected: " + expected);
        assertEquals(num, ts.values().size());
        assertTrue(c.get() < expected);
    }

    @Test
    public void userSubscriberUsingRequestSync() {
        AtomicInteger c = new AtomicInteger();
        final AtomicInteger totalReceived = new AtomicInteger();
        final AtomicInteger batches = new AtomicInteger();
        final AtomicInteger received = new AtomicInteger();
        incrementingIntegers(c).subscribe(new ResourceSubscriber<>() /* NFI */ {

            @Override
            public void onStart() {
                request(100);
            }

            @Override
            public void onComplete() {

            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onNext(Integer t) {
                int total = totalReceived.incrementAndGet();
                received.incrementAndGet();
                if (total >= 2000) {
                    dispose();
                }
                if (received.get() == 100) {
                    batches.incrementAndGet();
                    request(100);
                    received.set(0);
                }
            }

        });

        System.out.println("testUserSubscriberUsingRequestSync => Received: " + totalReceived.get() + "  Emitted: " + c.get() + " Request Batches: " + batches.get());
        assertEquals(2000, c.get());
        assertEquals(2000, totalReceived.get());
        assertEquals(20, batches.get());
    }

    @Test
    public void userSubscriberUsingRequestAsync() throws InterruptedException {
        AtomicInteger c = new AtomicInteger();
        final AtomicInteger totalReceived = new AtomicInteger();
        final AtomicInteger received = new AtomicInteger();
        final AtomicInteger batches = new AtomicInteger();
        final CountDownLatch latch = new CountDownLatch(1);
        incrementingIntegers(c).subscribeOn(Schedulers.newThread()).subscribe(
                new ResourceSubscriber<>() /* NFI */ {

                    @Override
                    public void onStart() {
                        request(100);
                    }

                    @Override
                    public void onComplete() {
                        latch.countDown();
                    }

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

                    @Override
                    public void onNext(Integer t) {
                        int total = totalReceived.incrementAndGet();
                        received.incrementAndGet();
                        boolean done = false;
                        if (total >= 2000) {
                            done = true;
                            dispose();
                        }
                        if (received.get() == 100) {
                            batches.incrementAndGet();
                            received.set(0);
                            if (!done) {
                                request(100);
                            }
                        }
                        if (done) {
                            latch.countDown();
                        }
                    }

                });

        latch.await();
        System.out.println("testUserSubscriberUsingRequestAsync => Received: " + totalReceived.get() + "  Emitted: " + c.get() + " Request Batches: " + batches.get());
        assertEquals(2000, c.get());
        assertEquals(2000, totalReceived.get());
        assertEquals(20, batches.get());
    }

    @Test
    public void firehoseFailsAsExpected() {
        AtomicInteger c = new AtomicInteger();
        TestSubscriber<Integer> ts = new TestSubscriber<>();

        firehose(c).observeOn(Schedulers.computation())
        .map(v -> {
            try {
                Thread.sleep(10);
            } catch (Exception e) {
                e.printStackTrace();
            }
            return v;
        })
        .subscribe(ts);

        ts.awaitDone(5, TimeUnit.SECONDS);
        System.out.println("testFirehoseFailsAsExpected => Received: " + ts.values().size() + "  Emitted: " + c.get());

        // FIXME it is possible slow is not slow enough or the main gets delayed and thus more than one source value is emitted.
        int vc = ts.values().size();
        assertTrue(vc <= 10, "10 < " + vc);

        ts.assertError(QueueOverflowException.class);
    }

    @Test
    public void firehoseFailsAsExpectedLoop() {
        for (int i = 0; i < 100; i++) {
            firehoseFailsAsExpected();
        }
    }

    @Test
    public void onBackpressureDrop() {
        long t = System.currentTimeMillis();
        for (int i = 0; i < 100; i++) {
            // stop the test if we are getting close to the timeout because slow machines
            // may not get through 100 iterations
            if (System.currentTimeMillis() - t > TimeUnit.SECONDS.toMillis(9)) {
                break;
            }
            int num = (int) (Flowable.bufferSize() * 1.1); // > 1 so that take doesn't prevent buffer overflow
            AtomicInteger c = new AtomicInteger();
            TestSubscriber<Integer> ts = new TestSubscriber<>();
            firehose(c).onBackpressureDrop()
            .observeOn(Schedulers.computation())
            .map(SLOW_PASS_THRU).take(num).subscribe(ts);
            ts.awaitDone(5, TimeUnit.SECONDS);
            ts.assertNoErrors();

            List<Integer> onNextEvents = ts.values();
            assertEquals(num, onNextEvents.size());

            Integer lastEvent = onNextEvents.get(num - 1);

            System.out.println("testOnBackpressureDrop => Received: " + onNextEvents.size() + "  Emitted: " + c.get() + " Last value: " + lastEvent);
            // it drops, so we should get some number far higher than what would have sequentially incremented
            assertTrue(num - 1 <= lastEvent);
        }
    }

    @Test
    public void onBackpressureDropWithAction(TestInfo testInfo) {
        for (int i = 0; i < 100; i++) {
            final AtomicInteger emitCount = new AtomicInteger();
            final AtomicInteger dropCount = new AtomicInteger();
            final AtomicInteger passCount = new AtomicInteger();
            final int num = Flowable.bufferSize() * 3; // > 1 so that take doesn't prevent buffer overflow
            TestSubscriber<Integer> ts = new TestSubscriber<>();

            firehose(emitCount)
            .onBackpressureDrop(_ -> dropCount.incrementAndGet())
            .doOnNext(_ -> passCount.incrementAndGet())
            .observeOn(Schedulers.computation())
            .map(SLOW_PASS_THRU)
            .take(num).subscribe(ts);

            ts.awaitDone(5, TimeUnit.SECONDS);
            ts.assertNoErrors();

            List<Integer> onNextEvents = ts.values();
            Integer lastEvent = onNextEvents.get(num - 1);
            System.out.println(testInfo.getDisplayName() + " => Received: " + onNextEvents.size() + " Passed: "
            + passCount.get() + " Dropped: " + dropCount.get() + "  Emitted: " + emitCount.get() + " Last value: " + lastEvent);
            assertEquals(num, onNextEvents.size());
            // in reality, num < passCount
            assertTrue(num <= passCount.get());
            // it drops, so we should get some number far higher than what would have sequentially incremented
            assertTrue(num - 1 <= lastEvent);
            assertTrue(0 < dropCount.get());
            assertEquals(emitCount.get(), passCount.get() + dropCount.get());
        }
    }

    @Test
    public void onBackpressureDropSynchronous() {
        for (int i = 0; i < 100; i++) {
            int num = (int) (Flowable.bufferSize() * 1.1); // > 1 so that take doesn't prevent buffer overflow
            AtomicInteger c = new AtomicInteger();
            TestSubscriber<Integer> ts = new TestSubscriber<>();
            firehose(c).onBackpressureDrop()
            .map(SLOW_PASS_THRU).take(num).subscribe(ts);
            ts.awaitDone(5, TimeUnit.SECONDS);
            ts.assertNoErrors();

            List<Integer> onNextEvents = ts.values();
            assertEquals(num, onNextEvents.size());

            Integer lastEvent = onNextEvents.get(num - 1);

            System.out.println("testOnBackpressureDrop => Received: " + onNextEvents.size() + "  Emitted: " + c.get() + " Last value: " + lastEvent);
            // it drops, so we should get some number far higher than what would have sequentially incremented
            assertTrue(num - 1 <= lastEvent);
        }
    }

    @Test
    public void onBackpressureDropSynchronousWithAction() {
        for (int i = 0; i < 100; i++) {
            final AtomicInteger dropCount = new AtomicInteger();
            int num = (int) (Flowable.bufferSize() * 1.1); // > 1 so that take doesn't prevent buffer overflow
            AtomicInteger c = new AtomicInteger();
            TestSubscriber<Integer> ts = new TestSubscriber<>();
            firehose(c).onBackpressureDrop(_ -> dropCount.incrementAndGet())
            .map(SLOW_PASS_THRU).take(num).subscribe(ts);
            ts.awaitDone(5, TimeUnit.SECONDS);
            ts.assertNoErrors();

            List<Integer> onNextEvents = ts.values();
            assertEquals(num, onNextEvents.size());

            Integer lastEvent = onNextEvents.get(num - 1);

            System.out.println("testOnBackpressureDrop => Received: " + onNextEvents.size() + " Dropped: " + dropCount.get()
                + "  Emitted: " + c.get() + " Last value: " + lastEvent);
            // it drops, so we should get some number far higher than what would have sequentially incremented
            assertTrue(num - 1 <= lastEvent);
            // no drop in synchronous mode
            assertEquals(0, dropCount.get());
            assertEquals(c.get(), onNextEvents.size());
        }
    }

    @Test
    public void onBackpressureBuffer() {
        int num = (int) (Flowable.bufferSize() * 1.1); // > 1 so that take doesn't prevent buffer overflow
        AtomicInteger c = new AtomicInteger();
        TestSubscriber<Integer> ts = new TestSubscriber<>();

        firehose(c)
        .takeWhile(t1 -> t1 < 100000)
        .onBackpressureBuffer()
        .observeOn(Schedulers.computation())
        .map(SLOW_PASS_THRU)
        .take(num)
        .subscribe(ts);

        ts.awaitDone(5, TimeUnit.SECONDS);
        ts.assertNoErrors();
        System.out.println("testOnBackpressureBuffer => Received: " + ts.values().size() + "  Emitted: " + c.get());
        assertEquals(num, ts.values().size());
        // it buffers, so we should get the right value sequentially
        assertEquals(num - 1, ts.values().get(num - 1).intValue());
    }

    /**
     * A synchronous Flowable that will emit incrementing integers as requested.
     *
     * @param counter the shared value to be incremented
     * @return the incrementing Flowable instance
     */
    private static Flowable<Integer> incrementingIntegers(final AtomicInteger counter) {
        return incrementingIntegers(counter, null);
    }

    private static Flowable<Integer> incrementingIntegers(final AtomicInteger counter, final ConcurrentLinkedQueue<Thread> threadsSeen) {
        return Flowable.unsafeCreate(s -> s.onSubscribe(new Subscription() /* NFI */ {
            int i;

            volatile boolean cancelled;

            final AtomicLong requested = new AtomicLong();

            @Override
            public void request(long n) {
                if (!SubscriptionHelper.validate(n)) {
                    return;
                }
                if (threadsSeen != null) {
                    threadsSeen.offer(Thread.currentThread());
                }
                long c = BackpressureHelper.add(requested, n);
                if (c == 0) {
                    while (!cancelled) {
                        counter.incrementAndGet();
                        s.onNext(i++);
                        if (requested.decrementAndGet() == 0) {
                            // we're done emitting the number requested so return
                            return;
                        }
                    }
                }
            }

            @Override
            public void cancel() {
                cancelled = true;
            }
        }));
    }

    /**
     * Incrementing int without backpressure.
     *
     * @param counter the shared value to increment
     * @return the Flowable doing the increments
     */
    private static Flowable<Integer> firehose(final AtomicInteger counter) {
        return Flowable.unsafeCreate(s -> {
            Subscription s2 = new FirehoseNoBackpressure(counter, s);
            s.onSubscribe(s2);
        });
    }

    static final Function<Integer, Integer> SLOW_PASS_THRU = new Function<>() /* NFI */ {
        volatile int sink;

        @Override
        public Integer apply(Integer t1) {
            // be slow ... but faster than Thread.sleep(1)
            String t = "";
            int s = sink;
            for (int i = 2000; i >= 0; i--) {
                t = String.valueOf(i + t.hashCode() + s);
            }
            sink = t.hashCode();
            return t1;
        }

    };
}