ObservablePublish extends ConnectableObservable<T> implements HasUpstreamObservableSource<T> { @Override public void connect(Consumer<? super Disposable> connection) { boolean doConnect; PublishObserver<T> ps; for (;;) { ps = current.get(); if (ps == null || ps.isDisposed()) { PublishObserver<T> u = new PublishObserver<T>(current); 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  ObservablePublish(ObservableSource<T> onSubscribe, ObservableSource<T> source,
                              final AtomicReference<PublishObserver<T>> current); static ConnectableObservable<T> create(ObservableSource<T> source); @Override ObservableSource<T> source(); @Override void connect(Consumer<? super Disposable> connection);  }
@Test public void testBackpressureFastSlow() { ConnectableObservable<Integer> is = Observable.range(1, Flowable.bufferSize() * 2).publish(); Observable<Integer> fast = is.observeOn(Schedulers.computation()) .doOnComplete(new Action() { @Override public void run() { System.out.println("^^^^^^^^^^^^^ completed FAST"); } }); Observable<Integer> slow = is.observeOn(Schedulers.computation()).map(new Function<Integer, Integer>() { int c; @Override public Integer apply(Integer i) { if (c == 0) { try { Thread.sleep(500); } catch (InterruptedException e) { } } c++; return i; } }).doOnComplete(new Action() { @Override public void run() { System.out.println("^^^^^^^^^^^^^ completed SLOW"); } }); TestObserver<Integer> to = new TestObserver<Integer>(); Observable.merge(fast, slow).subscribe(to); is.connect(); to.awaitTerminalEvent(); to.assertNoErrors(); assertEquals(Flowable.bufferSize() * 4, to.valueCount()); }
@Test public void testTakeUntilWithPublishedStream() { Observable<Integer> xs = Observable.range(0, Flowable.bufferSize() * 2); TestObserver<Integer> to = new TestObserver<Integer>(); ConnectableObservable<Integer> xsp = xs.publish(); xsp.takeUntil(xsp.skipWhile(new Predicate<Integer>() { @Override public boolean test(Integer i) { return i <= 3; } })).subscribe(to); xsp.connect(); System.out.println(to.values()); }
@Test public void testSubscribeAfterDisconnectThenConnect() { ConnectableObservable<Integer> source = Observable.just(1).publish(); TestObserver<Integer> to1 = new TestObserver<Integer>(); source.subscribe(to1); Disposable s = source.connect(); to1.assertValue(1); to1.assertNoErrors(); to1.assertTerminated(); TestObserver<Integer> to2 = new TestObserver<Integer>(); source.subscribe(to2); Disposable s2 = source.connect(); to2.assertValue(1); to2.assertNoErrors(); to2.assertTerminated(); System.out.println(s); System.out.println(s2); }
@Test public void testNoSubscriberRetentionOnCompleted() { ObservablePublish<Integer> source = (ObservablePublish<Integer>)Observable.just(1).publish(); TestObserver<Integer> to1 = new TestObserver<Integer>(); source.subscribe(to1); to1.assertNoValues(); to1.assertNoErrors(); to1.assertNotComplete(); source.connect(); to1.assertValue(1); to1.assertNoErrors(); to1.assertTerminated(); assertNull(source.current.get()); }
@Test public void testObserveOn() { ConnectableObservable<Integer> co = Observable.range(0, 1000).publish(); Observable<Integer> obs = co.observeOn(Schedulers.computation()); for (int i = 0; i < 1000; i++) { for (int j = 1; j < 6; j++) { List<TestObserver<Integer>> tos = new ArrayList<TestObserver<Integer>>(); for (int k = 1; k < j; k++) { TestObserver<Integer> to = new TestObserver<Integer>(); tos.add(to); obs.subscribe(to); } Disposable s = co.connect(); for (TestObserver<Integer> to : tos) { to.awaitTerminalEvent(2, TimeUnit.SECONDS); to.assertTerminated(); to.assertNoErrors(); assertEquals(1000, to.valueCount()); } s.dispose(); } } }
@Test public void preNextConnect() { for (int i = 0; i < TestHelper.RACE_DEFAULT_LOOPS; i++) { final ConnectableObservable<Integer> co = Observable.<Integer>empty().publish(); co.connect(); Runnable r1 = new Runnable() { @Override public void run() { co.test(); } }; TestHelper.race(r1, r1); } }
@Test public void connectRace() { for (int i = 0; i < TestHelper.RACE_DEFAULT_LOOPS; i++) { final ConnectableObservable<Integer> co = Observable.<Integer>empty().publish(); Runnable r1 = new Runnable() { @Override public void run() { co.connect(); } }; TestHelper.race(r1, r1); } }
@Test public void noErrorLoss() { List<Throwable> errors = TestHelper.trackPluginErrors(); try { ConnectableObservable<Object> co = Observable.error(new TestException()).publish(); co.connect(); TestHelper.assertUndeliverable(errors, 0, TestException.class); } finally { RxJavaPlugins.reset(); } }
@Test public void disposedUpfront() { ConnectableObservable<Integer> co = Observable.just(1) .concatWith(Observable.<Integer>never()) .publish(); TestObserver<Integer> to1 = co.test(); TestObserver<Integer> to2 = co.test(true); co.connect(); to1.assertValuesOnly(1); to2.assertEmpty(); ((ObservablePublish<Integer>)co).current.get().remove(null); }