@ Override public Position get ( Position pos , List < Event < T > > list ) { EventBatch < T > b ; // Return null if the position is out of retention or in the indexed form. if ( pos . getOffset ( ) < getOrigin ( ) || pos . isIndexed ( ) ) { return null ; } // Get events from _batch b = _batch ; if ( b . getOrigin ( ) <= pos . getOffset ( ) ) { long newOffset = b . get ( pos . getOffset ( ) , list ) ; Clock clock = pos . getOffset ( ) < newOffset ? b . getClock ( newOffset - 1 ) : pos . getClock ( ) ; return new SimplePosition ( getId ( ) , newOffset , clock ) ; } // Get events from _lastBatch b = _lastBatch ; if ( b != null && b . getOrigin ( ) <= pos . getOffset ( ) ) { long newOffset = b . get ( pos . getOffset ( ) , list ) ; Clock clock = pos . getOffset ( ) < newOffset ? b . getClock ( newOffset - 1 ) : pos . getClock ( ) ; return new SimplePosition ( getId ( ) , newOffset , clock ) ; } // Get events from batches in retention int cnt = 0 ; Iterator < EventBatchCursor > iter = _retentionQueue . iterator ( ) ; while ( iter . hasNext ( ) ) { EventBatchCursor c = iter . next ( ) ; if ( c . getHeader ( ) . getOrigin ( ) <= pos . getOffset ( ) ) { byte [ ] dat = _store . get ( c . getLookup ( ) ) ; try { b = _eventBatchSerializer . deserialize ( dat ) ; long newOffset = b . get ( pos . getOffset ( ) , list ) ; if ( pos . getOffset ( ) < newOffset ) { Clock clock = b . getClock ( newOffset - 1 ) ; return new SimplePosition ( getId ( ) , newOffset , clock ) ; } } catch ( Exception e ) { _logger . warn ( "Ignored EventBatch: " + c . getHeader ( ) . getOrigin ( ) ) ; } } else { // early stop if ( cnt == 0 ) { break ; } } cnt ++ ; } return null ; }
Gets a number of events starting from a given position in the Retention . The number of events is determined internally by the Retention and it is up to the batch size .