public InputStream acquire ( UnderFileSystem ufs , String path , long fileId , OpenOptions openOptions , boolean reuse ) throws IOException { if ( ! ufs . isSeekable ( ) || ! CACHE_ENABLED ) { // not able to cache, always return a new input stream return ufs . openExistingFile ( path , openOptions ) ; } // explicit cache cleanup mUnderFileInputStreamCache . cleanUp ( ) ; UfsInputStreamIdSet resources ; synchronized ( mFileIdToInputStreamIds ) { if ( mFileIdToInputStreamIds . containsKey ( fileId ) ) { resources = mFileIdToInputStreamIds . get ( fileId ) ; } else { resources = new UfsInputStreamIdSet ( ) ; mFileIdToInputStreamIds . put ( fileId , resources ) ; } } synchronized ( resources ) { long nextId = UNAVAILABLE_RESOURCE_ID ; CachedSeekableInputStream inputStream = null ; if ( reuse ) { // find the next available input stream from the cache for ( long id : resources . availableIds ( ) ) { inputStream = mUnderFileInputStreamCache . getIfPresent ( id ) ; if ( inputStream != null ) { nextId = id ; LOG . debug ( "Reused the under file input stream resource of {}" , nextId ) ; // for the cached ufs instream, seek to the requested position inputStream . seek ( openOptions . getOffset ( ) ) ; break ; } } } // no cached input stream is available, open a new one if ( nextId == UNAVAILABLE_RESOURCE_ID ) { nextId = IdUtils . getRandomNonNegativeLong ( ) ; final long newId = nextId ; try { inputStream = mUnderFileInputStreamCache . get ( nextId , ( ) -> { SeekableUnderFileInputStream ufsStream = ( SeekableUnderFileInputStream ) ufs . openExistingFile ( path , OpenOptions . defaults ( ) . setOffset ( openOptions . getOffset ( ) ) ) ; LOG . debug ( "Created the under file input stream resource of {}" , newId ) ; return new CachedSeekableInputStream ( ufsStream , newId , fileId , path ) ; } ) ; } catch ( ExecutionException e ) { LOG . warn ( "Failed to create a new cached ufs instream of file id {} and path {}" , fileId , path ) ; // fall back to a ufs creation. return ufs . openExistingFile ( path , OpenOptions . defaults ( ) . setOffset ( openOptions . getOffset ( ) ) ) ; } } // mark the input stream id as acquired resources . acquire ( nextId ) ; return inputStream ; } }
Acquires an input stream . For seekable input streams if there is an available input stream in the cache and reuse mode is specified reuse it and repositions the offset otherwise the manager opens a new input stream .