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