com.google.common.collect.PeekingIterator#peek ( )源码实例Demo

/** Compute how many slots in a slab will be used to do a default (new slab) allocation, and how many bytes it will consume
 *
 * @param slabSlotsUsed - number of available slots in a slab that have been used prior to this allocation
 * @param slabBytesUsed - number of bytes in a slab that have been used prior to this allocation
 * @param eventSizes - list of the size in bytes of all events that we want to insert into the slab
 *
 * @return a pair of integers, the left value is the number of slots that will be used by this allocation and the
 *         right value is the numb er of bytes that will be used by this allocation
 */
static Pair<Integer, Integer> defaultAllocationCount(int slabSlotsUsed, int slabBytesUsed, PeekingIterator<Integer> eventSizes) {
    int slabTotalSlotCount = slabSlotsUsed;
    int allocationSlotCount = 0;
    int slabTotalBytesUsed = slabBytesUsed;
    int allocationBytes = 0;
    while (eventSizes.hasNext()) {
        checkArgument(eventSizes.peek() <= Constants.MAX_EVENT_SIZE_IN_BYTES, "Event size (" + eventSizes.peek() + ") is greater than the maximum allowed (" + Constants.MAX_EVENT_SIZE_IN_BYTES + ") event size");
        if (slabTotalSlotCount + 1 <= Constants.MAX_SLAB_SIZE && slabTotalBytesUsed + eventSizes.peek() <= Constants.MAX_SLAB_SIZE_IN_BYTES) {
            slabTotalSlotCount++;
            allocationSlotCount++;
            int eventSize = eventSizes.next();
            slabTotalBytesUsed += eventSize;
            allocationBytes += eventSize;
        } else {
            break;
        }
    }
    return new ImmutablePair<>(allocationSlotCount, allocationBytes);
}
 

/**
 * 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;
}
 

/**
 * 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();
}
 

ListenableFuture<?> spill()
{
    if (spillInProgress.isPresent()) {
        // Spill can be triggered first in SpillablePagesToPagesIndexes#process(..) and then by Driver (via WindowOperator#startMemoryRevoke)
        return spillInProgress.get();
    }

    if (localRevocableMemoryContext.getBytes() == 0) {
        // This must be stale revoke request
        spillInProgress = Optional.of(Futures.immediateFuture(null));
        return spillInProgress.get();
    }

    if (spiller.isEmpty()) {
        spiller = Optional.of(spillerFactory.create(
                sourceTypes,
                operatorContext.getSpillContext(),
                operatorContext.newAggregateSystemMemoryContext()));
    }

    verify(inMemoryPagesIndexWithHashStrategies.pagesIndex.getPositionCount() > 0);
    sortPagesIndexIfNecessary(inMemoryPagesIndexWithHashStrategies, orderChannels, ordering);
    PeekingIterator<Page> sortedPages = peekingIterator(inMemoryPagesIndexWithHashStrategies.pagesIndex.getSortedPages());
    Page anyPage = sortedPages.peek();
    verify(anyPage.getPositionCount() != 0, "PagesIndex.getSortedPages returned an empty page");
    currentSpillGroupRowPage = Optional.of(anyPage.getSingleValuePage(/* any */0));
    spillInProgress = Optional.of(spiller.get().spill(sortedPages));

    return spillInProgress.get();
}
 

public static <T> List<LocalProperty<T>> stripLeadingConstants(List<? extends LocalProperty<T>> properties)
{
    PeekingIterator<? extends LocalProperty<T>> iterator = peekingIterator(properties.iterator());
    while (iterator.hasNext() && iterator.peek() instanceof ConstantProperty) {
        iterator.next();
    }
    return ImmutableList.copyOf(iterator);
}
 

public static <T> Set<T> extractLeadingConstants(List<? extends LocalProperty<T>> properties)
{
    ImmutableSet.Builder<T> builder = ImmutableSet.builder();
    PeekingIterator<? extends LocalProperty<T>> iterator = peekingIterator(properties.iterator());
    while (iterator.hasNext() && iterator.peek() instanceof ConstantProperty) {
        builder.add(((ConstantProperty<T>) iterator.next()).getColumn());
    }
    return builder.build();
}
 

