private static <T> List<T> insertNullEvery(int n, List<T> iterable)
{
    return newArrayList(() -> new AbstractIterator<T>()
    {
        private final Iterator<T> delegate = iterable.iterator();
        private int position;

        @Override
        protected T computeNext()
        {
            position++;
            if (position > n) {
                position = 0;
                return null;
            }

            if (!delegate.hasNext()) {
                return endOfData();
            }

            return delegate.next();
        }
    });
}
 

private static <T> Iterable<T> skipEvery(int n, Iterable<T> iterable)
{
    return () -> new AbstractIterator<T>()
    {
        private final Iterator<T> delegate = iterable.iterator();
        private int position;

        @Override
        protected T computeNext()
        {
            while (true) {
                if (!delegate.hasNext()) {
                    return endOfData();
                }

                T next = delegate.next();
                position++;
                if (position <= n) {
                    return next;
                }
                position = 0;
            }
        }
    };
}
 

private static <T> Iterable<T> repeatEach(int n, Iterable<T> iterable)
{
    return () -> new AbstractIterator<T>()
    {
        private final Iterator<T> delegate = iterable.iterator();
        private int position;
        private T value;

        @Override
        protected T computeNext()
        {
            if (position == 0) {
                if (!delegate.hasNext()) {
                    return endOfData();
                }
                value = delegate.next();
            }

            position++;
            if (position >= n) {
                position = 0;
            }
            return value;
        }
    };
}
 

@Override
public Iterable<T> reverseIterable() {
	return new Iterable<T>() {
		@Override
		public Iterator<T> iterator() {
			return new AbstractIterator<T>() {
				private int index = contents.size() - 1;

				@Override
				protected T computeNext() {
					if (index < 0)
						return endOfData();
					return contents.get(index--);
				}
			};
		}
	};
}
 

private static <T> Iterator<T> closeWhenExhausted(Iterator<T> iterator, Closeable resource)
{
    requireNonNull(iterator, "iterator is null");
    requireNonNull(resource, "resource is null");

    return new AbstractIterator<T>()
    {
        @Override
        protected T computeNext()
        {
            if (iterator.hasNext()) {
                return iterator.next();
            }
            try {
                resource.close();
            }
            catch (IOException e) {
                throw new UncheckedIOException(e);
            }
            return endOfData();
        }
    };
}
 

public Iterator<Page> getPages()
{
    return new AbstractIterator<Page>()
    {
        private int pageCounter;

        @Override
        protected Page computeNext()
        {
            if (pageCounter == channels[0].size()) {
                return endOfData();
            }

            Block[] blocks = Stream.of(channels)
                    .map(channel -> channel.get(pageCounter))
                    .toArray(Block[]::new);
            pageCounter++;
            return new Page(blocks);
        }
    };
}
 

public Iterator<Page> getSortedPages()
{
    return new AbstractIterator<Page>()
    {
        private int currentPosition;
        private final PageBuilder pageBuilder = new PageBuilder(types);
        private final int[] outputChannels = new int[types.size()];

        {
            Arrays.setAll(outputChannels, IntUnaryOperator.identity());
        }

        @Override
        public Page computeNext()
        {
            currentPosition = buildPage(currentPosition, outputChannels, pageBuilder);
            if (pageBuilder.isEmpty()) {
                return endOfData();
            }
            Page page = pageBuilder.build();
            pageBuilder.reset();
            return page;
        }
    };
}
 

Iterator<Partition<T>> beginConsumption()
{
    Queue<Partition<T>> partitions = new ArrayDeque<>(requireNonNull(this.partitions, "partitions is already null"));
    if (consumed.incrementAndGet() >= consumersCount) {
        // Unreference futures to allow GC
        this.partitions = null;
    }
    return new AbstractIterator<Partition<T>>()
    {
        @Override
        protected Partition<T> computeNext()
        {
            Partition<T> next = partitions.poll();
            if (next != null) {
                return next;
            }
            else {
                return endOfData();
            }
        }
    };
}
 

