Java 类com.google.common.collect.PeekingIterator 实例源码

/**
 * Retrieves non-overlapping bounds for the list of input bounds
 *
 * Assume we have the following bounds
 * (brackets representing left/right bound):
 * [   ] [   ]    [   ]   [  ]
 * [   ]         [       ]
 * This method will return the following bounds:
 * [         ]    [          ]
 *
 * @param bounds unsorted bounds to find overlaps
 * @return the non-overlapping bounds
 */
public static <T extends RingPosition<T>> Set<Bounds<T>> getNonOverlappingBounds(Iterable<Bounds<T>> bounds)
{
    ArrayList<Bounds<T>> sortedBounds = Lists.newArrayList(bounds);
    Collections.sort(sortedBounds, new Comparator<Bounds<T>>()
    {
        public int compare(Bounds<T> o1, Bounds<T> o2)
        {
            return o1.left.compareTo(o2.left);
        }
    });

    Set<Bounds<T>> nonOverlappingBounds = Sets.newHashSet();

    PeekingIterator<Bounds<T>> it = Iterators.peekingIterator(sortedBounds.iterator());
    while (it.hasNext())
    {
        Bounds<T> beginBound = it.next();
        Bounds<T> endBound = beginBound;
        while (it.hasNext() && endBound.right.compareTo(it.peek().left) >= 0)
            endBound = it.next();
        nonOverlappingBounds.add(new Bounds<>(beginBound.left, endBound.right));
    }

    return nonOverlappingBounds;
}

private void visitStatements(List<? extends StatementTree> statements) {
    boolean first = true;
    PeekingIterator<StatementTree> it =
            Iterators.<StatementTree>peekingIterator(statements.iterator());
    dropEmptyDeclarations();
    while (it.hasNext()) {
        StatementTree tree = it.next();
        builder.forcedBreak();
        if (!first) {
            builder.blankLineWanted(BlankLineWanted.PRESERVE);
        }
        markForPartialFormat();
        first = false;
        List<VariableTree> fragments = variableFragments(it, tree);
        if (!fragments.isEmpty()) {
            visitVariables(
                    fragments,
                    DeclarationKind.NONE,
                    canLocalHaveHorizontalAnnotations(fragments.get(0).getModifiers()));
        } else {
            scan(tree, null);
        }
    }
}

/**
 * Replaces whitespace after a {@code href=...>} token with an "optional link break." This allows
 * us to output either {@code <a href=foo>foo</a>} or {@code <a href=foo>\nfoo</a>}, depending on
 * how much space we have left on the line.
 * <p>
 * <p>This method must be called after {@link #joinAdjacentLiteralsAndAdjacentWhitespace}, as it
 * assumes that adjacent whitespace tokens have already been joined.
 */
private static ImmutableList<Token> optionalizeSpacesAfterLinks(List<Token> input) {
    ImmutableList.Builder<Token> output = ImmutableList.builder();

    for (PeekingIterator<Token> tokens = peekingIterator(input.iterator()); tokens.hasNext(); ) {
        if (tokens.peek().getType() == LITERAL && tokens.peek().getValue().matches("^href=[^>]*>")) {
            output.add(tokens.next());

            if (tokens.peek().getType() == WHITESPACE) {
                output.add(new Token(OPTIONAL_LINE_BREAK, tokens.next().getValue()));
            }
        } else {
            output.add(tokens.next());
        }
    }

    return output.build();

/*
 * Note: We do not want to insert <p> tags inside <pre>. Fortunately, the formatter gets that
 * right without special effort on our part. The reason: Line breaks inside a <pre> section are
 * of type FORCED_NEWLINE rather than WHITESPACE.
 */
}

/**
 * Adjust indentation inside `<pre>{@code` blocks.
 * <p>
 * <p>Also trim leading and trailing blank lines, and move the trailing `}` to its own line.
 */
private static ImmutableList<Token> deindentPreCodeBlocks(List<Token> input) {
    ImmutableList.Builder<Token> output = ImmutableList.builder();
    for (PeekingIterator<Token> tokens = peekingIterator(input.iterator()); tokens.hasNext(); ) {
        if (tokens.peek().getType() != PRE_OPEN_TAG) {
            output.add(tokens.next());
            continue;
        }

        output.add(tokens.next());
        List<Token> initialNewlines = new ArrayList<>();
        while (tokens.hasNext() && tokens.peek().getType() == FORCED_NEWLINE) {
            initialNewlines.add(tokens.next());
        }
        if (tokens.peek().getType() != LITERAL
                || !tokens.peek().getValue().matches("[ \t]*[{]@code")) {
            output.addAll(initialNewlines);
            output.add(tokens.next());
            continue;
        }

        deindentPreCodeBlock(output, tokens);
    }
    return output.build();
}