private void advance() {
	VideoRecordingSize latestTs = new VideoRecordingSize(IrisUUID.minTimeUUID(), false);
	Iterator<VideoRecordingSize> nextIt = null; 
	// find the largest value
	for(PeekingIterator<VideoRecordingSize> it: iterators) {
		if(!it.hasNext()) {
			continue;
		}
		
		if(currentId != null && (currentId.equals(it.peek()) || (currentId.getRecordingId().equals(it.peek().getRecordingId()) && !it.peek().isFavorite()))) {
			// throw away duplicate entry when the next record's recordingId is the same as currentId and not favorite
			it.next();
			if(!it.hasNext()) {
				continue;
			}
		}
		int comp = compare(latestTs.getRecordingId(), it.peek().getRecordingId());
		if(comp > 0){
			latestTs = it.peek(); //Set it because recordingId is larger 
			nextIt = it;
		}else if(comp == 0) {
			if(latestTs.isFavorite() || !it.peek().isFavorite()) {
				it.next(); //skip
			}else {
				latestTs = it.peek(); //Set it because the next one is favorite and the current on is not 
				nextIt = it;
			}
		}
	}
	if(nextIt == null) {
		currentId = null;
	}
	else {
		currentId = nextIt.next();
	}
}
 

private void advance() {
	currentId = null;
	
	UUID nextId = null;
	int matches = 0;
	while(matches < iterators.size()) {
		matches = 0;
		for(PeekingIterator<UUID> it: iterators) {
			while(nextId != null && it.hasNext() && compare(nextId, it.peek()) > 0) {
				// fast forward to where the current id is
				it.next();
			}
			if(!it.hasNext()) {
				// since its an intersection if any iterator is done, the whole thing is done
				return;
			}
			else if(nextId == null || it.peek().equals(nextId)) {
				// advance the iterator if it matches the current id
				nextId = it.next();
				matches++;
			}
			else if(nextId != null && compare(nextId, it.peek()) < 0) {
				// if this iterator is farther along then the others, reset nextId and start the loop over
				nextId = it.peek();
				break;
			}
		}
	}
	currentId = nextId;
}
 

/**
 * Produce an Iterator which merges the underlying SortedSets in a sorted fashion,
 * de-duplicating entries based on their equality-semantics.
 *
 * <p>The iterated order of values which are not equal but are equivalent according to the
 * Comparator is unspecified, but stable for a given instance of MergedSortedSet.
 *
 * <p>The return Iterator is not threadsafe.
 */
@Override
public Iterator<T> iterator() {
  PeekingIterator<T> left = Iterators.peekingIterator(a.iterator());
  PeekingIterator<T> right = Iterators.peekingIterator(b.iterator());
  return new Iterator<T>() {

    @Override
    public boolean hasNext() {
      return left.hasNext() || right.hasNext();
    }

    @Override
    public T next() {
      if (!left.hasNext()) {
        return right.next();
      } else if (!right.hasNext()) {
        return left.next();
      } else {
        T lval = left.peek();
        T rval = right.peek();
        if (lval.equals(rval)) {
          right.next();
          return left.next();
        } else {
          return comparator.compare(lval, rval) > 0 ? right.next() : left.next();
        }
      }
    }
  };
}
 

private void advance() {
	currentId = null;
	
	UUID nextId = null;
	int matches = 0;
	while(matches < iterators.size()) {
		matches = 0;
		for(PeekingIterator<UUID> it: iterators) {
			while(nextId != null && it.hasNext() && compare(nextId, it.peek()) > 0) {
				// fast forward to where the current id is
				it.next();
			}
			if(!it.hasNext()) {
				// since its an intersection if any iterator is done, the whole thing is done
				return;
			}
			else if(nextId == null || it.peek().equals(nextId)) {
				// advance the iterator if it matches the current id
				nextId = it.next();
				matches++;
			}
			else if(nextId != null && compare(nextId, it.peek()) < 0) {
				// if this iterator is farther along then the others, reset nextId and start the loop over
				nextId = it.peek();
				break;
			}
		}
	}
	currentId = nextId;
}
 

