@Test public void testSingleSourceManyIterators() throws InterruptedException { Observable<Long> o = Observable.interval(100, TimeUnit.MILLISECONDS); PublishSubject<Void> terminal = PublishSubject.create(); BlockingObservable<Long> source = o.takeUntil(terminal).toBlocking(); Iterable<Long> iter = source.next(); for (int j = 0; j < 3; j++) { BlockingOperatorNext.NextIterator<Long> it = (BlockingOperatorNext.NextIterator<Long>)iter.iterator(); for (long i = 0; i < 10; i++) { Assert.assertEquals(true, it.hasNext()); Assert.assertEquals(j + "th iteration next", Long.valueOf(i), it.next()); } terminal.onNext(null); } }
public static <T> Iterable<T> next(final Observable<? extends T> items) { return new Iterable<T>() { @Override public Iterator<T> iterator() { NextObserver<T> nextObserver = new NextObserver<T>(); return new NextIterator<T>(items, nextObserver); } }; }
BlockingOperatorNext { public static <T> Iterable<T> next(final Observable<? extends T> items) { return new Iterable<T>() { @Override public Iterator<T> iterator() { NextObserver<T> nextObserver = new NextObserver<T>(); return new NextIterator<T>(items, nextObserver); } }; } }
BlockingOperatorNext { public static <T> Iterable<T> next(final Observable<? extends T> items) { return new Iterable<T>() { @Override public Iterator<T> iterator() { NextObserver<T> nextObserver = new NextObserver<T>(); return new NextIterator<T>(items, nextObserver); } }; } private  BlockingOperatorNext(); }
BlockingOperatorNext { public static <T> Iterable<T> next(final Observable<? extends T> items) { return new Iterable<T>() { @Override public Iterator<T> iterator() { NextObserver<T> nextObserver = new NextObserver<T>(); return new NextIterator<T>(items, nextObserver); } }; } private  BlockingOperatorNext(); static Iterable<T> next(final Observable<? extends T> items); }
BlockingOperatorNext { public static <T> Iterable<T> next(final Observable<? extends T> items) { return new Iterable<T>() { @Override public Iterator<T> iterator() { NextObserver<T> nextObserver = new NextObserver<T>(); return new NextIterator<T>(items, nextObserver); } }; } private  BlockingOperatorNext(); static Iterable<T> next(final Observable<? extends T> items);  }