private List<Event> peekOrPoll(String subscription,@Nullable Duration claimTtl,int limit){
  List<Item> items=Lists.newArrayList();
  Map<Coordinate,Item> uniqueItems=Maps.newHashMap();
  Map<Coordinate,Integer> eventOrder=Maps.newHashMap();
  boolean repeatable=claimTtl != null && claimTtl.getMillis() > 0;
  Stopwatch stopwatch=Stopwatch.createStarted(_ticker);
  int padding=0;
  do {
    int remaining=limit - items.size();
    if (remaining == 0) {
      break;
    }
    ConsolidatingEventSink sink=new ConsolidatingEventSink(remaining + padding);
    boolean more=(claimTtl == null) ? _eventStore.peek(subscription,sink) : _eventStore.poll(subscription,claimTtl,sink);
    Map<Coordinate,EventList> rawEvents=sink.getEvents();
    if (rawEvents.isEmpty()) {
      break;
    }
    List<String> eventIdsToDiscard=Lists.newArrayList();
    List<String> recentUnknownEventIds=Lists.newArrayList();
    List<String> eventIdsToUnclaim=Lists.newArrayList();
    DataProvider.AnnotatedGet annotatedGet=_dataProvider.prepareGetAnnotated(ReadConsistency.STRONG);
    for (    Map.Entry<Coordinate,EventList> entry : rawEvents.entrySet()) {
      Coordinate coord=entry.getKey();
      try {
        annotatedGet.add(coord.getTable(),coord.getId());
      }
 catch (      UnknownTableException e) {
        EventList list=entry.getValue();
        for (        Pair<String,UUID> pair : list.getEventAndChangeIds()) {
          eventIdsToDiscard.add(pair.first());
        }
        _discardedMeter.mark(list.size());
      }
      if (!eventOrder.containsKey(coord)) {
        eventOrder.put(coord,eventOrder.size());
      }
    }
    Iterator<DataProvider.AnnotatedContent> readResultIter=annotatedGet.execute();
    while (readResultIter.hasNext()) {
      DataProvider.AnnotatedContent readResult=readResultIter.next();
      Map<String,Object> content=readResult.getContent();
      Coordinate coord=Coordinate.fromJson(content);
      EventList eventList=rawEvents.get(coord);
      List<List<String>> tags=eventList.getTags();
      for (      Pair<String,UUID> eventData : eventList.getEventAndChangeIds()) {
        String eventId=eventData.first();
        UUID changeId=eventData.second();
        if (readResult.isChangeDeltaPending(changeId)) {
          if (isRecent(changeId)) {
            recentUnknownEventIds.add(eventId);
            _recentUnknownMeter.mark();
          }
 else {
            _staleUnknownMeter.mark();
          }
          continue;
        }
        if (readResult.isChangeDeltaRedundant(changeId)) {
          eventIdsToDiscard.add(eventId);
          _redundantMeter.mark();
          continue;
        }
        Item previousItem=uniqueItems.get(coord);
        if (previousItem != null && previousItem.consolidateWith(eventId,content,tags)) {
          _consolidatedMeter.mark();
          continue;
        }
        if (items.size() == limit) {
          eventIdsToUnclaim.add(eventId);
          continue;
        }
        Item item=new Item(eventId,eventOrder.get(coord),content,tags);
        items.add(item);
        uniqueItems.put(coord,item);
      }
    }
    if (!eventIdsToUnclaim.isEmpty()) {
      _eventStore.renew(subscription,eventIdsToUnclaim,Duration.ZERO,false);
    }
    if (!recentUnknownEventIds.isEmpty()) {
      _eventStore.renew(subscription,recentUnknownEventIds,RECENT_UNKNOWN_RETRY,false);
    }
    if (!eventIdsToDiscard.isEmpty()) {
      _eventStore.delete(subscription,eventIdsToDiscard,true);
    }
    if (!more) {
      break;
    }
    padding=10;
  }
 while (repeatable && stopwatch.elapsed(TimeUnit.MILLISECONDS) < MAX_POLL_TIME.getMillis());
  Collections.sort(items);
  List<Event> events=Lists.newArrayListWithCapacity(items.size());
  for (  Item item : items) {
    events.add(item.toEvent());
  }
  return events;
}