private InstanceParsePair<NlpFocus<DepNode, DepTree>> nextTree(PeekingIterator<NlpFocus<DepNode,
        DepTree>> iterator) {
    if (!iterator.hasNext()) {
        return null;
    }
    List<NlpFocus<DepNode, DepTree>> instances = new ArrayList<>();
    NlpFocus<DepNode, DepTree> current = iterator.next();
    DepTree tree = current.sequence();
    instances.add(current);
    while (iterator.hasNext()) {
        current = iterator.peek();
        if (tree.index() == current.sequence().index()) {
            instances.add(iterator.next());
            continue;
        }
        break;
    }
    return new InstanceParsePair<>(instances, (DefaultDepTree) tree);
}

/**
 * Converts a set of integers into a set of contiguous closed ranges that equally represent the
 * input integers.
 * <p>
 * The resulting ranges will be in ascending order.
 * <p>
 * TODO(wfarner): Change this to return a canonicalized RangeSet.
 *
 * @param values Values to transform to ranges.
 * @return Closed ranges with identical members to the input set.
 */
public static Set<Range<Integer>> toRanges(Iterable<Integer> values) {
  ImmutableSet.Builder<Range<Integer>> builder = ImmutableSet.builder();

  PeekingIterator<Integer> iterator =
      Iterators.peekingIterator(Sets.newTreeSet(values).iterator());

  // Build ranges until there are no numbers left.
  while (iterator.hasNext()) {
    // Start a new range.
    int start = iterator.next();
    int end = start;
    // Increment the end until the range is non-contiguous.
    while (iterator.hasNext() && iterator.peek() == end + 1) {
      end++;
      iterator.next();
    }

    builder.add(Range.closed(start, end));
  }

  return builder.build();
}

private void expectStateTransitions(
    String taskId,
    ScheduleStatus initial,
    ScheduleStatus next,
    ScheduleStatus... others) {

  List<ScheduleStatus> statuses = ImmutableList.<ScheduleStatus>builder()
      .add(initial)
      .add(next)
      .add(others)
      .build();
  PeekingIterator<ScheduleStatus> it = Iterators.peekingIterator(statuses.iterator());
  while (it.hasNext()) {
    ScheduleStatus cur = it.next();
    try {
      eventSink.post(matchStateChange(taskId, cur, it.peek()));
    } catch (NoSuchElementException e) {
      // Expected.
    }
  }
}

private static PZXBlock getPzxBlock(PeekingIterator<TapeBlock> iterator) {
    TapeBlock block = iterator.next();
    PulseList blockPulseList = block.getPulseList();
    PZXBlock pzxBlock = new PZXNullBlock();
    switch (block.getBlockType()) {
        case UNKNOWN:
            // TODO if this is the beginning of the tape and followed by a pilot discard
            // TODO if this is about a second after a data block with a tail but before a pulse block use a pause block?
            pzxBlock = new PZXPulseBlock(blockPulseList);
            break;
        case PILOT:
            pzxBlock = getPzxPulseBlock(iterator, blockPulseList);
            break;
        case SYNC_CANDIDATE:
            pzxBlock = new PZXPulseBlock(blockPulseList);
            break;
        case DATA:
            pzxBlock = getPzxDataBlock(iterator, block);
            break;
        case TAIL_CANDIDATE:
            pzxBlock = new PZXPulseBlock(blockPulseList);
            break;
    }
    return pzxBlock;
}

private static PZXDataBlock getPzxDataBlock(PeekingIterator<TapeBlock> iterator, TapeBlock block) {
    long tailLength = getTailLength(iterator);

    List<Long> zeroPulseLengths = block.getZeroBit().getPulses();
    List<Long> onePulseLengths = block.getOneBit().getPulses();
    DataBuilder dataBuilder = new DataBuilder();

    ImmutableList<Long> pulseLengths = block.getPulseList().getPulseLengths();
    for (int i = 0; i < pulseLengths.size(); i += 2) {
        ImmutableList<Long> pulses = pulseLengths.subList(i, i + 2);
        if (isSpecifiedPulseSequence(zeroPulseLengths, pulses)) {
            dataBuilder.addBit(0);
        } else if (isSpecifiedPulseSequence(onePulseLengths, pulses)) {
            dataBuilder.addBit(1);
        }
        // FIXME: Some kind of error, fall back to PulseBlock?
    }

    int numBitsInLastByte = dataBuilder.getNumBitsInCurrentByte();
    return new PZXDataBlock(block.getPulseList(), zeroPulseLengths, onePulseLengths, tailLength,
            numBitsInLastByte, dataBuilder.getData());
}

