@Test public void testBackpressureBounded() { final AtomicLong requested = new AtomicLong(); Flowable<Integer> source = Flowable.range(1, 1000) .doOnRequest(new LongConsumer() { @Override public void accept(long t) { requested.addAndGet(t); } }); ConnectableFlowable<Integer> cf = source.replay(50); TestSubscriber<Integer> ts1 = new TestSubscriber<Integer>(10L); TestSubscriber<Integer> ts2 = new TestSubscriber<Integer>(90L); cf.subscribe(ts1); cf.subscribe(ts2); ts2.request(10); cf.connect(); ts1.assertValueCount(10); ts1.assertNotTerminated(); ts2.assertValueCount(100); ts2.assertNotTerminated(); Assert.assertEquals(100, requested.get()); }
@Override public void connect(Consumer<? super Disposable> connection) { boolean doConnect; ReplaySubscriber<T> ps; for (;;) { ps = current.get(); if (ps == null || ps.isDisposed()) { ReplayBuffer<T> buf; try { buf = bufferFactory.call(); } catch (Throwable ex) { Exceptions.throwIfFatal(ex); throw ExceptionHelper.wrapOrThrow(ex); } ReplaySubscriber<T> u = new ReplaySubscriber<T>(buf); if (!current.compareAndSet(ps, u)) { continue; } ps = u; } doConnect = !ps.shouldConnect.get() && ps.shouldConnect.compareAndSet(false, true); break; } try { connection.accept(ps); } catch (Throwable ex) { if (doConnect) { ps.shouldConnect.compareAndSet(true, false); } Exceptions.throwIfFatal(ex); throw ExceptionHelper.wrapOrThrow(ex); } if (doConnect) { source.subscribe(ps); } }
FlowableReplay extends ConnectableFlowable<T> implements HasUpstreamPublisher<T>, Disposable { @Override public void connect(Consumer<? super Disposable> connection) { boolean doConnect; ReplaySubscriber<T> ps; for (;;) { ps = current.get(); if (ps == null || ps.isDisposed()) { ReplayBuffer<T> buf; try { buf = bufferFactory.call(); } catch (Throwable ex) { Exceptions.throwIfFatal(ex); throw ExceptionHelper.wrapOrThrow(ex); } ReplaySubscriber<T> u = new ReplaySubscriber<T>(buf); if (!current.compareAndSet(ps, u)) { continue; } ps = u; } doConnect = !ps.shouldConnect.get() && ps.shouldConnect.compareAndSet(false, true); break; } try { connection.accept(ps); } catch (Throwable ex) { if (doConnect) { ps.shouldConnect.compareAndSet(true, false); } Exceptions.throwIfFatal(ex); throw ExceptionHelper.wrapOrThrow(ex); } if (doConnect) { source.subscribe(ps); } } }
FlowableReplay extends ConnectableFlowable<T> implements HasUpstreamPublisher<T>, Disposable { @Override public void connect(Consumer<? super Disposable> connection) { boolean doConnect; ReplaySubscriber<T> ps; for (;;) { ps = current.get(); if (ps == null || ps.isDisposed()) { ReplayBuffer<T> buf; try { buf = bufferFactory.call(); } catch (Throwable ex) { Exceptions.throwIfFatal(ex); throw ExceptionHelper.wrapOrThrow(ex); } ReplaySubscriber<T> u = new ReplaySubscriber<T>(buf); if (!current.compareAndSet(ps, u)) { continue; } ps = u; } doConnect = !ps.shouldConnect.get() && ps.shouldConnect.compareAndSet(false, true); break; } try { connection.accept(ps); } catch (Throwable ex) { if (doConnect) { ps.shouldConnect.compareAndSet(true, false); } Exceptions.throwIfFatal(ex); throw ExceptionHelper.wrapOrThrow(ex); } if (doConnect) { source.subscribe(ps); } } private  FlowableReplay(Publisher<T> onSubscribe, Flowable<T> source,
            final AtomicReference<ReplaySubscriber<T>> current,
            final Callable<? extends ReplayBuffer<T>> bufferFactory); }
