@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); } }
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)); }
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)); } }
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); }
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); }
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);  }