@Test(expected = TestException.class) public void testToIteratorWithException() { Observable<String> obs = Observable.create(new OnSubscribeFunc<String>() { @Override public Subscription onSubscribe(Observer<? super String> observer) { observer.onNext("one"); observer.onError(new TestException()); return Subscriptions.empty(); } }); Iterator<String> it = toIterator(obs); assertEquals(true, it.hasNext()); assertEquals("one", it.next()); assertEquals(true, it.hasNext()); it.next(); }
public static <T> Iterator<T> toIterator(Observable<? extends T> source) { final BlockingQueue<Notification<? extends T>> notifications = new LinkedBlockingQueue<Notification<? extends T>>(); source.materialize().subscribe(new Observer<Notification<? extends T>>() { @Override public void onCompleted() { } @Override public void onError(Throwable e) { } @Override public void onNext(Notification<? extends T> args) { notifications.offer(args); } }); return new Iterator<T>() { private Notification<? extends T> buf; @Override public boolean hasNext() { if (buf == null) { buf = take(); } if (buf.isOnError()) { throw Exceptions.propagate(buf.getThrowable()); } return !buf.isOnCompleted(); } @Override public T next() { if (hasNext()) { T result = buf.getValue(); buf = null; return result; } throw new NoSuchElementException(); } private Notification<? extends T> take() { try { return notifications.take(); } catch (InterruptedException e) { throw Exceptions.propagate(e); } } @Override public void remove() { throw new UnsupportedOperationException("Read-only iterator"); } }; }
OperationToIterator { public static <T> Iterator<T> toIterator(Observable<? extends T> source) { final BlockingQueue<Notification<? extends T>> notifications = new LinkedBlockingQueue<Notification<? extends T>>(); source.materialize().subscribe(new Observer<Notification<? extends T>>() { @Override public void onCompleted() { } @Override public void onError(Throwable e) { } @Override public void onNext(Notification<? extends T> args) { notifications.offer(args); } }); return new Iterator<T>() { private Notification<? extends T> buf; @Override public boolean hasNext() { if (buf == null) { buf = take(); } if (buf.isOnError()) { throw Exceptions.propagate(buf.getThrowable()); } return !buf.isOnCompleted(); } @Override public T next() { if (hasNext()) { T result = buf.getValue(); buf = null; return result; } throw new NoSuchElementException(); } private Notification<? extends T> take() { try { return notifications.take(); } catch (InterruptedException e) { throw Exceptions.propagate(e); } } @Override public void remove() { throw new UnsupportedOperationException("Read-only iterator"); } }; } }
OperationToIterator { public static <T> Iterator<T> toIterator(Observable<? extends T> source) { final BlockingQueue<Notification<? extends T>> notifications = new LinkedBlockingQueue<Notification<? extends T>>(); source.materialize().subscribe(new Observer<Notification<? extends T>>() { @Override public void onCompleted() { } @Override public void onError(Throwable e) { } @Override public void onNext(Notification<? extends T> args) { notifications.offer(args); } }); return new Iterator<T>() { private Notification<? extends T> buf; @Override public boolean hasNext() { if (buf == null) { buf = take(); } if (buf.isOnError()) { throw Exceptions.propagate(buf.getThrowable()); } return !buf.isOnCompleted(); } @Override public T next() { if (hasNext()) { T result = buf.getValue(); buf = null; return result; } throw new NoSuchElementException(); } private Notification<? extends T> take() { try { return notifications.take(); } catch (InterruptedException e) { throw Exceptions.propagate(e); } } @Override public void remove() { throw new UnsupportedOperationException("Read-only iterator"); } }; }  }
OperationToIterator { public static <T> Iterator<T> toIterator(Observable<? extends T> source) { final BlockingQueue<Notification<? extends T>> notifications = new LinkedBlockingQueue<Notification<? extends T>>(); source.materialize().subscribe(new Observer<Notification<? extends T>>() { @Override public void onCompleted() { } @Override public void onError(Throwable e) { } @Override public void onNext(Notification<? extends T> args) { notifications.offer(args); } }); return new Iterator<T>() { private Notification<? extends T> buf; @Override public boolean hasNext() { if (buf == null) { buf = take(); } if (buf.isOnError()) { throw Exceptions.propagate(buf.getThrowable()); } return !buf.isOnCompleted(); } @Override public T next() { if (hasNext()) { T result = buf.getValue(); buf = null; return result; } throw new NoSuchElementException(); } private Notification<? extends T> take() { try { return notifications.take(); } catch (InterruptedException e) { throw Exceptions.propagate(e); } } @Override public void remove() { throw new UnsupportedOperationException("Read-only iterator"); } }; }  static Iterator<T> toIterator(Observable<? extends T> source); }
OperationToIterator { public static <T> Iterator<T> toIterator(Observable<? extends T> source) { final BlockingQueue<Notification<? extends T>> notifications = new LinkedBlockingQueue<Notification<? extends T>>(); source.materialize().subscribe(new Observer<Notification<? extends T>>() { @Override public void onCompleted() { } @Override public void onError(Throwable e) { } @Override public void onNext(Notification<? extends T> args) { notifications.offer(args); } }); return new Iterator<T>() { private Notification<? extends T> buf; @Override public boolean hasNext() { if (buf == null) { buf = take(); } if (buf.isOnError()) { throw Exceptions.propagate(buf.getThrowable()); } return !buf.isOnCompleted(); } @Override public T next() { if (hasNext()) { T result = buf.getValue(); buf = null; return result; } throw new NoSuchElementException(); } private Notification<? extends T> take() { try { return notifications.take(); } catch (InterruptedException e) { throw Exceptions.propagate(e); } } @Override public void remove() { throw new UnsupportedOperationException("Read-only iterator"); } }; }  static Iterator<T> toIterator(Observable<? extends T> source);  }