public WindowFuture offer ( K key , R request , long offerTimeoutMillis , long expireTimeoutMillis , boolean callerWaitingHint ) throws DuplicateKeyException , OfferTimeoutException , PendingOfferAbortedException , InterruptedException { if ( offerTimeoutMillis < 0 ) { throw new IllegalArgumentException ( "offerTimeoutMillis must be >= 0 [actual=" + offerTimeoutMillis + "]" ) ; } // does this key already exist? if ( this . futures . containsKey ( key ) ) { throw new DuplicateKeyException ( "The key [" + key + "] already exists in the window" ) ; } long offerTimestamp = System . currentTimeMillis ( ) ; this . lock . lockInterruptibly ( ) ; try { // does enough room exist in the "window" for another pending request? // NOTE: wait for room up to the offerTimeoutMillis // NOTE: multiple signals may be received that will need to be ignored while ( getFreeSize ( ) <= 0 ) { // check if there time remaining to wait long currentOfferTime = System . currentTimeMillis ( ) - offerTimestamp ; if ( currentOfferTime >= offerTimeoutMillis ) { throw new OfferTimeoutException ( "Unable to accept offer within [" + offerTimeoutMillis + " ms] (window full)" ) ; } // check if slow waiting was canceled (terminate early) if ( this . pendingOffersAborted . get ( ) ) { throw new PendingOfferAbortedException ( "Pending offer aborted (by an explicit call to abortPendingOffers())" ) ; } // calculate the amount of timeout remaining long remainingOfferTime = offerTimeoutMillis - currentOfferTime ; try { // await for a new signal for this max amount of time this . beginPendingOffer ( ) ; this . completedCondition . await ( remainingOfferTime , TimeUnit . MILLISECONDS ) ; } finally { boolean abortPendingOffer = this . endPendingOffer ( ) ; if ( abortPendingOffer ) { throw new PendingOfferAbortedException ( "Pending offer aborted (by an explicit call to abortPendingOffers())" ) ; } } } long acceptTimestamp = System . currentTimeMillis ( ) ; long expireTimestamp = ( expireTimeoutMillis > 0 ? ( acceptTimestamp + expireTimeoutMillis ) : - 1 ) ; int callerStateHint = ( callerWaitingHint ? WindowFuture . CALLER_WAITING : WindowFuture . CALLER_NOT_WAITING ) ; DefaultWindowFuture < K , R , P > future = new DefaultWindowFuture < K , R , P > ( this , lock , completedCondition , key , request , callerStateHint , offerTimeoutMillis , ( futures . size ( ) + 1 ) , offerTimestamp , acceptTimestamp , expireTimestamp ) ; this . futures . put ( key , future ) ; return future ; } finally { this . lock . unlock ( ) ; } }
Offers a request for acceptance waiting for the specified amount of time in case it could not immediately accepted .