void process ( Iterator < Operation > operations ) throws DataCorruptionException { HashSet < Long > segmentIds = new HashSet <> ( ) ; while ( operations . hasNext ( ) ) { Operation op = operations . next ( ) ; process ( op ) ; if ( op instanceof SegmentOperation ) { // Record recent activity on stream segment, if applicable. This should be recorded for any kind // of Operation that touches a Segment, since when we issue 'triggerFutureReads' on the readIndex, // it should include 'sealed' StreamSegments too - any Future Reads waiting on that Offset will be cancelled. segmentIds . add ( ( ( SegmentOperation ) op ) . getStreamSegmentId ( ) ) ; } } if ( ! this . recoveryMode . get ( ) ) { // Trigger Future Reads on those segments which were touched by Appends or Seals. this . readIndex . triggerFutureReads ( segmentIds ) ; if ( this . commitSuccess != null ) { this . commitSuccess . run ( ) ; } } }
Processes the given operations and applies them to the ReadIndex and InMemory OperationLog .