@ Override public CompletableFuture < AttributeIndex > forSegment ( long streamSegmentId , Duration timeout ) { Exceptions . checkNotClosed ( this . closed . get ( ) , this ) ; SegmentMetadata sm = this . containerMetadata . getStreamSegmentMetadata ( streamSegmentId ) ; if ( sm . isDeleted ( ) ) { return Futures . failedFuture ( new StreamSegmentNotExistsException ( sm . getName ( ) ) ) ; } // Figure out if we already have this AttributeIndex cached. If not, we need to initialize it. CompletableFuture < AttributeIndex > result ; AtomicReference < SegmentAttributeBTreeIndex > toInitialize = new AtomicReference <> ( ) ; synchronized ( this . attributeIndices ) { result = this . attributeIndices . computeIfAbsent ( streamSegmentId , id -> { toInitialize . set ( new SegmentAttributeBTreeIndex ( sm , this . storage , this . cache , this . config , this . executor ) ) ; return new CompletableFuture <> ( ) ; } ) ; } if ( toInitialize . get ( ) == null ) { // We already have it cached - return its future (which should be already completed or will complete once // its initialization is done). return result ; } else { try { // Need to initialize the AttributeIndex and complete the future that we just registered. // If this fails, we must fail the Future that we previously registered and unregister any pointers to // this index. toInitialize . get ( ) . initialize ( timeout ) . thenRun ( ( ) -> this . cacheManager . register ( toInitialize . get ( ) ) ) . whenComplete ( ( r , ex ) -> { if ( ex == null ) { result . complete ( toInitialize . get ( ) ) ; } else { indexInitializationFailed ( streamSegmentId , result , ex ) ; } } ) ; } catch ( Throwable ex ) { if ( ! Exceptions . mustRethrow ( ex ) ) { indexInitializationFailed ( streamSegmentId , result , ex ) ; } throw ex ; } } return result ; }
region ContainerAttributeIndex Implementation