void gc ( ) { if ( stopped ) { return ; } long expirationTime = System . currentTimeMillis ( ) - timeout ; for ( Iterator < Map . Entry < InputStreamKey , TimedStreamEntry > > iter = streamMap . entrySet ( ) . iterator ( ) ; iter . hasNext ( ) ; ) { if ( stopped ) { return ; } Map . Entry < InputStreamKey , TimedStreamEntry > entry = iter . next ( ) ; TimedStreamEntry timedStreamEntry = entry . getValue ( ) ; if ( timedStreamEntry . timestamp . get ( ) <= expirationTime ) { iter . remove ( ) ; InputStreamKey key = entry . getKey ( ) ; //noinspection SynchronizationOnLocalVariableOrMethodParameter synchronized ( timedStreamEntry ) { // ensure there's no race with a request that got a ref before we removed it closeStreamEntry ( timedStreamEntry , key . requestId , key . index ) ; } } } }
Close and remove expired streams . Package protected to allow unit tests to invoke it .