void cleanupQueue ( SubscriptionImpl subscription , Queue < Object > queue ) { final Throwable cause = ClosedPublisherException . get ( ) ; for ( ; ; ) { final Object e = queue . poll ( ) ; if ( e == null ) { break ; } try { if ( e instanceof CloseEvent ) { notifySubscriberOfCloseEvent ( subscription , ( CloseEvent ) e ) ; continue ; } if ( e instanceof CompletableFuture ) { ( ( CompletableFuture < ? > ) e ) . completeExceptionally ( cause ) ; } @ SuppressWarnings ( "unchecked" ) final T obj = ( T ) e ; onRemoval ( obj ) ; } finally { ReferenceCountUtil . safeRelease ( e ) ; } } }
Helper method for the common case of cleaning up all elements in a queue when shutting down the stream .