NatsMessage accumulate ( long maxSize , long maxMessages , Duration timeout ) throws InterruptedException { if ( ! this . singleThreadedReader ) { throw new IllegalStateException ( "Accumulate is only supported in single reader mode." ) ; } if ( ! this . isRunning ( ) ) { return null ; } NatsMessage msg = this . queue . poll ( ) ; if ( msg == null ) { msg = waitForTimeout ( timeout ) ; if ( ! this . isRunning ( ) || ( msg == null ) ) { return null ; } } long size = msg . getSizeInBytes ( ) ; if ( maxMessages <= 1 || size >= maxSize ) { this . sizeInBytes . addAndGet ( - size ) ; this . length . decrementAndGet ( ) ; signalIfNotEmpty ( ) ; return msg ; } long count = 1 ; NatsMessage cursor = msg ; while ( cursor != null ) { NatsMessage next = this . queue . peek ( ) ; if ( next != null ) { long s = next . getSizeInBytes ( ) ; if ( maxSize < 0 || ( size + s ) < maxSize ) { // keep going size += s ; count ++ ; cursor . next = this . queue . poll ( ) ; cursor = cursor . next ; if ( count == maxMessages ) { break ; } } else { // One more is too far break ; } } else { // Didn't meet max condition break ; } } this . sizeInBytes . addAndGet ( - size ) ; this . length . addAndGet ( - count ) ; signalIfNotEmpty ( ) ; return msg ; }
readers are present you could get out of order message delivery .