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