@Test
public void testExpirationTime() {
    long start = System.currentTimeMillis();
    Iterator<Long> unlimitedIter = countForever();
    PeekingIterator<Long> limitedIter = TimeLimitedIterator.create(unlimitedIter, Duration.millis(10), 0);
    long previous = -1;
    while (limitedIter.hasNext()) {
        long next = limitedIter.next();
        assertEquals(next, previous + 1);
        previous = next;
    }
    long stop = System.currentTimeMillis();
    long elapsed = stop - start;

    assertTrue(elapsed >= 10);
    assertTrue(elapsed < 100);  // be fairly relaxed about the upper bound to avoid spurious test failures on slow machines.
    assertEquals(unlimitedIter.next(), (previous + 1));
}

void readAll(String channel, SlabFilter filter, EventSink sink, ConsistencyLevel consistency) {
    // PeekingIterator is needed so that we can look ahead and see the next slab Id
    PeekingIterator<Column<ByteBuffer>> manifestColumns = Iterators.peekingIterator(executePaginated(
            _keyspace.prepareQuery(ColumnFamilies.MANIFEST, consistency)
                    .getKey(channel)
                    .withColumnRange(new RangeBuilder().setLimit(50).build())
                    .autoPaginate(true)).iterator());

    while (manifestColumns.hasNext()) {
        Column<ByteBuffer> manifestColumn = manifestColumns.next();
        ByteBuffer slabId = manifestColumn.getName();
        ByteBuffer nextSlabId = manifestColumns.hasNext() ? manifestColumns.peek().getName() : null;
        boolean open = manifestColumn.getBooleanValue();
        if (filter != null && !filter.accept(slabId, open, nextSlabId)) {
            continue;
        }
        if (!readSlab(channel, slabId, new SlabCursor(), open, sink)) {
            break;
        }
    }
}

private PeekingIterator<ILeafNode> createReversedLeafIterator(INode root, INode candidate, LinkedList<ILeafNode> sameGrammarElement) {
    EObject grammarElement = null;
    PeekingIterator<ILeafNode> iterator = Iterators.peekingIterator(Iterators.filter(root.getAsTreeIterable().reverse().iterator(), ILeafNode.class));
    // traverse until we find the current candidate
    while(iterator.hasNext()) {
        ILeafNode next = iterator.next();
        if (candidate.equals(next)) {
            break;
        } else if (next.getTotalLength() == 0) {
            EObject otherGrammarElement = tryGetGrammarElementAsRule(next);
            if (grammarElement == null) {
                grammarElement = otherGrammarElement;
            }
            if (otherGrammarElement.equals(grammarElement)) {
                sameGrammarElement.add(next);
            } else {
                sameGrammarElement.removeLast();
            }
        }
    }
    return iterator;
}

/**
 * Restore the contents of one transaction to Datastore.
 *
 * <p>The objects to delete are listed in the {@link CommitLogManifest}, which will be the first
 * object in the iterable. The objects to save follow, each as a {@link CommitLogMutation}. We
 * restore by deleting the deletes and recreating the saves from their proto form. We also save
 * the commit logs themselves back to Datastore, so that the commit log system itself is
 * transparently restored alongside the data.
 *
 * @return the manifest, for use in restoring the {@link CommitLogBucket}.
 */
private CommitLogManifest restoreOneTransaction(PeekingIterator<ImmutableObject> commitLogs) {
  final CommitLogManifest manifest = (CommitLogManifest) commitLogs.next();
  Result<?> deleteResult = deleteAsync(manifest.getDeletions());
  List<Entity> entitiesToSave = Lists.newArrayList(ofy().save().toEntity(manifest));
  while (commitLogs.hasNext() && commitLogs.peek() instanceof CommitLogMutation) {
    CommitLogMutation mutation = (CommitLogMutation) commitLogs.next();
    entitiesToSave.add(ofy().save().toEntity(mutation));
    entitiesToSave.add(EntityTranslator.createFromPbBytes(mutation.getEntityProtoBytes()));
  }
  saveRaw(entitiesToSave);
  try {
    deleteResult.now();
  } catch (Exception e) {
    retrier.callWithRetry(
        () -> deleteAsync(manifest.getDeletions()).now(), RuntimeException.class);
  }
  return manifest;
}

/**
 * Attempt to match the desired properties to a sequence of known properties.
 * <p>
 * Returns a list of the same length as the original. Entries are:
 * - Optional.empty(): the property was satisfied completely
 * - non-empty: the (simplified) property that was not satisfied
 */
public static <T> List<Optional<LocalProperty<T>>> match(List<LocalProperty<T>> actuals, List<LocalProperty<T>> desired)
{
    // After normalizing actuals, each symbol should only appear once
    PeekingIterator<LocalProperty<T>> actualIterator = peekingIterator(normalizeAndPrune(actuals).iterator());

    Set<T> constants = new HashSet<>();
    boolean consumeMoreActuals = true;
    List<Optional<LocalProperty<T>>> result = new ArrayList<>(desired.size());
    for (LocalProperty<T> desiredProperty : desired) {
        while (consumeMoreActuals && actualIterator.hasNext() && desiredProperty.isSimplifiedBy(actualIterator.peek())) {
            constants.addAll(actualIterator.next().getColumns());
        }
        Optional<LocalProperty<T>> simplifiedDesired = desiredProperty.withConstants(constants);
        consumeMoreActuals &= !simplifiedDesired.isPresent(); // Only continue processing actuals if all previous desired properties were fully satisfied
        result.add(simplifiedDesired);
    }
    return result;
}