@Override
public Iterator<T> iteratorAfter(T segment) {
	int searchResult = 1 + Collections.binarySearch(contents, segment, new RegionComparator<T>(getRegionAccess()));
	if (searchResult < 1) {
		return Collections.emptyIterator();
	}
	return new AbstractIterator<T>() {
		private int index = searchResult;

		@Override
		protected T computeNext() {
			if (index >= contents.size())
				return endOfData();
			return contents.get(index++);
		}
	};
}
 

private static <T> Iterable<T> insertNullEvery(int n, Iterable<T> iterable)
{
    return () -> new AbstractIterator<T>()
    {
        private final Iterator<T> delegate = iterable.iterator();
        private int position;

        @Override
        protected T computeNext()
        {
            position++;
            if (position > n) {
                position = 0;
                return null;
            }

            if (!delegate.hasNext()) {
                return endOfData();
            }

            return delegate.next();
        }
    };
}
 

/**
 * Returns an iterable over the lines in the string. If the string ends in a newline, a final
 * empty string is not included to match the behavior of BufferedReader/LineReader.readLine().
 */
private Iterable<String> lines() {
  return new Iterable<String>() {
    @Override
    public Iterator<String> iterator() {
      return new AbstractIterator<String>() {
        Iterator<String> lines = LINE_SPLITTER.split(seq).iterator();

        @Override
        protected String computeNext() {
          if (lines.hasNext()) {
            String next = lines.next();
            // skip last line if it's empty
            if (lines.hasNext() || !next.isEmpty()) {
              return next;
            }
          }
          return endOfData();
        }
      };
    }
  };
}
 

private static <T> Iterable<T> skipEvery(int n, Iterable<T> iterable)
{
    return () -> new AbstractIterator<T>()
    {
        private final Iterator<T> delegate = iterable.iterator();
        private int position;

        @Override
        protected T computeNext()
        {
            while (true) {
                if (!delegate.hasNext()) {
                    return endOfData();
                }

                T next = delegate.next();
                position++;
                if (position <= n) {
                    return next;
                }
                position = 0;
            }
        }
    };
}
 

private static <T> Iterable<T> repeatEach(int n, Iterable<T> iterable)
{
    return () -> new AbstractIterator<T>()
    {
        private final Iterator<T> delegate = iterable.iterator();
        private int position;
        private T value;

        @Override
        protected T computeNext()
        {
            if (position == 0) {
                if (!delegate.hasNext()) {
                    return endOfData();
                }
                value = delegate.next();
            }

            position++;
            if (position >= n) {
                position = 0;
            }
            return value;
        }
    };
}
 

@Override
protected Integer computeNext()
{
    RepValueIterator current = repValueIteratorList.get(iteratorIndex);
    while (iteratorIndex > 0 && current.end()) {
        current = repValueIteratorList.get(--iteratorIndex);
    }

    while (current.hasNext()) {
        RepetitionLevel currentRepValue = current.next();
        if (currentRepValue.isNull() || iteratorIndex == repValueIteratorList.size() - 1) {
            return currentRepValue.value();
        }
        int lastValue = currentRepValue.value();
        current = repValueIteratorList.get(iteratorIndex + 1);
        current.setBase(lastValue);
        iteratorIndex++;
    }
    checkState(repValueIteratorList.stream().noneMatch(AbstractIterator::hasNext));
    return endOfData();
}
 

@Override
protected Integer computeNext()
{
    DefLevelIterator current = iterators.get(iteratorIndex);
    while (iteratorIndex > 0 && current.end()) {
        iteratorIndex--;
        current = iterators.get(iteratorIndex);
    }

    while (current.hasNext()) {
        OptionalInt next = current.next();
        if (next.isPresent()) {
            return next.getAsInt();
        }
        iteratorIndex++;
        current = iterators.get(iteratorIndex);
    }
    checkState(iterators.stream().noneMatch(AbstractIterator::hasNext));
    return endOfData();
}
 

