@Test public void oneIntegerToOneString() { when(source.hasNext()).thenReturn(true); when(source.next()).thenReturn(1); when(mapper.apply(1)).thenReturn(singletonList("1")); final BlockingIterator<String> iterator = flatMap.iterator(); verifyIteratorHasItem(iterator, "1"); }
@Override public BlockingIterator<R> iterator() { final BlockingIterator<T> originalIterator = original.iterator(); return new BlockingIterator<R>() { @Nullable private Iterator<R> intermediate; @Nullable private Object prefetched; @Override public boolean hasNext(final long timeout, final TimeUnit unit) throws TimeoutException { if (prefetched != null) { return true; } long remainingTimeoutNanos = unit.toNanos(timeout); long timeStampANanos = nanoTime(); for (;;) { if (tryFetchFromIntermediate()) { return true; } if (!originalIterator.hasNext(remainingTimeoutNanos, NANOSECONDS)) { return false; } final long timeStampBNanos = nanoTime(); remainingTimeoutNanos -= timeStampBNanos - timeStampANanos; intermediate = mapper.apply(originalIterator.next(remainingTimeoutNanos, NANOSECONDS)).iterator(); timeStampANanos = nanoTime(); remainingTimeoutNanos -= timeStampANanos - timeStampBNanos; } } @Nullable @Override public R next(final long timeout, final TimeUnit unit) throws TimeoutException { if (!hasNext(timeout, unit)) { throw new NoSuchElementException(); } return resetPrefetchAndReturn(); } @Override public void close() throws Exception { originalIterator.close(); } @Override public boolean hasNext() { if (prefetched != null) { return true; } for (;;) { if (tryFetchFromIntermediate()) { return true; } if (!originalIterator.hasNext()) { return false; } intermediate = mapper.apply(originalIterator.next()).iterator(); } } @Nullable @Override public R next() { if (!hasNext()) { throw new NoSuchElementException(); } return resetPrefetchAndReturn(); } private boolean tryFetchFromIntermediate() { if (intermediate != null) { if (intermediate.hasNext()) { prefetched = intermediate.next(); if (prefetched == null) { prefetched = NULL_PLACEHOLDER; } return true; } intermediate = null; } return false; } @Nullable private R resetPrefetchAndReturn() { assert prefetched != null; Object next = prefetched; prefetched = null; return unwrapNullPlaceHolder(next); } @Nullable @SuppressWarnings("unchecked") private R unwrapNullPlaceHolder(final Object next) { return next == NULL_PLACEHOLDER ? null : (R) next; } }; }
BlockingIterableFlatMap implements BlockingIterable<R> { @Override public BlockingIterator<R> iterator() { final BlockingIterator<T> originalIterator = original.iterator(); return new BlockingIterator<R>() { @Nullable private Iterator<R> intermediate; @Nullable private Object prefetched; @Override public boolean hasNext(final long timeout, final TimeUnit unit) throws TimeoutException { if (prefetched != null) { return true; } long remainingTimeoutNanos = unit.toNanos(timeout); long timeStampANanos = nanoTime(); for (;;) { if (tryFetchFromIntermediate()) { return true; } if (!originalIterator.hasNext(remainingTimeoutNanos, NANOSECONDS)) { return false; } final long timeStampBNanos = nanoTime(); remainingTimeoutNanos -= timeStampBNanos - timeStampANanos; intermediate = mapper.apply(originalIterator.next(remainingTimeoutNanos, NANOSECONDS)).iterator(); timeStampANanos = nanoTime(); remainingTimeoutNanos -= timeStampANanos - timeStampBNanos; } } @Nullable @Override public R next(final long timeout, final TimeUnit unit) throws TimeoutException { if (!hasNext(timeout, unit)) { throw new NoSuchElementException(); } return resetPrefetchAndReturn(); } @Override public void close() throws Exception { originalIterator.close(); } @Override public boolean hasNext() { if (prefetched != null) { return true; } for (;;) { if (tryFetchFromIntermediate()) { return true; } if (!originalIterator.hasNext()) { return false; } intermediate = mapper.apply(originalIterator.next()).iterator(); } } @Nullable @Override public R next() { if (!hasNext()) { throw new NoSuchElementException(); } return resetPrefetchAndReturn(); } private boolean tryFetchFromIntermediate() { if (intermediate != null) { if (intermediate.hasNext()) { prefetched = intermediate.next(); if (prefetched == null) { prefetched = NULL_PLACEHOLDER; } return true; } intermediate = null; } return false; } @Nullable private R resetPrefetchAndReturn() { assert prefetched != null; Object next = prefetched; prefetched = null; return unwrapNullPlaceHolder(next); } @Nullable @SuppressWarnings("unchecked") private R unwrapNullPlaceHolder(final Object next) { return next == NULL_PLACEHOLDER ? null : (R) next; } }; } }
BlockingIterableFlatMap implements BlockingIterable<R> { @Override public BlockingIterator<R> iterator() { final BlockingIterator<T> originalIterator = original.iterator(); return new BlockingIterator<R>() { @Nullable private Iterator<R> intermediate; @Nullable private Object prefetched; @Override public boolean hasNext(final long timeout, final TimeUnit unit) throws TimeoutException { if (prefetched != null) { return true; } long remainingTimeoutNanos = unit.toNanos(timeout); long timeStampANanos = nanoTime(); for (;;) { if (tryFetchFromIntermediate()) { return true; } if (!originalIterator.hasNext(remainingTimeoutNanos, NANOSECONDS)) { return false; } final long timeStampBNanos = nanoTime(); remainingTimeoutNanos -= timeStampBNanos - timeStampANanos; intermediate = mapper.apply(originalIterator.next(remainingTimeoutNanos, NANOSECONDS)).iterator(); timeStampANanos = nanoTime(); remainingTimeoutNanos -= timeStampANanos - timeStampBNanos; } } @Nullable @Override public R next(final long timeout, final TimeUnit unit) throws TimeoutException { if (!hasNext(timeout, unit)) { throw new NoSuchElementException(); } return resetPrefetchAndReturn(); } @Override public void close() throws Exception { originalIterator.close(); } @Override public boolean hasNext() { if (prefetched != null) { return true; } for (;;) { if (tryFetchFromIntermediate()) { return true; } if (!originalIterator.hasNext()) { return false; } intermediate = mapper.apply(originalIterator.next()).iterator(); } } @Nullable @Override public R next() { if (!hasNext()) { throw new NoSuchElementException(); } return resetPrefetchAndReturn(); } private boolean tryFetchFromIntermediate() { if (intermediate != null) { if (intermediate.hasNext()) { prefetched = intermediate.next(); if (prefetched == null) { prefetched = NULL_PLACEHOLDER; } return true; } intermediate = null; } return false; } @Nullable private R resetPrefetchAndReturn() { assert prefetched != null; Object next = prefetched; prefetched = null; return unwrapNullPlaceHolder(next); } @Nullable @SuppressWarnings("unchecked") private R unwrapNullPlaceHolder(final Object next) { return next == NULL_PLACEHOLDER ? null : (R) next; } }; } BlockingIterableFlatMap(final BlockingIterable<T> original, final Function<T, Iterable<R>> mapper); }
BlockingIterableFlatMap implements BlockingIterable<R> { @Override public BlockingIterator<R> iterator() { final BlockingIterator<T> originalIterator = original.iterator(); return new BlockingIterator<R>() { @Nullable private Iterator<R> intermediate; @Nullable private Object prefetched; @Override public boolean hasNext(final long timeout, final TimeUnit unit) throws TimeoutException { if (prefetched != null) { return true; } long remainingTimeoutNanos = unit.toNanos(timeout); long timeStampANanos = nanoTime(); for (;;) { if (tryFetchFromIntermediate()) { return true; } if (!originalIterator.hasNext(remainingTimeoutNanos, NANOSECONDS)) { return false; } final long timeStampBNanos = nanoTime(); remainingTimeoutNanos -= timeStampBNanos - timeStampANanos; intermediate = mapper.apply(originalIterator.next(remainingTimeoutNanos, NANOSECONDS)).iterator(); timeStampANanos = nanoTime(); remainingTimeoutNanos -= timeStampANanos - timeStampBNanos; } } @Nullable @Override public R next(final long timeout, final TimeUnit unit) throws TimeoutException { if (!hasNext(timeout, unit)) { throw new NoSuchElementException(); } return resetPrefetchAndReturn(); } @Override public void close() throws Exception { originalIterator.close(); } @Override public boolean hasNext() { if (prefetched != null) { return true; } for (;;) { if (tryFetchFromIntermediate()) { return true; } if (!originalIterator.hasNext()) { return false; } intermediate = mapper.apply(originalIterator.next()).iterator(); } } @Nullable @Override public R next() { if (!hasNext()) { throw new NoSuchElementException(); } return resetPrefetchAndReturn(); } private boolean tryFetchFromIntermediate() { if (intermediate != null) { if (intermediate.hasNext()) { prefetched = intermediate.next(); if (prefetched == null) { prefetched = NULL_PLACEHOLDER; } return true; } intermediate = null; } return false; } @Nullable private R resetPrefetchAndReturn() { assert prefetched != null; Object next = prefetched; prefetched = null; return unwrapNullPlaceHolder(next); } @Nullable @SuppressWarnings("unchecked") private R unwrapNullPlaceHolder(final Object next) { return next == NULL_PLACEHOLDER ? null : (R) next; } }; } BlockingIterableFlatMap(final BlockingIterable<T> original, final Function<T, Iterable<R>> mapper); @Override BlockingIterator<R> iterator(); }
BlockingIterableFlatMap implements BlockingIterable<R> { @Override public BlockingIterator<R> iterator() { final BlockingIterator<T> originalIterator = original.iterator(); return new BlockingIterator<R>() { @Nullable private Iterator<R> intermediate; @Nullable private Object prefetched; @Override public boolean hasNext(final long timeout, final TimeUnit unit) throws TimeoutException { if (prefetched != null) { return true; } long remainingTimeoutNanos = unit.toNanos(timeout); long timeStampANanos = nanoTime(); for (;;) { if (tryFetchFromIntermediate()) { return true; } if (!originalIterator.hasNext(remainingTimeoutNanos, NANOSECONDS)) { return false; } final long timeStampBNanos = nanoTime(); remainingTimeoutNanos -= timeStampBNanos - timeStampANanos; intermediate = mapper.apply(originalIterator.next(remainingTimeoutNanos, NANOSECONDS)).iterator(); timeStampANanos = nanoTime(); remainingTimeoutNanos -= timeStampANanos - timeStampBNanos; } } @Nullable @Override public R next(final long timeout, final TimeUnit unit) throws TimeoutException { if (!hasNext(timeout, unit)) { throw new NoSuchElementException(); } return resetPrefetchAndReturn(); } @Override public void close() throws Exception { originalIterator.close(); } @Override public boolean hasNext() { if (prefetched != null) { return true; } for (;;) { if (tryFetchFromIntermediate()) { return true; } if (!originalIterator.hasNext()) { return false; } intermediate = mapper.apply(originalIterator.next()).iterator(); } } @Nullable @Override public R next() { if (!hasNext()) { throw new NoSuchElementException(); } return resetPrefetchAndReturn(); } private boolean tryFetchFromIntermediate() { if (intermediate != null) { if (intermediate.hasNext()) { prefetched = intermediate.next(); if (prefetched == null) { prefetched = NULL_PLACEHOLDER; } return true; } intermediate = null; } return false; } @Nullable private R resetPrefetchAndReturn() { assert prefetched != null; Object next = prefetched; prefetched = null; return unwrapNullPlaceHolder(next); } @Nullable @SuppressWarnings("unchecked") private R unwrapNullPlaceHolder(final Object next) { return next == NULL_PLACEHOLDER ? null : (R) next; } }; } BlockingIterableFlatMap(final BlockingIterable<T> original, final Function<T, Iterable<R>> mapper); @Override BlockingIterator<R> iterator();  }