private static <T> Iterator<T> streamingIterator(Iterator<T> iterator) {
    // Force the calculation of at least the first item in the iterator so that, if an exception occurs, we find
    // out before writing the HTTP response code & headers.  Otherwise we will at best report a 500 error instead
    // of applying Jersey exception mappings and maybe returning a 400 error etc.
    PeekingIterator<T> peekingIterator = Iterators.peekingIterator(iterator);
    if (peekingIterator.hasNext()) {
        peekingIterator.peek();
    }

    return new LoggingIterator<>(peekingIterator, _log);
}
 

private static <T> Iterator<T> streamingIterator(Iterator<T> iterator) {
    // Force the calculation of at least the first item in the iterator so that, if an exception occurs, we find
    // out before writing the HTTP response code & headers.  Otherwise we will at best report a 500 error instead
    // of applying Jersey exception mappings and maybe returning a 400 error etc.
    PeekingIterator<T> peekingIterator = Iterators.peekingIterator(iterator);
    if (peekingIterator.hasNext()) {
        peekingIterator.peek();
    }

    return new LoggingIterator<>(peekingIterator, _log);
}
 

private static <T> Iterator<T> streamingIterator(Iterator<T> iterator, BooleanParam debug) {
    // If debugging, sort the individual json objects so they're easier to understand in a browser
    iterator = optionallyOrdered(iterator, debug);

    // Force the calculation of at least the first item in the iterator so that, if an exception occurs, we find
    // out before writing the HTTP response code & headers.  Otherwise we will at best report a 500 error instead
    // of applying Jersey exception mappings and maybe returning a 400 error etc.
    PeekingIterator<T> peekingIterator = Iterators.peekingIterator(iterator);
    if (peekingIterator.hasNext()) {
        peekingIterator.peek();
    }

    return new LoggingIterator<>(peekingIterator, _log);
}
 

public FilteredJsonStreamingOutput(Iterator<T> iterator, long limit) {
    // Force the first item to be read before streaming results, so if there are any errors they are thrown before
    // writing any content
    PeekingIterator<T> peekingIterator = Iterators.peekingIterator(iterator);
    if (peekingIterator.hasNext()) {
        peekingIterator.peek();
    }

    _iterator = peekingIterator;
    _limit = Math.max(limit, 0);
}
 

private static <T> Iterator<T> streamingIterator(InputStream in, TypeReference<T> typeReference) {
    PeekingIterator<T> iter = new JsonStreamingArrayParser<>(in, typeReference);

    // Fetch the first element in the result stream immediately, while still wrapped by the Ostrich retry logic.
    // If we can't get the first element then Ostrich should retry immediately.  If we fail to get subsequent
    // elements then clients must catch JsonStreamingEOFException and deal with it themselves.  They are highly
    // encouraged to use the DataStoreStreaming class which handles the restart logic automatically.
    if (iter.hasNext()) {
        iter.peek();
    }

    return iter;
}
 

private LinkedList<ILeafNode> collectLeafsWithSameOffset(ILeafNode candidate, PeekingIterator<ILeafNode> iterator) {
	LinkedList<ILeafNode> sameOffset = Lists.newLinkedList();
	sameOffset.add(candidate);
	int offset = candidate.getTotalOffset();
	while(iterator.hasNext()) {
		ILeafNode peek = iterator.peek();
		if (peek.getTotalOffset() == offset) {
			sameOffset.add(peek);
			iterator.next();
		} else {
			break;
		}
	}
	return sameOffset;
}
 

