@ Override public void setMaxActiveMessages ( int maxActiveMessages ) { if ( TraceComponent . isAnyTracingEnabled ( ) && tc . isEntryEnabled ( ) ) SibTr . entry ( tc , "setMaxActiveMessages" , Integer . valueOf ( maxActiveMessages ) ) ; synchronized ( _asynchConsumerBusyLock ) { this . lock ( ) ; try { synchronized ( _maxActiveMessageLock ) { // If the new value is > previous value and // check if consumer is suspended and resume consumer. // But only do this if the message count is less than the curent active count if ( maxActiveMessages > _maxActiveMessages && maxActiveMessages < _currentActiveMessages ) { // If the consumer was suspended - reenable it if ( _consumerSuspended ) { // If this is part of an asynch callback we don't need to try and kick // off a new thread as the nextLocked will return us another message on // exit of the consumeMessages() method. if ( _runningAsynchConsumer ) { _consumerSuspended = false ; _suspendFlags &= ~ DispatchableConsumerPoint . SUSPEND_FLAG_ACTIVE_MSGS ; } // Otherwise perform a full resume else { resumeConsumer ( DispatchableConsumerPoint . SUSPEND_FLAG_ACTIVE_MSGS ) ; } } } else if ( maxActiveMessages <= _currentActiveMessages && ! _consumerSuspended ) { // If the maxActiveMessages has been set to something lower than the current active message // count, then suspend the consumer until the messages have been processed _consumerSuspended = true ; _suspendFlags |= DispatchableConsumerPoint . SUSPEND_FLAG_ACTIVE_MSGS ; } _maxActiveMessages = maxActiveMessages ; } // active message lock } // this lock finally { this . unlock ( ) ; } } // async lock if ( TraceComponent . isAnyTracingEnabled ( ) && tc . isEntryEnabled ( ) ) SibTr . exit ( tc , "setMaxActiveMessages" ) ; }
Update the max active messages field