question: 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 ; }
target:  Implements peek() or poll() based on whether claimTtl is null or non-null. 