private void visitStatements(List<? extends StatementTree> statements) {
  boolean first = true;
  PeekingIterator<StatementTree> it = Iterators.peekingIterator(statements.iterator());
  dropEmptyDeclarations();
  while (it.hasNext()) {
    StatementTree tree = it.next();
    builder.forcedBreak();
    if (!first) {
      builder.blankLineWanted(BlankLineWanted.PRESERVE);
    }
    markForPartialFormat();
    first = false;
    List<VariableTree> fragments = variableFragments(it, tree);
    if (!fragments.isEmpty()) {
      visitVariables(
          fragments,
          DeclarationKind.NONE,
          canLocalHaveHorizontalAnnotations(fragments.get(0).getModifiers()));
    } else {
      scan(tree, null);
    }
  }
}

/**
 * Replaces whitespace after a {@code href=...>} token with an "optional link break." This allows
 * us to output either {@code <a href=foo>foo</a>} or {@code <a href=foo>\nfoo</a>}, depending on
 * how much space we have left on the line.
 *
 * <p>This method must be called after {@link #joinAdjacentLiteralsAndAdjacentWhitespace}, as it
 * assumes that adjacent whitespace tokens have already been joined.
 */
private static ImmutableList<Token> optionalizeSpacesAfterLinks(List<Token> input) {
  ImmutableList.Builder<Token> output = ImmutableList.builder();

  for (PeekingIterator<Token> tokens = peekingIterator(input.iterator()); tokens.hasNext(); ) {
    if (tokens.peek().getType() == LITERAL && tokens.peek().getValue().matches("^href=[^>]*>")) {
      output.add(tokens.next());

      if (tokens.peek().getType() == WHITESPACE) {
        output.add(new Token(OPTIONAL_LINE_BREAK, tokens.next().getValue()));
      }
    } else {
      output.add(tokens.next());
    }
  }

  return output.build();

  /*
   * Note: We do not want to insert <p> tags inside <pre>. Fortunately, the formatter gets that
   * right without special effort on our part. The reason: Line breaks inside a <pre> section are
   * of type FORCED_NEWLINE rather than WHITESPACE.
   */
}

/**
 * Adjust indentation inside `<pre>{@code` blocks.
 *
 * <p>Also trim leading and trailing blank lines, and move the trailing `}` to its own line.
 */
private static ImmutableList<Token> deindentPreCodeBlocks(List<Token> input) {
  ImmutableList.Builder<Token> output = ImmutableList.builder();
  for (PeekingIterator<Token> tokens = peekingIterator(input.iterator()); tokens.hasNext(); ) {
    if (tokens.peek().getType() != PRE_OPEN_TAG) {
      output.add(tokens.next());
      continue;
    }

    output.add(tokens.next());
    List<Token> initialNewlines = new ArrayList<>();
    while (tokens.hasNext() && tokens.peek().getType() == FORCED_NEWLINE) {
      initialNewlines.add(tokens.next());
    }
    if (tokens.peek().getType() != LITERAL
        || !tokens.peek().getValue().matches("[ \t]*[{]@code")) {
      output.addAll(initialNewlines);
      output.add(tokens.next());
      continue;
    }

    deindentPreCodeBlock(output, tokens);
  }
  return output.build();
}

@Override
public GTRecord next() {
    if (!hasNext()) {
        throw new NoSuchElementException();
    }
    // get smallest record
    PeekingIterator<GTRecord> it = heap.poll();
    // WATCH OUT! record got from PartitionResultIterator.next() may changed later,
    // so we must make a shallow copy of it.
    record.shallowCopyFrom(it.next());

    if (it.hasNext()) {
        heap.offer(it);
    }

    return record;
}

/**
 * Retrieves non-overlapping bounds for the list of input bounds
 *
 * Assume we have the following bounds
 * (brackets representing left/right bound):
 * [   ] [   ]    [   ]   [  ]
 * [   ]         [       ]
 * This method will return the following bounds:
 * [         ]    [          ]
 *
 * @param bounds unsorted bounds to find overlaps
 * @return the non-overlapping bounds
 */