public static TaskStateChanges create(final TaskExecution previousExecution, final TaskExecution currentExecution, final FileCollectionSnapshot inputFilesSnapshot) {
    return new TaskStateChanges() {
        public Iterator<TaskStateChange> iterator() {
            if (previousExecution.getInputFilesSnapshot() == null) {
                return Collections.<TaskStateChange>singleton(new DescriptiveChange("Input file history is not available.")).iterator();
            }

            return new AbstractIterator<TaskStateChange>() {
                final FileCollectionSnapshot.ChangeIterator<String> changeIterator = inputFilesSnapshot.iterateChangesSince(previousExecution.getInputFilesSnapshot());
                final ChangeListenerAdapter listenerAdapter = new ChangeListenerAdapter();

                @Override
                protected TaskStateChange computeNext() {
                    if (changeIterator.next(listenerAdapter)) {
                        return listenerAdapter.lastChange;
                    }
                    return endOfData();
                }
            };
        }

        public void snapshotAfterTask() {
            currentExecution.setInputFilesSnapshot(inputFilesSnapshot);
        }
    };
}
 

/**
 * Provides an efficient summary of the changes, without doing too much unnecessary work.
 * - Will only emit changes of a single type (from a single delegate change set)
 * - Will return no more than the specified maximum of number of changes
 */
public Iterator<TaskStateChange> iterator() {

    return new AbstractIterator<TaskStateChange>() {
        Iterator<TaskStateChange> changes;
        int count;

        @Override
        protected TaskStateChange computeNext() {
            if (changes == null) {
                changes = firstDirtyIterator();
            }

            if (count < maxReportedChanges && changes != null && changes.hasNext()) {
                count++;
                return changes.next();
            }
            return endOfData();
        }
    };
}
 

@Override
public Iterable<IResourceDescription> getAllResourceDescriptions() {
	if (data != null) {
		return data.getAllResourceDescriptions();
	}
	return Iterables.filter(new Iterable<IResourceDescription>() {
		@Override
		public Iterator<IResourceDescription> iterator() {
			return new AbstractIterator<IResourceDescription>() {
				int index = 0;
				List<Resource> resources = resourceSet.getResources();

				@Override
				protected IResourceDescription computeNext() {
					if (resources.size() <= index)
						return endOfData();
					Resource resource = resources.get(index);
					index++;
					return getResourceDescription(resource.getURI());
				}
			};
		}
	}, Predicates.notNull());

}
 

@Override
public Iterator<C> iterator() {
  return new AbstractIterator<C>() {
    Iterator<Object[]> it = map.values().iterator();
    Object[] counters = it.hasNext() ? it.next() : null;
    int i = 0;
    @Override
    protected C computeNext() {
      while (counters != null) {
        while (i < counters.length) {
          @SuppressWarnings("unchecked")
          C counter = (C) counters[i++];
          if (counter != null) return counter;
        }
        i = 0;
        counters = it.hasNext() ? it.next() : null;
      }
      return endOfData();
    }
  };
}
 

@Override public Iterable<AbstractMetric> metrics() {
  return new Iterable<AbstractMetric>() {
    final Iterator<AbstractMetric> it = delegate.metrics().iterator();
    @Override public Iterator<AbstractMetric> iterator() {
      return new AbstractIterator<AbstractMetric>() {
        @Override public AbstractMetric computeNext() {
          while (it.hasNext()) {
            AbstractMetric next = it.next();
            if (filter.accepts(next.name())) {
              return next;
            }
          }
          return endOfData();
        }
      };
    }
  };
}
 

public static <T extends Object> Iterator<T> getWrappingIterator(final int startIdx, final List<T> list)
{
	return new AbstractIterator<T>()
	{
		private int idx = Math.max(0, Math.min(startIdx, list.size() - 1));

		@Override
		protected T computeNext()
		{
			T o = list.get(idx);

			idx++;
			if(idx >= list.size())
			{
				idx = 0;
			}

			return o;
		}
	};
}
 

