FlowablePublish extends ConnectableFlowable<T> implements HasUpstreamPublisher<T> { public static <T> ConnectableFlowable<T> create(Flowable<T> source, final int bufferSize) { final AtomicReference<PublishSubscriber<T>> curr = new AtomicReference<PublishSubscriber<T>>(); Publisher<T> onSubscribe = new FlowablePublisher<T>(curr, bufferSize); return RxJavaPlugins.onAssembly(new FlowablePublish<T>(onSubscribe, source, curr, bufferSize)); } 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 nextCancelRace() { for (int i = 0; i < TestHelper.RACE_DEFAULT_LOOPS; i++) { final PublishProcessor<Integer> pp = PublishProcessor.create(); final ConnectableFlowable<Integer> cf = pp.publish(); final TestSubscriber<Integer> ts = cf.test(); Runnable r1 = new Runnable() { @Override public void run() { pp.onNext(1); } }; Runnable r2 = new Runnable() { @Override public void run() { ts.cancel(); } }; TestHelper.race(r1, r2); } }
@Test public void selectorDisconnectsIndependentSource() { PublishProcessor<Integer> pp = PublishProcessor.create(); pp.publish(new Function<Flowable<Integer>, Flowable<Integer>>() { @Override public Flowable<Integer> apply(Flowable<Integer> v) throws Exception { return Flowable.range(1, 2); } }) .test() .assertResult(1, 2); assertFalse(pp.hasSubscribers()); }
@Test public void selectorInnerError() { PublishProcessor<Integer> pp = PublishProcessor.create(); pp.publish(new Function<Flowable<Integer>, Flowable<Integer>>() { @Override public Flowable<Integer> apply(Flowable<Integer> v) throws Exception { return Flowable.error(new TestException()); } }) .test() .assertFailure(TestException.class); assertFalse(pp.hasSubscribers()); }
@Test public void overflowQueue() { Flowable.create(new FlowableOnSubscribe<Object>() { @Override public void subscribe(FlowableEmitter<Object> s) throws Exception { for (int i = 0; i < 10; i++) { s.onNext(i); } } }, BackpressureStrategy.MISSING) .publish(8) .autoConnect() .test(0L) .assertFailure(MissingBackpressureException.class); }
@Test public void leavingSubscriberOverrequests() { final AtomicReference<Flowable<Integer>> ref = new AtomicReference<Flowable<Integer>>(); PublishProcessor<Integer> pp = PublishProcessor.create(); pp.publish(new Function<Flowable<Integer>, Publisher<Integer>>() { @Override public Publisher<Integer> apply(Flowable<Integer> f) throws Exception { ref.set(f); return Flowable.never(); } }).test(); TestSubscriber<Integer> ts1 = ref.get().take(2).test(); pp.onNext(1); pp.onNext(2); ts1.assertResult(1, 2); pp.onNext(3); pp.onNext(4); TestSubscriber<Integer> ts2 = ref.get().test(0L); ts2.assertEmpty(); ts2.requestMore(2); ts2.assertValuesOnly(3, 4); }
@Test public void publishFunctionCancelOuterAfterOneInner() { final AtomicReference<Flowable<Integer>> ref = new AtomicReference<Flowable<Integer>>(); PublishProcessor<Integer> pp = PublishProcessor.create(); final TestSubscriber<Integer> ts = pp.publish(new Function<Flowable<Integer>, Publisher<Integer>>() { @Override public Publisher<Integer> apply(Flowable<Integer> f) throws Exception { ref.set(f); return Flowable.never(); } }).test(); ref.get().subscribe(new TestSubscriber<Integer>() { @Override public void onNext(Integer t) { super.onNext(t); onComplete(); ts.cancel(); } }); pp.onNext(1); }
@Test public void publishFunctionCancelOuterAfterOneInnerBackpressured() { final AtomicReference<Flowable<Integer>> ref = new AtomicReference<Flowable<Integer>>(); PublishProcessor<Integer> pp = PublishProcessor.create(); final TestSubscriber<Integer> ts = pp.publish(new Function<Flowable<Integer>, Publisher<Integer>>() { @Override public Publisher<Integer> apply(Flowable<Integer> f) throws Exception { ref.set(f); return Flowable.never(); } }).test(); ref.get().subscribe(new TestSubscriber<Integer>(1L) { @Override public void onNext(Integer t) { super.onNext(t); onComplete(); ts.cancel(); } }); pp.onNext(1); }
@Test public void publishCancelOneAsync() { for (int i = 0; i < TestHelper.RACE_LONG_LOOPS; i++) { final PublishProcessor<Integer> pp = PublishProcessor.create(); final AtomicReference<Flowable<Integer>> ref = new AtomicReference<Flowable<Integer>>(); pp.publish(new Function<Flowable<Integer>, Publisher<Integer>>() { @Override public Publisher<Integer> apply(Flowable<Integer> f) throws Exception { ref.set(f); return Flowable.never(); } }).test(); final TestSubscriber<Integer> ts1 = ref.get().test(); TestSubscriber<Integer> ts2 = ref.get().test(); Runnable r1 = new Runnable() { @Override public void run() { pp.onNext(1); } }; Runnable r2 = new Runnable() { @Override public void run() { ts1.cancel(); } }; TestHelper.race(r1, r2); ts2.assertValuesOnly(1); } }