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