/**
 * Returns an iterable over the lines in the string. If the string ends in a newline, a final
 * empty string is not included to match the behavior of BufferedReader/LineReader.readLine().
 */

private Iterable<String> lines() {
  return new Iterable<String>() {
    @Override
    public Iterator<String> iterator() {
      return new AbstractIterator<String>() {
        Iterator<String> lines = LINE_SPLITTER.split(seq).iterator();


        @Override
        protected String computeNext() {
          if (lines.hasNext()) {
            String next = lines.next();
            // skip last line if it's empty
            if (lines.hasNext() || !next.isEmpty()) {
              return next;
            }
          }
          return endOfData();
        }
      };
    }
  };
        }
 

@Override
public Iterator<ObjectEntry<T, R>> iterator() {
	return new AbstractIterator<ObjectEntry<T, R>>() {
		ObjectEntry<T, R> next = first;

		@Override
		protected ObjectEntry<T, R> computeNext() {
			if (next == null) {
				return endOfData();
			}
			ObjectEntry<T, R> current = next;
			next = next.getTrailingObject();
			return current;
		}
	};
}
 

public Iterable<SeparatorEntry<T, R>> separators() {
	return new Iterable<SeparatorEntry<T, R>>() {
		@Override
		public Iterator<SeparatorEntry<T, R>> iterator() {

			return new AbstractIterator<SeparatorEntry<T, R>>() {
				SeparatorEntry<T, R> next = first.getTrailingSeparator();

				@Override
				protected SeparatorEntry<T, R> computeNext() {
					if (next == null) {
						return endOfData();
					}
					SeparatorEntry<T, R> current = next;
					next = next.getTrailingSeparator();
					return current;
				}
			};
		}
	};
}
 

/**
 * Returns a view on this iterator that provides at most the first <code>count</code> entries.
 * 
 * @param iterator
 *            the iterator. May not be <code>null</code>.
 * @param count
 *            the number of elements that should be returned at most.
 * @return an iterator with <code>count</code> elements. Never <code>null</code>.
 * @throws IllegalArgumentException
 *             if <code>count</code> is negative.
 */
@Pure
public static <T> Iterator<T> take(final Iterator<T> iterator, final int count) {
	if (iterator == null)
		throw new NullPointerException("iterator");
	if (count < 0)
		throw new IllegalArgumentException("Cannot take a negative number of elements. Argument 'count' was: "
				+ count);
	if (count == 0)
		return ImmutableSet.<T>of().iterator();
	return new AbstractIterator<T>() {

		private int remaining = count;

		@Override
		protected T computeNext() {
			if (remaining <= 0)
				return endOfData();
			if (!iterator.hasNext())
				return endOfData();
			remaining--;
			return iterator.next();
		}
	};
}
 

/**
 * Returns a view on this iterator that provides all elements except the first <code>count</code> entries.
 * 
 * @param iterator
 *            the iterator. May not be <code>null</code>.
 * @param count
 *            the number of elements that should be dropped.
 * @return an iterator without the first <code>count</code> elements. Never <code>null</code>.
 * @throws IllegalArgumentException
 *             if <code>count</code> is negative.
 */
public static <T> Iterator<T> drop(final Iterator<T> iterator, final int count) {
	if (iterator == null)
		throw new NullPointerException("iterator");
	if (count == 0)
		return iterator;
	if (count < 0)
		throw new IllegalArgumentException("Cannot drop a negative number of elements. Argument 'count' was: "
				+ count);
	return new AbstractIterator<T>() {
		{
			int i = count;
			while (i > 0 && iterator.hasNext()) {
				iterator.next();
				i--;
			}
		}

		@Override
		protected T computeNext() {
			if (!iterator.hasNext())
				return endOfData();
			return iterator.next();
		}
	};
}
 