public static <T extends RingPosition<T>> Set<Bounds<T>> getNonOverlappingBounds(Iterable<Bounds<T>> bounds)
{
    ArrayList<Bounds<T>> sortedBounds = Lists.newArrayList(bounds);
    Collections.sort(sortedBounds, new Comparator<Bounds<T>>()
    {
        public int compare(Bounds<T> o1, Bounds<T> o2)
        {
            return o1.left.compareTo(o2.left);
        }
    });

    Set<Bounds<T>> nonOverlappingBounds = Sets.newHashSet();

    PeekingIterator<Bounds<T>> it = Iterators.peekingIterator(sortedBounds.iterator());
    while (it.hasNext())
    {
        Bounds<T> beginBound = it.next();
        Bounds<T> endBound = beginBound;
        while (it.hasNext() && endBound.right.compareTo(it.peek().left) >= 0)
            endBound = it.next();
        nonOverlappingBounds.add(new Bounds<>(beginBound.left, endBound.right));
    }

    return nonOverlappingBounds;
}

private void updateForRow(PeekingIterator<PartitionUpdate.CounterMark> markIter, Row row, ColumnFamilyStore cfs)
{
    int cmp = 0;
    // If the mark is before the row, we have no value for this mark, just consume it
    while (markIter.hasNext() && (cmp = compare(markIter.peek().clustering(), row.clustering(), cfs)) < 0)
        markIter.next();

    if (!markIter.hasNext())
        return;

    while (cmp == 0)
    {
        PartitionUpdate.CounterMark mark = markIter.next();
        Cell cell = mark.path() == null ? row.getCell(mark.column()) : row.getCell(mark.column(), mark.path());
        if (cell != null)
        {
            updateWithCurrentValue(mark, CounterContext.instance().getLocalClockAndCount(cell.value()), cfs);
            markIter.remove();
        }
        if (!markIter.hasNext())
            return;

        cmp = compare(markIter.peek().clustering(), row.clustering(), cfs);
    }
}

Description scanBlock(MethodTree tree, BlockTree block, VisitorState state) {
  PeekingIterator<? extends StatementTree> it =
      Iterators.peekingIterator(block.getStatements().iterator());
  while (it.hasNext() && !MATCHER.matches(it.peek(), state)) {
    it.next();
  }
  List<Tree> expectations = new ArrayList<>();
  while (it.hasNext() && MATCHER.matches(it.peek(), state)) {
    expectations.add(it.next());
  }
  if (expectations.isEmpty()) {
    return NO_MATCH;
  }
  Deque<StatementTree> suffix = new ArrayDeque<>();
  StatementTree failure = null;
  Iterators.addAll(suffix, it);
  if (!suffix.isEmpty() && FAIL_MATCHER.matches(suffix.peekLast(), state)) {
    failure = suffix.removeLast();
  }
  return handleMatch(tree, state, expectations, ImmutableList.copyOf(suffix), failure);
}

@Override
public Iterator<List<T>> iterator() {
  final PeekingIterator<T> iter = Iterators.peekingIterator(delegate.iterator());
  return new AbstractIterator<List<T>>() {
    @Override
    protected List<T> computeNext() {
      if (!iter.hasNext()) {
        return endOfData();
      }
      ArrayList<T> result = Lists.newArrayList();
      T last = iter.next();
      result.add(last);
      while (iter.hasNext() && equiv.equivalent(last, iter.peek())) {
        last = iter.next(); // get peeked value
        result.add(last);
      }
      return result;
    }
  };
}

private static void writeEntries(JsonGenerator jg, List<Trace.Entry> entries)
        throws IOException {
    jg.writeStartArray();
    PeekingIterator<Trace.Entry> i = Iterators.peekingIterator(entries.iterator());
    while (i.hasNext()) {
        Trace.Entry entry = i.next();
        int depth = entry.getDepth();
        jg.writeStartObject();
        writeJson(entry, jg);
        int nextDepth = i.hasNext() ? i.peek().getDepth() : 0;
        if (nextDepth > depth) {
            jg.writeArrayFieldStart("childEntries");
        } else if (nextDepth < depth) {
            jg.writeEndObject();
            for (int j = depth; j > nextDepth; j--) {
                jg.writeEndArray();
                jg.writeEndObject();
            }
        } else {
            jg.writeEndObject();
        }
    }
    jg.writeEndArray();
}

private void merge(List<Profile.ProfileNode> flatNodes,
        List<ProfileNode> destinationRootNodes) {
    destinationStack.push(destinationRootNodes);
    PeekingIterator<Profile.ProfileNode> i =
            Iterators.peekingIterator(flatNodes.iterator());
    while (i.hasNext()) {
        Profile.ProfileNode flatNode = i.next();
        int destinationDepth = destinationStack.size() - 1;
        for (int j = 0; j < destinationDepth - flatNode.getDepth(); j++) {
            // TODO optimize: faster way to pop multiple elements at once
            destinationStack.pop();
        }
        ProfileNode destinationNode = mergeOne(flatNode, destinationStack.getFirst());
        if (i.hasNext() && i.peek().getDepth() > flatNode.getDepth()) {
            destinationStack.push(destinationNode.childNodes);
        }
    }
}

