public void removeActiveMessages(int messageCount)
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.entry(this, tc, "removeActiveMessages", Integer.valueOf(messageCount));

    boolean resumeConsumers = false;
    
    if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
      SibTr.debug(tc, "maxActiveMessagePrepareReadLock.lock(): " + maxActiveMessagePrepareLock);
    
    // Take a read lock on the prepare to ensure the set isn't in the process
    // of suspending itself.
    maxActiveMessagePrepareReadLock.lock();
    
    try
    {
      // Lock down the active message count
      synchronized(maxActiveMessageLock)
      {
        // Decrement the count
        currentActiveMessages -= messageCount;

        // We should never go negative, if we do something has gone wrong - FFDC
        if(currentActiveMessages < 0)
        {
          SIErrorException e = new SIErrorException(
              nls.getFormattedMessage(
                "INTERNAL_MESSAGING_ERROR_CWSIP0001",
                new Object[] {
                  "com.ibm.ws.sib.processor.impl.JSConsumerSet.removeActiveMessages",
                  "1:677:1.16",
                  Integer.valueOf(preparedActiveMessages),
                  Integer.valueOf(currentActiveMessages),
                  Integer.valueOf(maxActiveMessages),
                  Boolean.valueOf(consumerSetSuspended)
                  },
                null));

            FFDCFilter.processException(
              e,
              "com.ibm.ws.sib.processor.impl.JSConsumerSet.removeActiveMessages",
              "1:688:1.16",
              this);

            SibTr.exception(tc, e);
            SibTr.error(tc, "INTERNAL_MESSAGING_ERROR_CWSIP0001",
              new Object[] {
                "com.ibm.ws.sib.processor.impl.JSConsumerSet.removeActiveMessages",
                "1:695:1.16" });
        }
        
        // If the set is currently suspended then we'll check to see if
        // we've fallen below the threashold
        if(consumerSetSuspended)
        {
          // The threashold that we resume a set is potentially a few messages
          // below the max, so we don't constantly switch from suspended to resumed
          if(currentActiveMessages == (resumeThreshold - messageCount))
          {
            // Reset the consumerSet flag to make any inquiring consumers aware
            // of the current state.
            consumerSetSuspended = false;
            
            // Now we need to resume any currently suspended consumers. We can't
            // do this while we hold the maxActiveMessageLock as the resume will
            // need to take the LCP lock for each consumer. Instead we release
            // this lock and perform the resume under the consumerList lock to
            // prevent the list changing. We're also reliant on each consumer being
            // resumed re-checking with us that we're still resumed (as we no
            // longer hold the maxActiveMessageLock at the time of the resume - so
            // some other consumer could have got in and caused us to suspend again)
            resumeConsumers = true;
          }
        }
        
        if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
          SibTr.debug(tc, "Active Messages: current: " + currentActiveMessages +
                            ", prepared: " + preparedActiveMessages +
                            ", maximum: " + maxActiveMessages +
                            " (suspended: " + consumerSetSuspended + ")");
      } // synchronized
    }
    finally
    {
      // Release the read lock
      maxActiveMessagePrepareReadLock.unlock();

      if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
        SibTr.debug(tc, "maxActiveMessagePrepareReadLock.unlock(): " + maxActiveMessagePrepareLock);
    }
    
    // Now we've release the lock, resume the consumers if needed
    if(resumeConsumers)
      resumeConsumers();
    
    if(TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.exit(this, tc, "removeActiveMessages");
  }
public void removeActiveMessages ( int messageCount ) { if ( TraceComponent . isAnyTracingEnabled ( ) && tc . isEntryEnabled ( ) ) SibTr . entry ( this , tc , "removeActiveMessages" , Integer . valueOf ( messageCount ) ) ; boolean resumeConsumers = false ; if ( TraceComponent . isAnyTracingEnabled ( ) && tc . isDebugEnabled ( ) ) SibTr . debug ( tc , "maxActiveMessagePrepareReadLock.lock(): " + maxActiveMessagePrepareLock ) ; maxActiveMessagePrepareReadLock . lock ( ) ; try { synchronized ( maxActiveMessageLock ) { currentActiveMessages -= messageCount ; if ( currentActiveMessages < 0 ) { SIErrorException e = new SIErrorException ( nls . getFormattedMessage ( "INTERNAL_MESSAGING_ERROR_CWSIP0001" , new Object [ ] { "com.ibm.ws.sib.processor.impl.JSConsumerSet.removeActiveMessages" , "1:677:1.16" , Integer . valueOf ( preparedActiveMessages ) , Integer . valueOf ( currentActiveMessages ) , Integer . valueOf ( maxActiveMessages ) , Boolean . valueOf ( consumerSetSuspended ) } , null ) ) ; FFDCFilter . processException ( e , "com.ibm.ws.sib.processor.impl.JSConsumerSet.removeActiveMessages" , "1:688:1.16" , this ) ; SibTr . exception ( tc , e ) ; SibTr . error ( tc , "INTERNAL_MESSAGING_ERROR_CWSIP0001" , new Object [ ] { "com.ibm.ws.sib.processor.impl.JSConsumerSet.removeActiveMessages" , "1:695:1.16" } ) ; } if ( consumerSetSuspended ) { if ( currentActiveMessages == ( resumeThreshold - messageCount ) ) { consumerSetSuspended = false ; resumeConsumers = true ; } } if ( TraceComponent . isAnyTracingEnabled ( ) && tc . isDebugEnabled ( ) ) SibTr . debug ( tc , "Active Messages: current: " + currentActiveMessages + ", prepared: " + preparedActiveMessages + ", maximum: " + maxActiveMessages + " (suspended: " + consumerSetSuspended + ")" ) ; } } finally { maxActiveMessagePrepareReadLock . unlock ( ) ; if ( TraceComponent . isAnyTracingEnabled ( ) && tc . isDebugEnabled ( ) ) SibTr . debug ( tc , "maxActiveMessagePrepareReadLock.unlock(): " + maxActiveMessagePrepareLock ) ; } if ( resumeConsumers ) resumeConsumers ( ) ; if ( TraceComponent . isAnyTracingEnabled ( ) && tc . isEntryEnabled ( ) ) SibTr . exit ( this , tc , "removeActiveMessages" ) ; }
Decrement the active message count

@param messageCount number of messages to remove from the count