/**
 * Returns an Iterator containing all elements starting from the head of the source up to and excluding the first
 * element that violates the predicate. The resulting Iterator is a lazily computed view, so any modifications to the
 * underlying Iterators will be reflected on iteration. The result does not support {@link Iterator#remove()}
 * 
 * @param iterator
 *            the elements from which to take. May not be <code>null</code>.
 * @param predicate
 *            the predicate which decides whether to keep taking elements. May not be <code>null</code>.
 * @return the taken elements
 * @since 2.7
 */
public static <T> Iterator<T> takeWhile(final Iterator<? extends T> iterator, final Function1<? super T, Boolean> predicate) {
	if (iterator == null)
		throw new NullPointerException("iterator");
	if (predicate == null)
		throw new NullPointerException("predicate");
	return new AbstractIterator<T>() {
		@Override
		protected T computeNext() {
			if (!iterator.hasNext())
				return endOfData();
			T next = iterator.next();
			if (predicate.apply(next)) {
				return next;
			} else {
				return endOfData();
			}
		}
	};
}
 

/**
 * Returns an Iterator of Pairs where the nth pair is created by taking the nth element of the source as the value
 * and its 0-based index as the key. E.g.
 * <code>zipWitIndex(#["a", "b", "c"]) == #[(0, "a"), (1, "b"), (2, "c")]</code>
 * 
 * If the index would overflow, {@link Integer#MAX_VALUE} is returned for all subsequent elements.
 * 
 * The resulting Iterator is a lazily computed view, so any modifications to the underlying Iterator will be
 * reflected on iteration. The result does not support {@link Iterator#remove()}
 * 
 * @param iterator
 *            the elements. May not be <code>null</code>.
 * @return the zipped result
 * @since 2.7
 */
public static <A> Iterator<Pair<Integer, A>> indexed(final Iterator<? extends A> iterator) {
	if (iterator == null)
		throw new NullPointerException("iterator");
	return new AbstractIterator<Pair<Integer, A>>() {
		int i = 0;
		@Override
		protected Pair<Integer, A> computeNext() {
			if (iterator.hasNext()) {
				Pair<Integer, A> next = new Pair<Integer, A>(i, iterator.next());
				if (i != Integer.MAX_VALUE)
					i++;
				return next;
			} else {
				return endOfData();
			}
		}
	};
}
 

/**
 * Creates an Iterator of FetchedMessage based on the given message set. The iterator would also updates
 * the offset while iterating.
 */
private Iterator<FetchedMessage> createFetchedMessages(ByteBufferMessageSet messageSet, final AtomicLong offset) {
  final Iterator<MessageAndOffset> messages = messageSet.iterator();
  return new AbstractIterator<FetchedMessage>() {
    @Override
    protected FetchedMessage computeNext() {
      while (messages.hasNext()) {
        MessageAndOffset message = messages.next();
        long msgOffset = message.offset();
        if (msgOffset < offset.get()) {
          LOG.trace("Received old offset {}, expecting {} on {}. Message Ignored.",
                    msgOffset, offset.get(), topicPart);
          continue;
        }

        fetchedMessage.setPayload(message.message().payload());
        fetchedMessage.setOffset(message.offset());
        fetchedMessage.setNextOffset(message.nextOffset());

        return fetchedMessage;
      }
      return endOfData();
    }
  };
}
 

@Override
public Iterator<C> iterator() {
  return new AbstractIterator<C>() {
    Iterator<Object[]> it = map.values().iterator();
    Object[] counters = it.hasNext() ? it.next() : null;
    int i = 0;
    @Override
    protected C computeNext() {
      while (counters != null) {
        while (i < counters.length) {
          @SuppressWarnings("unchecked")
          C counter = (C) counters[i++];
          if (counter != null) return counter;
        }
        i = 0;
        counters = it.hasNext() ? it.next() : null;
      }
      return endOfData();
    }
  };
}