private void processPolicy(BouncePolicy policy) {
    logger.info("processing policy {} {}", policy.getClass(), status.container);

    ListContainerOptions options = new ListContainerOptions().recursive();

    PeekingIterator<StorageMetadata> destinationIterator = Iterators.peekingIterator(
            Utils.crawlBlobStore(policy.getDestination(), container, options).iterator());
    PeekingIterator<StorageMetadata> sourceIterator = Iterators.peekingIterator(
            Utils.crawlBlobStore(policy, container, options).iterator());

    policy.prepareBounce(container);
    StreamSupport.stream(Spliterators.spliteratorUnknownSize(
            new ReconcileIterator(sourceIterator, destinationIterator), Spliterator.CONCURRENT), true)
            .filter(p -> p.getRight() == null || !WriteBackPolicy.isSwiftSegmentBlob(p.getRight().getName()))
            .forEach((p) -> {
                BounceStorageMetadata sourceObject = p.getLeft();
                StorageMetadata destinationObject = p.getRight();

                reconcileObject(policy, sourceObject, destinationObject);
            });
}

@Override
protected Generation.Entry<K, V> computeNext() {
    if (heap.isEmpty()) {
        return endOfData();
    }

    PeekingIterator<EntryAndGenerationId<K,V>> first = heap.poll();
    EntryAndGenerationId<K,V> ret = first.next();
    if (first.hasNext()) {
        temp.add(first);
    }
    while (!heap.isEmpty() && keyComparator.compare(ret.entry.getKey(), heap.peek().peek().entry.getKey()) == 0) {
        PeekingIterator<EntryAndGenerationId<K, V>> iter = heap.poll();
        iter.next();
        if (iter.hasNext()) {
            temp.add(iter);
        }
    }
    heap.addAll(temp);
    temp.clear();
    return ret.entry;
}

/**
 * Computes the suffix of {@code xs} starting at the first node for which
 * {@code p} fails.
 *
 * {@code Iterables.concat(takeWhile(xs, p), dropWhile(xs, p)) = xs}
 */
private <T> Iterable<T> dropWhile(
    final Iterable<T> xs, final Predicate<? super T> p) {
  return new Iterable<T>() {
    @Override
    public Iterator<T> iterator() {
      PeekingIterator<T> xsi = Iterators.peekingIterator(xs.iterator());
      while (xsi.hasNext()) {
        if (p.apply(xsi.peek())) {
          break;
        }
        xsi.next();
      }
      return xsi;
    }
  };
}

private static List<String> parseMacroArgList(PeekingIterator<Token> tokens) {
    final Token firstToken = expectTokens(tokens, TokenType.SYMBOL, TokenType.RIGHT_BRACKET);

    final List<String> args = Lists.newArrayList();
    if (firstToken.type == TokenType.RIGHT_BRACKET) {
        Preconditions.checkState(firstToken.value.equals(")"), "Unexpected bracket: '%s'", firstToken.value);
        return args;
    }

    args.add(firstToken.value);

    while (true) {
        final Token token = expectTokens(tokens, TokenType.SEPARATOR, TokenType.RIGHT_BRACKET);
        if (token.type == TokenType.RIGHT_BRACKET) {
            Preconditions.checkState(token.value.equals(")"), "Unexpected bracket: '%s'", token.value);
            break;
        }

        final String arg = expectToken(tokens, TokenType.SYMBOL);
        args.add(arg);
    }
    return args;
}

@Override
public IModifierStateTransition<Node> getStateForModifier(String modifier) {
    if (MODIFIER_OP.equals(modifier)) {
        return new SingleStateTransition.ForModifier<Node>() {

            @Override
            public Node createRootNode(Node child) {
                return child;
            }

            @Override
            public Node parseSymbol(IParserState<Node> state, PeekingIterator<Token> input) {
                Preconditions.checkState(input.hasNext(), "Unexpected end out input");
                final Token token = input.next();
                Preconditions.checkState(token.type == TokenType.OPERATOR, "Unexpected token, expected operator, got %s", token);
                NodeOp operator = OPERATORS.getOperator(token.value, OperatorArity.BINARY);
                if (operator == null) operator = OPERATORS.getOperator(token.value, OperatorArity.UNARY);
                if (operator == null) throw new IllegalArgumentException("Unknown operator: " + token.value);
                return new Node(operator);
            }

        };
    } else {
        throw new UnsupportedOperationException("Modifier: " + modifier);
    }
}

private static Row getNextRow(CellGrouper grouper, PeekingIterator<? extends LegacyAtom> cells)
{
    if (!cells.hasNext())
        return null;

    grouper.reset();
    while (cells.hasNext() && grouper.addAtom(cells.peek()))
    {
        // We've added the cell already in the grouper, so just skip it
        cells.next();
    }
    return grouper.getRow();
}

