private void triggerFutureReads ( Collection < FutureReadResultEntry > futureReads ) { for ( FutureReadResultEntry r : futureReads ) { ReadResultEntry entry = getSingleReadResultEntry ( r . getStreamSegmentOffset ( ) , r . getRequestedReadLength ( ) ) ; assert entry != null : "Serving a StorageReadResultEntry with a null result" ; assert ! ( entry instanceof FutureReadResultEntry ) : "Serving a FutureReadResultEntry with another FutureReadResultEntry." ; log . trace ( "{}: triggerFutureReads (Offset = {}, Type = {})." , this . traceObjectId , r . getStreamSegmentOffset ( ) , entry . getType ( ) ) ; if ( entry . getType ( ) == ReadResultEntryType . EndOfStreamSegment ) { // We have attempted to read beyond the end of the stream. Fail the read request with the appropriate message. r . fail ( new StreamSegmentSealedException ( String . format ( "StreamSegment has been sealed at offset %d. There can be no more reads beyond this offset." , this . metadata . getLength ( ) ) ) ) ; } else { if ( ! entry . getContent ( ) . isDone ( ) ) { // Normally, all Future Reads are served from Cache, since they reflect data that has just been appended. // However, it's possible that after recovery, we get a read for some data that we do not have in the // cache (but it's not a tail read) - this data exists in Storage but our StorageLength has not yet been // updated. As such, the only solution we have is to return a FutureRead which will be satisfied when // the Writer updates the StorageLength (and trigger future reads). In that scenario, entry we get // will likely not be auto-fetched, so we need to request the content. entry . requestContent ( this . config . getStorageReadDefaultTimeout ( ) ) ; } CompletableFuture < ReadResultEntryContents > entryContent = entry . getContent ( ) ; entryContent . thenAccept ( r :: complete ) ; Futures . exceptionListener ( entryContent , r :: fail ) ; } } }
Triggers all the Future Reads in the given collection .