private List < Event > peekOrPoll ( String subscription , @ Nullable Duration claimTtl , int limit ) { List < Item > items = Lists . newArrayList ( ) ; Map < Coordinate , Item > uniqueItems = Maps . newHashMap ( ) ; Map < Coordinate , Integer > eventOrder = Maps . newHashMap ( ) ; boolean repeatable = claimTtl != null && claimTtl . getMillis ( ) > 0 ; Stopwatch stopwatch = Stopwatch . createStarted ( _ticker ) ; int padding = 0 ; do { int remaining = limit - items . size ( ) ; if ( remaining == 0 ) { break ; } ConsolidatingEventSink sink = new ConsolidatingEventSink ( remaining + padding ) ; boolean more = ( claimTtl == null ) ? _eventStore . peek ( subscription , sink ) : _eventStore . poll ( subscription , claimTtl , sink ) ; Map < Coordinate , EventList > rawEvents = sink . getEvents ( ) ; if ( rawEvents . isEmpty ( ) ) { break ; } List < String > eventIdsToDiscard = Lists . newArrayList ( ) ; List < String > recentUnknownEventIds = Lists . newArrayList ( ) ; List < String > eventIdsToUnclaim = Lists . newArrayList ( ) ; DataProvider . AnnotatedGet annotatedGet = _dataProvider . prepareGetAnnotated ( ReadConsistency . STRONG ) ; for ( Map . Entry < Coordinate , EventList > entry : rawEvents . entrySet ( ) ) { Coordinate coord = entry . getKey ( ) ; try { annotatedGet . add ( coord . getTable ( ) , coord . getId ( ) ) ; } catch ( UnknownTableException e ) { EventList list = entry . getValue ( ) ; for ( Pair < String , UUID > pair : list . getEventAndChangeIds ( ) ) { eventIdsToDiscard . add ( pair . first ( ) ) ; } _discardedMeter . mark ( list . size ( ) ) ; } if ( ! eventOrder . containsKey ( coord ) ) { eventOrder . put ( coord , eventOrder . size ( ) ) ; } } Iterator < DataProvider . AnnotatedContent > readResultIter = annotatedGet . execute ( ) ; while ( readResultIter . hasNext ( ) ) { DataProvider . AnnotatedContent readResult = readResultIter . next ( ) ; Map < String , Object > content = readResult . getContent ( ) ; Coordinate coord = Coordinate . fromJson ( content ) ; EventList eventList = rawEvents . get ( coord ) ; List < List < String > > tags = eventList . getTags ( ) ; for ( Pair < String , UUID > eventData : eventList . getEventAndChangeIds ( ) ) { String eventId = eventData . first ( ) ; UUID changeId = eventData . second ( ) ; if ( readResult . isChangeDeltaPending ( changeId ) ) { if ( isRecent ( changeId ) ) { recentUnknownEventIds . add ( eventId ) ; _recentUnknownMeter . mark ( ) ; } else { _staleUnknownMeter . mark ( ) ; } continue ; } if ( readResult . isChangeDeltaRedundant ( changeId ) ) { eventIdsToDiscard . add ( eventId ) ; _redundantMeter . mark ( ) ; continue ; } Item previousItem = uniqueItems . get ( coord ) ; if ( previousItem != null && previousItem . consolidateWith ( eventId , content , tags ) ) { _consolidatedMeter . mark ( ) ; continue ; } if ( items . size ( ) == limit ) { eventIdsToUnclaim . add ( eventId ) ; continue ; } Item item = new Item ( eventId , eventOrder . get ( coord ) , content , tags ) ; items . add ( item ) ; uniqueItems . put ( coord , item ) ; } } if ( ! eventIdsToUnclaim . isEmpty ( ) ) { _eventStore . renew ( subscription , eventIdsToUnclaim , Duration . ZERO , false ) ; } if ( ! recentUnknownEventIds . isEmpty ( ) ) { _eventStore . renew ( subscription , recentUnknownEventIds , RECENT_UNKNOWN_RETRY , false ) ; } if ( ! eventIdsToDiscard . isEmpty ( ) ) { _eventStore . delete ( subscription , eventIdsToDiscard , true ) ; } if ( ! more ) { break ; } padding = 10 ; } while ( repeatable && stopwatch . elapsed ( TimeUnit . MILLISECONDS ) < MAX_POLL_TIME . getMillis ( ) ) ; Collections . sort ( items ) ; List < Event > events = Lists . newArrayListWithCapacity ( items . size ( ) ) ; for ( Item item : items ) { events . add ( item . toEvent ( ) ) ; } return events ; }
implements peek ( ) or poll ( ) based on whether claimttl is null or non - null .