/**
 * Gather {@link MonitoredItemNotification}s and send them using {@code service}, if present.
 *
 * @param iterator a {@link PeekingIterator} over the current {@link BaseMonitoredItem}s.
 * @param service  a {@link ServiceRequest}, if available.
 */
private void gatherAndSend(PeekingIterator<BaseMonitoredItem<?>> iterator,
                           Optional<ServiceRequest<PublishRequest, PublishResponse>> service) {

    if (service.isPresent()) {
        List<UaStructure> notifications = Lists.newArrayList();

        while (notifications.size() < maxNotificationsPerPublish && iterator.hasNext()) {
            BaseMonitoredItem<?> item = iterator.peek();

            boolean gatheredAllForItem = gather(item, notifications, maxNotificationsPerPublish);

            if (gatheredAllForItem && iterator.hasNext()) {
                iterator.next();
            }
        }

        moreNotifications = iterator.hasNext();

        sendNotifications(service.get(), notifications);

        if (moreNotifications) {
            gatherAndSend(iterator, Optional.ofNullable(publishQueue().poll()));
        }
    } else {
        if (moreNotifications) {
            publishQueue().addSubscription(this);
        }
    }
}
 

public List<TaggingTokenCluster> cluster() {
        
        List<TaggingTokenCluster> result = new ArrayList<>();

        PeekingIterator<LabeledTokensContainer> it = Iterators.peekingIterator(taggingTokenSynchronizer);
        if (!it.hasNext() || (it.peek() == null)) {
            return Collections.emptyList();
        }

        // a boolean is introduced to indicate the start of the sequence in the case the label
        // has no beginning indicator (e.g. I-)
        boolean begin = true;
        TaggingTokenCluster curCluster = new TaggingTokenCluster(it.peek().getTaggingLabel());
        BoundingBox curBox=null;
 
        
        
        while (it.hasNext()) {
            LabeledTokensContainer cont = it.next();
            BoundingBox b = BoundingBox.fromLayoutToken(cont.getLayoutTokens().get(0));
            if(!curCluster.concatTokens().isEmpty()){
                curBox = BoundingBox.fromLayoutToken(curCluster.concatTokens().get(0));
                if(b.distanceTo(curBox)>600){
                    curCluster = new TaggingTokenCluster(cont.getTaggingLabel());
                    result.add(curCluster);
                }
            }
            
            if (begin || cont.isBeginning() || cont.getTaggingLabel() != curCluster.getTaggingLabel()) {
                curCluster = new TaggingTokenCluster(cont.getTaggingLabel());
                result.add(curCluster);
            }
            
            //for table, seperate caption and content
            if(curCluster!=null){
                String tableStr = LayoutTokensUtil.normalizeText(curCluster.concatTokens());
                if(tableStr.matches(".*?(Table|TABLE) \\d+(:|\\.| [A-Z]).*?")){
//                if(tableStr.matches(".*?(Table|TABLE|Figure|FIGURE) \\d+(:|\\.).*?")){
                    if(toText(curCluster.getLastContainer().getLayoutTokens()).equalsIgnoreCase(". \n\n")){ 
                        curCluster = new TaggingTokenCluster(cont.getTaggingLabel());
                        result.add(curCluster);
                    }
                }
            }
                    
                    
            curCluster.addLabeledTokensContainer(cont);
            if (begin)
                begin = false;
        }

        return result;
    }
 