private static Iterator<Row> convertToRows(final CellGrouper grouper, final PeekingIterator<LegacyAtom> atoms)
{
    return new AbstractIterator<Row>()
    {
        protected Row computeNext()
        {
            if (!atoms.hasNext())
                return endOfData();

            return getNextRow(grouper, atoms);
        }
    };
}

@Override
public Stream<ResolvedEvent> readAllForwards(Position positionExclusive) {
    try(Stream<ResolvedEvent> allForwards = underlying.readAllForwards(positionExclusive)) {
        PeekingIterator<ResolvedEvent> allForwardsIterator = Iterators.peekingIterator(allForwards.iterator());

        if (allForwardsIterator.hasNext() && sortKeyExtractor.apply(allForwardsIterator.peek()).compareTo(cutoffSortKey) < 0) {
            allForwards.close();
            return bufferedAndSortedReadAllForwards(positionExclusive);
        }

        return stream(spliteratorUnknownSize(allForwardsIterator, ORDERED), false).onClose(allForwards::close);
    }
}

private Stream<ResolvedEvent> bufferedAndSortedReadAllForwards(Position positionExclusive) {
    try(Stream<ResolvedEvent> allForwards = underlying.readAllForwards()) {

        Iterator<ResolvedEvent> remainder = allForwards.iterator();
        PeekingIterator<EventWithSortKey<T>> sortCandidates = Iterators.peekingIterator(
                Iterators.transform(remainder, re -> new EventWithSortKey<>(re, sortKeyExtractor.apply(re)))
        );

        final LinkedList<EventWithSortKey<T>> buffer = new LinkedList<>();

        while (sortCandidates.hasNext() && sortCandidates.peek().sortKey.compareTo(cutoffSortKey) < 0) {
            buffer.add(sortCandidates.next());
        }

        if (!sortCandidates.hasNext()) {
            return Stream.empty();
        }

        buffer.sort(Comparator.naturalOrder());

        if (!positionExclusive.equals(underlying.emptyStorePosition())) {
            Iterator<EventWithSortKey<T>> bufferIterator = buffer.iterator();
            while (!bufferIterator.next().event.position().equals(positionExclusive)) {
                bufferIterator.remove();
            }
            bufferIterator.remove();
        }

        Stream<EventWithSortKey<T>> reorderedEvents = buffer.stream().onClose(buffer::clear);
        Stream<EventWithSortKey<T>> eventInTheGap = Stream.of(sortCandidates.peek());
        Stream<ResolvedEvent> remainingEvents = stream(spliteratorUnknownSize(remainder, ORDERED), false);

        return concat(concat(reorderedEvents, eventInTheGap).map(EventWithSortKey::toResolvedEvent), remainingEvents).onClose(allForwards::close);
    }
}

private static <T> Iterator<Iterator<T>> batchBy(Iterator<T> it, Function<T, Object> grouping) {
    return new Iterator<Iterator<T>>() {
        PeekingIterator<T> peekingIterator = peekingIterator(it);

        @Override
        public boolean hasNext() {
            return peekingIterator.hasNext();
        }

        @Override
        public Iterator<T> next() {
            return new Iterator<T>() {
                private Object currentGroup = grouping.apply(peekingIterator.peek());

                private boolean isSameGroup() {
                    return currentGroup.equals(grouping.apply(peekingIterator.peek()));
                }

                @Override
                public boolean hasNext() {
                    return peekingIterator.hasNext() && isSameGroup();
                }

                @Override
                public T next() {
                    if (!isSameGroup()) {
                        throw new NoSuchElementException();
                    }
                    return peekingIterator.next();
                }
            };
        }
    };
}

private Iterator<EventInIdentifiedStream> getIteratorWhoseHeadIsNext() {
    if (iteratorWhoseHeadIsNext != null) {
        return iteratorWhoseHeadIsNext;
    }
    Iterator<PeekingIterator<EventInIdentifiedStream>> streams = underlying.iterator();
    while (streams.hasNext()) {
        PeekingIterator<EventInIdentifiedStream> eventStream = streams.next();

        Instant potentialCutoffTime = clock.instant();
        if (eventStream.hasNext()) {
            if (cutOffTime != null && eventStream.peek().event.effectiveTimestamp().isAfter(cutOffTime)) {
                streams.remove();
            } else if (iteratorWhoseHeadIsNext == null || order.compare(eventStream.peek(), iteratorWhoseHeadIsNext.peek()) < 0) {
                iteratorWhoseHeadIsNext = eventStream;
            }
        } else {
            streams.remove();
            if (this.cutOffTime == null) {
                this.cutOffTime = potentialCutoffTime.minus(delay).toEpochMilli();
            }
        }
    }

    if (iteratorWhoseHeadIsNext != null) {
        long cutoff = this.cutOffTime == null ? clock.instant().minus(delay).toEpochMilli() : this.cutOffTime;
        if (iteratorWhoseHeadIsNext.peek().event.effectiveTimestamp().isAfter(cutoff)) {
            underlying.clear();
            return null;
        }
    }
    return iteratorWhoseHeadIsNext;
}

