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