FlowablePublish extends ConnectableFlowable<T> implements HasUpstreamPublisher<T> { @Override public void connect(Consumer<? super Disposable> connection) { boolean doConnect; PublishSubscriber<T> ps; for (;;) { ps = current.get(); if (ps == null || ps.isDisposed()) { PublishSubscriber<T> u = new PublishSubscriber<T>(current, bufferSize); 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) { Exceptions.throwIfFatal(ex); throw ExceptionHelper.wrapOrThrow(ex); } if (doConnect) { source.subscribe(ps); } } private  FlowablePublish(Publisher<T> onSubscribe, Flowable<T> source,
            final AtomicReference<PublishSubscriber<T>> current, int bufferSize); static ConnectableFlowable<T> create(Flowable<T> source, final int bufferSize); @Override Publisher<T> source(); @Override void connect(Consumer<? super Disposable> connection);  }
@Test public void testConnectWithNoSubscriber() { TestScheduler scheduler = new TestScheduler(); ConnectableFlowable<Long> cf = Flowable.interval(10, 10, TimeUnit.MILLISECONDS, scheduler).take(3).publish(); cf.connect(); scheduler.advanceTimeBy(15, TimeUnit.MILLISECONDS); TestSubscriber<Long> subscriber = new TestSubscriber<Long>(); cf.subscribe(subscriber); scheduler.advanceTimeBy(50, TimeUnit.MILLISECONDS); subscriber.assertValues(1L, 2L); subscriber.assertNoErrors(); subscriber.assertTerminated(); }
@Test public void syncFusedObserveOn() { ConnectableFlowable<Integer> cf = Flowable.range(0, 1000).publish(); Flowable<Integer> obs = cf.observeOn(Schedulers.computation()); for (int i = 0; i < 1000; i++) { for (int j = 1; j < 6; j++) { List<TestSubscriber<Integer>> tss = new ArrayList<TestSubscriber<Integer>>(); for (int k = 1; k < j; k++) { TestSubscriber<Integer> ts = new TestSubscriber<Integer>(); tss.add(ts); obs.subscribe(ts); } Disposable s = cf.connect(); for (TestSubscriber<Integer> ts : tss) { ts.awaitDone(5, TimeUnit.SECONDS) .assertSubscribed() .assertValueCount(1000) .assertNoErrors() .assertComplete(); } s.dispose(); } } }
@Test public void syncFusedObserveOn2() { ConnectableFlowable<Integer> cf = Flowable.range(0, 1000).publish(); Flowable<Integer> obs = cf.observeOn(ImmediateThinScheduler.INSTANCE); for (int i = 0; i < 1000; i++) { for (int j = 1; j < 6; j++) { List<TestSubscriber<Integer>> tss = new ArrayList<TestSubscriber<Integer>>(); for (int k = 1; k < j; k++) { TestSubscriber<Integer> ts = new TestSubscriber<Integer>(); tss.add(ts); obs.subscribe(ts); } Disposable s = cf.connect(); for (TestSubscriber<Integer> ts : tss) { ts.awaitDone(5, TimeUnit.SECONDS) .assertSubscribed() .assertValueCount(1000) .assertNoErrors() .assertComplete(); } s.dispose(); } } }
@Test public void asyncFusedObserveOn() { ConnectableFlowable<Integer> cf = Flowable.range(0, 1000).observeOn(ImmediateThinScheduler.INSTANCE).publish(); for (int i = 0; i < 1000; i++) { for (int j = 1; j < 6; j++) { List<TestSubscriber<Integer>> tss = new ArrayList<TestSubscriber<Integer>>(); for (int k = 1; k < j; k++) { TestSubscriber<Integer> ts = new TestSubscriber<Integer>(); tss.add(ts); cf.subscribe(ts); } Disposable s = cf.connect(); for (TestSubscriber<Integer> ts : tss) { ts.awaitDone(5, TimeUnit.SECONDS) .assertSubscribed() .assertValueCount(1000) .assertNoErrors() .assertComplete(); } s.dispose(); } } }
@Test public void testObserveOn() { ConnectableFlowable<Integer> cf = Flowable.range(0, 1000).hide().publish(); Flowable<Integer> obs = cf.observeOn(Schedulers.computation()); for (int i = 0; i < 1000; i++) { for (int j = 1; j < 6; j++) { List<TestSubscriber<Integer>> tss = new ArrayList<TestSubscriber<Integer>>(); for (int k = 1; k < j; k++) { TestSubscriber<Integer> ts = new TestSubscriber<Integer>(); tss.add(ts); obs.subscribe(ts); } Disposable s = cf.connect(); for (TestSubscriber<Integer> ts : tss) { ts.awaitDone(5, TimeUnit.SECONDS) .assertSubscribed() .assertValueCount(1000) .assertNoErrors() .assertComplete(); } s.dispose(); } } }
@Test public void connectThrows() { ConnectableFlowable<Integer> cf = Flowable.<Integer>empty().publish(); try { cf.connect(new Consumer<Disposable>() { @Override public void accept(Disposable s) throws Exception { throw new TestException(); } }); } catch (TestException ex) { } }
@Test public void empty() { ConnectableFlowable<Integer> cf = Flowable.<Integer>empty().publish(); cf.connect(); }
@Test public void take() { ConnectableFlowable<Integer> cf = Flowable.range(1, 2).publish(); TestSubscriber<Integer> ts = cf.take(1).test(); cf.connect(); ts.assertResult(1); }
@Test public void noErrorLoss() { List<Throwable> errors = TestHelper.trackPluginErrors(); try { ConnectableFlowable<Object> cf = Flowable.error(new TestException()).publish(); cf.connect(); TestHelper.assertUndeliverable(errors, 0, TestException.class); } finally { RxJavaPlugins.reset(); } }
@Test public void preNextConnect() { for (int i = 0; i < TestHelper.RACE_DEFAULT_LOOPS; i++) { final ConnectableFlowable<Integer> cf = Flowable.<Integer>empty().publish(); cf.connect(); Runnable r1 = new Runnable() { @Override public void run() { cf.test(); } }; TestHelper.race(r1, r1); } }
@Test public void connectRace() { for (int i = 0; i < TestHelper.RACE_DEFAULT_LOOPS; i++) { final ConnectableFlowable<Integer> cf = Flowable.<Integer>empty().publish(); Runnable r1 = new Runnable() { @Override public void run() { cf.connect(); } }; TestHelper.race(r1, r1); } }
@Test public void disposeRace() { for (int i = 0; i < TestHelper.RACE_DEFAULT_LOOPS; i++) { final AtomicReference<Disposable> ref = new AtomicReference<Disposable>(); final ConnectableFlowable<Integer> cf = new Flowable<Integer>() { @Override protected void subscribeActual(Subscriber<? super Integer> s) { s.onSubscribe(new BooleanSubscription()); ref.set((Disposable)s); } }.publish(); cf.connect(); Runnable r1 = new Runnable() { @Override public void run() { ref.get().dispose(); } }; TestHelper.race(r1, r1); } }
@Test public void removeNotPresent() { final AtomicReference<PublishSubscriber<Integer>> ref = new AtomicReference<PublishSubscriber<Integer>>(); final ConnectableFlowable<Integer> cf = new Flowable<Integer>() { @Override @SuppressWarnings("unchecked") protected void subscribeActual(Subscriber<? super Integer> s) { s.onSubscribe(new BooleanSubscription()); ref.set((PublishSubscriber<Integer>)s); } }.publish(); cf.connect(); ref.get().add(new InnerSubscriber<Integer>(new TestSubscriber<Integer>())); ref.get().remove(null); }
@Test @Ignore("publish() keeps consuming the upstream if there are no subscribers, 3.x should change this") public void subscriberSwap() { final ConnectableFlowable<Integer> cf = Flowable.range(1, 5).publish(); cf.connect(); TestSubscriber<Integer> ts1 = new TestSubscriber<Integer>() { @Override public void onNext(Integer t) { super.onNext(t); cancel(); onComplete(); } }; cf.subscribe(ts1); ts1.assertResult(1); TestSubscriber<Integer> ts2 = new TestSubscriber<Integer>(0); cf.subscribe(ts2); ts2 .assertEmpty() .requestMore(4) .assertResult(2, 3, 4, 5); }
@Test public void subscriberLiveSwap() { final ConnectableFlowable<Integer> cf = Flowable.range(1, 5).publish(); final TestSubscriber<Integer> ts2 = new TestSubscriber<Integer>(0); TestSubscriber<Integer> ts1 = new TestSubscriber<Integer>() { @Override public void onNext(Integer t) { super.onNext(t); cancel(); onComplete(); cf.subscribe(ts2); } }; cf.subscribe(ts1); cf.connect(); ts1.assertResult(1); ts2 .assertEmpty() .requestMore(4) .assertResult(2, 3, 4, 5); }