/**
 * The parser expands multi-variable declarations into separate single-variable declarations. All
 * of the fragments in the original declaration have the same start position, so we use that as a
 * signal to collect them and preserve the multi-variable declaration in the output.
 * <p>
 * <p>e.g. {@code int x, y;} is parsed as {@code int x; int y;}.
 */
private List<VariableTree> variableFragments(PeekingIterator<? extends Tree> it, Tree first) {
    List<VariableTree> fragments = new ArrayList<>();
    if (first.getKind() == VARIABLE) {
        int start = getStartPosition(first);
        fragments.add((VariableTree) first);
        while (it.hasNext()
                && it.peek().getKind() == VARIABLE
                && getStartPosition(it.peek()) == start) {
            fragments.add((VariableTree) it.next());
        }
    }
    return fragments;
}

/**
 * Where the input has two consecutive line breaks between literals, insert a {@code <p>} tag
 * between the literals.
 * <p>
 * <p>This method must be called after {@link #joinAdjacentLiteralsAndAdjacentWhitespace}, as it
 * assumes that adjacent whitespace tokens have already been joined.
 */
private static ImmutableList<Token> inferParagraphTags(List<Token> input) {
    ImmutableList.Builder<Token> output = ImmutableList.builder();

    for (PeekingIterator<Token> tokens = peekingIterator(input.iterator()); tokens.hasNext(); ) {
        if (tokens.peek().getType() == LITERAL) {
            output.add(tokens.next());

            if (tokens.peek().getType() == WHITESPACE
                    && hasMultipleNewlines(tokens.peek().getValue())) {
                output.add(tokens.next());

                if (tokens.peek().getType() == LITERAL) {
                    output.add(new Token(PARAGRAPH_OPEN_TAG, "<p>"));
                }
            }
        } else {
            // TODO(cpovirk): Or just `continue` from the <p> case and move this out of the `else`?
            output.add(tokens.next());
        }
    }

    return output.build();

/*
 * Note: We do not want to insert <p> tags inside <pre>. Fortunately, the formatter gets that
 * right without special effort on our part. The reason: Line breaks inside a <pre> section are
 * of type FORCED_NEWLINE rather than WHITESPACE.
 */
}

public PlainSortedMergeIterator(Iterable<? extends KeyIterable<TKey, TRow>> iterables, final Comparator<? super TRow> itemComparator) {

        Comparator<PeekingIterator<TRow>> heapComparator = new Comparator<PeekingIterator<TRow>>() {
            @Override
            public int compare(PeekingIterator<TRow> o1, PeekingIterator<TRow> o2) {
                return itemComparator.compare(o1.peek(), o2.peek());
            }
        };
        queue = new PriorityQueue<>(2, heapComparator);
        addIterators(iterables);
    }

private void skipToNextColumn(Cell cell, PeekingIterator<Map.Entry<Cell, Optional<Cell>>> iter) {
    while (iter.hasNext()
            && CellUtil.matchingFamily(iter.peek().getKey(), cell)
            && CellUtil.matchingQualifier(iter.peek().getKey(), cell)) {
        iter.next();
    }
}

/**
    * Return the on-disk PZX format data for the supplied PulseList
 * @param pulseList the PulseList to encode into the disk representation
    * @return the byte[] with the PZX disk format data
     */
public static byte[] getPZXBlockDiskRepresentation(PulseList pulseList) {
    // iterate through the pulse array doing a run length encoding of the number of repeated values
       PeekingIterator<Long> iterator = Iterators.peekingIterator(pulseList.getPulseLengths().iterator());
       int count;
       // We will probably have a similar number of bytes output as source pulses * 2 16 bit values
       ArrayList<Byte> output = new ArrayList<>(pulseList.getPulseLengths().size()*4);

       // The pulse level is low at start of the block by default. However initial
       // pulse of zero duration may be easily used to make it high.
       if( pulseList.getFirstPulseLevel() == 1 ) {
           PZXEncodeUtils.addBytesFor(0, 1, output);
       }

       // RLE the pulses found in the block for encoding
       while(iterator.hasNext()) {
           long pulse = iterator.next();
           count = 1;
           while(iterator.hasNext() && iterator.peek() == pulse) { 
               iterator.next();
               count += 1;
           }

           // Write the desired output bytes to the output list
           PZXEncodeUtils.addBytesFor(pulse, count, output);
       }

       return addPZXBlockHeader("PULS", output);
}