OperatorPublish extends ConnectableObservable<T> { @Override public void connect(Action1<? super Subscription> connection) { boolean shouldSubscribe = false; OriginSubscriber<T> origin = requestHandler.state.getOrigin(); if (origin == null) { shouldSubscribe = true; requestHandler.state.setOrigin(new OriginSubscriber<T>(requestHandler)); } if (shouldSubscribe) { connection.call(Subscriptions.create(new Action0() { @Override public void call() { OriginSubscriber<T> s = requestHandler.state.getOrigin(); requestHandler.state.setOrigin(null); if (s != null) { s.unsubscribe(); } } })); OriginSubscriber<T> os = requestHandler.state.getOrigin(); if (os != null) { source.unsafeSubscribe(os); } } } private  OperatorPublish(Observable<? extends T> source); private  OperatorPublish(Observable<? extends T> source, final Object guard, final RequestHandler<T> requestHandler); static ConnectableObservable<T> create(Observable<? extends T> source); static Observable<R> create(final Observable<? extends T> source, final Func1<? super Observable<T>, ? extends Observable<R>> selector); @Override void connect(Action1<? super Subscription> connection);  }
@Test public void testBackpressureFastSlow() { ConnectableObservable<Integer> is = Observable.range(1, RxRingBuffer.SIZE * 2).publish(); Observable<Integer> fast = is.observeOn(Schedulers.computation()).doOnCompleted(new Action0() { @Override public void call() { System.out.println("^^^^^^^^^^^^^ completed FAST"); } }); Observable<Integer> slow = is.observeOn(Schedulers.computation()).map(new Func1<Integer, Integer>() { int c = 0; @Override public Integer call(Integer i) { if (c == 0) { try { Thread.sleep(500); } catch (InterruptedException e) { } } c++; return i; } }).doOnCompleted(new Action0() { @Override public void call() { System.out.println("^^^^^^^^^^^^^ completed SLOW"); } }); TestSubscriber<Integer> ts = new TestSubscriber<Integer>(); Observable.merge(fast, slow).subscribe(ts); is.connect(); ts.awaitTerminalEvent(); ts.assertNoErrors(); assertEquals(RxRingBuffer.SIZE * 4, ts.getOnNextEvents().size()); }
@Test public void testTakeUntilWithPublishedStream() { Observable<Integer> xs = Observable.range(0, RxRingBuffer.SIZE * 2); TestSubscriber<Integer> ts = new TestSubscriber<Integer>(); ConnectableObservable<Integer> xsp = xs.publish(); xsp.takeUntil(xsp.skipWhile(new Func1<Integer, Boolean>() { @Override public Boolean call(Integer i) { return i <= 3; } })).subscribe(ts); xsp.connect(); System.out.println(ts.getOnNextEvents()); }
@Test public void testConnectWithNoSubscriber() { TestScheduler scheduler = new TestScheduler(); ConnectableObservable<Long> co = Observable.timer(10, 10, TimeUnit.MILLISECONDS, scheduler).take(3).publish(); co.connect(); scheduler.advanceTimeBy(15, TimeUnit.MILLISECONDS); TestSubscriber subscriber = new TestSubscriber<Long>(); co.subscribe(subscriber); scheduler.advanceTimeBy(50, TimeUnit.MILLISECONDS); subscriber.assertReceivedOnNext(Arrays.asList(1L, 2L)); subscriber.assertNoErrors(); subscriber.assertTerminalEvent(); }