FlowableReplay extends ConnectableFlowable<T> implements HasUpstreamPublisher<T>, Disposable { @Override public void connect(Consumer<? super Disposable> connection) { boolean doConnect; ReplaySubscriber<T> ps; for (;;) { ps = current.get(); if (ps == null || ps.isDisposed()) { ReplayBuffer<T> buf; try { buf = bufferFactory.call(); } catch (Throwable ex) { Exceptions.throwIfFatal(ex); throw ExceptionHelper.wrapOrThrow(ex); } ReplaySubscriber<T> u = new ReplaySubscriber<T>(buf); if (!current.compareAndSet(ps, u)) { continue; } ps = u; } doConnect = !ps.shouldConnect.get() && ps.shouldConnect.compareAndSet(false, true); break; } try { connection.accept(ps); } catch (Throwable ex) { if (doConnect) { ps.shouldConnect.compareAndSet(true, false); } Exceptions.throwIfFatal(ex); throw ExceptionHelper.wrapOrThrow(ex); } if (doConnect) { source.subscribe(ps); } } private  FlowableReplay(Publisher<T> onSubscribe, Flowable<T> source,
            final AtomicReference<ReplaySubscriber<T>> current,
            final Callable<? extends ReplayBuffer<T>> bufferFactory); static Flowable<R> multicastSelector(
            final Callable<? extends ConnectableFlowable<U>> connectableFactory,
            final Function<? super Flowable<U>, ? extends Publisher<R>> selector); static ConnectableFlowable<T> observeOn(final ConnectableFlowable<T> cf, final Scheduler scheduler); @SuppressWarnings("unchecked") static ConnectableFlowable<T> createFrom(Flowable<? extends T> source); static ConnectableFlowable<T> create(Flowable<T> source,
            final int bufferSize); static ConnectableFlowable<T> create(Flowable<T> source,
            long maxAge, TimeUnit unit, Scheduler scheduler); static ConnectableFlowable<T> create(Flowable<T> source,
            final long maxAge, final TimeUnit unit, final Scheduler scheduler, final int bufferSize); @Override Publisher<T> source(); @Override void dispose(); @Override boolean isDisposed(); @Override void connect(Consumer<? super Disposable> connection); }
FlowableReplay extends ConnectableFlowable<T> implements HasUpstreamPublisher<T>, Disposable { @Override public void connect(Consumer<? super Disposable> connection) { boolean doConnect; ReplaySubscriber<T> ps; for (;;) { ps = current.get(); if (ps == null || ps.isDisposed()) { ReplayBuffer<T> buf; try { buf = bufferFactory.call(); } catch (Throwable ex) { Exceptions.throwIfFatal(ex); throw ExceptionHelper.wrapOrThrow(ex); } ReplaySubscriber<T> u = new ReplaySubscriber<T>(buf); if (!current.compareAndSet(ps, u)) { continue; } ps = u; } doConnect = !ps.shouldConnect.get() && ps.shouldConnect.compareAndSet(false, true); break; } try { connection.accept(ps); } catch (Throwable ex) { if (doConnect) { ps.shouldConnect.compareAndSet(true, false); } Exceptions.throwIfFatal(ex); throw ExceptionHelper.wrapOrThrow(ex); } if (doConnect) { source.subscribe(ps); } } private  FlowableReplay(Publisher<T> onSubscribe, Flowable<T> source,
            final AtomicReference<ReplaySubscriber<T>> current,
            final Callable<? extends ReplayBuffer<T>> bufferFactory); static Flowable<R> multicastSelector(
            final Callable<? extends ConnectableFlowable<U>> connectableFactory,
            final Function<? super Flowable<U>, ? extends Publisher<R>> selector); static ConnectableFlowable<T> observeOn(final ConnectableFlowable<T> cf, final Scheduler scheduler); @SuppressWarnings("unchecked") static ConnectableFlowable<T> createFrom(Flowable<? extends T> source); static ConnectableFlowable<T> create(Flowable<T> source,
            final int bufferSize); static ConnectableFlowable<T> create(Flowable<T> source,
            long maxAge, TimeUnit unit, Scheduler scheduler); static ConnectableFlowable<T> create(Flowable<T> source,
            final long maxAge, final TimeUnit unit, final Scheduler scheduler, final int bufferSize); @Override Publisher<T> source(); @Override void dispose(); @Override boolean isDisposed(); @Override void connect(Consumer<? super Disposable> connection);  }