public void addMissingRequests ( ) { int missed = 1 ; long toRequest = 0L ; do { long localQueued = queued . getAndSet ( 0l ) ; Subscription localSub = next . getAndSet ( null ) ; long missedOutput = produced . get ( ) ; Subscription localActive = active . get ( ) ; long reqs = requested . get ( ) + localQueued ; if ( reqs < 0 || toRequest < 0 ) { processAll = true ; if ( localSub != null ) localSub . request ( Long . MAX_VALUE ) ; if ( localActive != null ) localActive . request ( Long . MAX_VALUE ) ; return ; } requested . set ( reqs ) ; if ( localSub != null ) { active . set ( localSub ) ; toRequest += reqs ; } else if ( localQueued != 0 && localActive != null ) { toRequest += reqs ; } missed = wip . accumulateAndGet ( missed , ( a , b ) -> a - b ) ; } while ( missed != 0 ) ; if ( toRequest > 0 ) active . get ( ) . request ( toRequest ) ; }
transfer demand from previous to next