private void updateAll(String apiKey, Iterable<Update> updates, boolean facade, Set<String> tags) {
    // This method takes an Iterable instead of an Iterator so it can be retried (by Ostrich etc.) if it fails.

    // If just one update, use the slightly more compact single record REST api.
    if (updates instanceof Collection && ((Collection) updates).size() == 1) {
        Update update = Iterables.getOnlyElement(updates);
        update(apiKey, update.getTable(), update.getKey(), update.getChangeId(), update.getDelta(), update.getAudit(),
                update.getConsistency(), facade, tags);
        return;
    }

    // Otherwise, use the streaming API to send multiple updates per HTTP request.  Break the updates into batches
    // such that this makes approximately one HTTP request per second.  The goal is to make requests big enough to
    // get the performance benefits of batching while being small enough that they show up with regularity in the
    // request logs--don't want an hour long POST that doesn't show up in the request log until the end of the hour.
    Iterator<Update> updatesIter = updates.iterator();
    for (long batchIdx = 0; updatesIter.hasNext(); batchIdx++) {
        PeekingIterator<Update> batchIter = TimeLimitedIterator.create(updatesIter, UPDATE_ALL_REQUEST_DURATION, 1);

        // Grab the first update, assume it's representative (but note it may not be) and copy some of its
        // attributes into the URL query parameters for the *sole* purpose of making the server request logs easier
        // to read.  The server ignores the query parameters--only the body of the POST actually matters.
        Update first = batchIter.peek();
        try {
            UriBuilder uriBuilder = _dataStore.clone()
                    .segment(facade ? "_facade" : "", "_stream")
                    .queryParam("batch", batchIdx)
                    .queryParam("table", first.getTable())
                    .queryParam("key", first.getKey())
                    .queryParam("audit", RisonHelper.asORison(first.getAudit()))
                    .queryParam("consistency", first.getConsistency());
            for(String tag : tags) {
                uriBuilder.queryParam("tag", tag);
            }
            _client.resource(uriBuilder.build())
                    .type(MediaType.APPLICATION_JSON_TYPE)
                    .header(ApiKeyRequest.AUTHENTICATION_HEADER, apiKey)
                    .post(batchIter);
        } catch (EmoClientException e) {
            throw convertException(e);
        }
    }
}
 

/**
 * Provide iterator via OpenCSV's CSVReader.
 * Provides a way to skip top rows by providing regex.(This is useful when CSV file comes with comments on top rows, but not in fixed size.
 * It also provides validation on schema by matching header names between property's schema and header name in CSV file.
 *
 * {@inheritDoc}
 * @see org.apache.gobblin.source.extractor.filebased.FileDownloader#downloadFile(java.lang.String)
 */
@SuppressWarnings("unchecked")
@Override
public Iterator<String[]> downloadFile(String file) throws IOException {

  log.info("Beginning to download file: " + file);
  final State state = fileBasedExtractor.workUnitState;

  CSVReader reader;
  try {
    if (state.contains(DELIMITER)) {
      String delimiterStr = state.getProp(DELIMITER).trim();
      Preconditions.checkArgument(delimiterStr.length() == 1, "Delimiter should be a character.");

      char delimiter = delimiterStr.charAt(0);
      log.info("Using " + delimiter + " as a delimiter.");

      reader = this.fileBasedExtractor.getCloser().register(
          new CSVReader(new InputStreamReader(
                            this.fileBasedExtractor.getFsHelper().getFileStream(file),
                            ConfigurationKeys.DEFAULT_CHARSET_ENCODING), delimiter));
    } else {
      reader = this.fileBasedExtractor.getCloser().register(
          new CSVReader(new InputStreamReader(
                            this.fileBasedExtractor.getFsHelper().getFileStream(file),
                            ConfigurationKeys.DEFAULT_CHARSET_ENCODING)));
    }

  } catch (FileBasedHelperException e) {
    throw new IOException(e);
  }

  PeekingIterator<String[]> iterator = Iterators.peekingIterator(reader.iterator());

  if (state.contains(SKIP_TOP_ROWS_REGEX)) {
    String regex = state.getProp(SKIP_TOP_ROWS_REGEX);
    log.info("Trying to skip with regex: " + regex);
    while (iterator.hasNext()) {
      String[] row = iterator.peek();
      if (row.length == 0) {
        break;
      }

      if (!row[0].matches(regex)) {
        break;
      }
      iterator.next();
    }
  }

  if (this.fileBasedExtractor.isShouldSkipFirstRecord() && iterator.hasNext()) {
    log.info("Skipping first record");
    iterator.next();
  }
  return iterator;
}