void postMessage(List<JetstreamMessage> msgs,DispatchQueueStats stats) throws MessageServiceException {
  m_msgRcvCounter.addAndGet(msgs.size());
  if ((monitorUpstreamQueueAndPauseTraffic() == UpstreamQueueState.FULL) && (m_paused.get())) {
    if (!m_msgProcessor.hasAvailableCapacity(m_twentyPercentCapacity)) {
      m_totalMsgsDropped.increment();
      return;
    }
  }
  List<Runnable> requests=new ArrayList<Runnable>(msgs.size());
  for (int i=0, t=msgs.size(); i < t; i++) {
    JetstreamMessage tm=msgs.get(i);
    if (tm.getTopic() == null) {
      m_totalMsgsDropped.increment();
      if (LOGGER.isDebugEnabled()) {
        LOGGER.debug("Topic is not present in incoming message");
      }
      continue;
    }
    MessageServiceRequest msr=new MessageServiceRequest(tm);
    msr.setPriority(tm.getPriority());
    msr.setSequenceid(tm.getSequenceId());
    if (msr.getPriority() == JetstreamMessage.INTERNAL_MSG_PRIORITY) {
      if (!m_internalMsgProcessor.processRequest(msr)) {
        m_totalMsgsDropped.increment();
        throw new MessageServiceException(MessageServiceException.BUFFER_FULL,"Dispatch Queue Full");
      }
      if (m_msgsRcvdPerSec.addAndGet(1) < 0)       m_msgsRcvdPerSec.set(0);
      m_totalMsgsRcvd.increment();
    }
 else {
      requests.add(msr);
    }
  }
  if (!requests.isEmpty()) {
    int batchsize=requests.size();
    if (!m_msgProcessor.processBatch(requests)) {
      m_totalMsgsDropped.addAndGet(batchsize);
      throw new MessageServiceException(MessageServiceException.BUFFER_FULL,"High Priority Dispatch Queue Full - " + " Requested capacity = " + batchsize + " : available capacity = "+ m_msgProcessor.getAvailableCapacity());
    }
    m_avgMsgsRcvdPerSec.add(batchsize);
    m_totalMsgsRcvd.addAndGet(batchsize);
  }
  if (stats != null) {
    stats.setHighPriorityQueueDepth((int)m_msgProcessor.getPendingRequests());
    stats.setLowPriorityQueueDepth((int)m_msgProcessor.getPendingRequests());
    stats.setMaxQueueDepth((int)m_msgProcessor.getMaxQueueSz());
  }
}

void postMessage ( List < JetstreamMessage > msgs , DispatchQueueStats stats ) throws MessageServiceException { m_msgRcvCounter . addAndGet ( msgs . size ( ) ) ; if ( ( monitorUpstreamQueueAndPauseTraffic ( ) == UpstreamQueueState . FULL ) && ( m_paused . get ( ) ) ) { if ( ! m_msgProcessor . hasAvailableCapacity ( m_twentyPercentCapacity ) ) { m_totalMsgsDropped . increment ( ) ; return ; } } List < Runnable > requests = new ArrayList < Runnable > ( msgs . size ( ) ) ; for ( int i = 0 , t = msgs . size ( ) ; i < t ; i ++ ) { JetstreamMessage tm = msgs . get ( i ) ; if ( tm . getTopic ( ) == null ) { m_totalMsgsDropped . increment ( ) ; if ( LOGGER . isDebugEnabled ( ) ) { LOGGER . debug ( "Topic is not present in incoming message" ) ; } continue ; } MessageServiceRequest msr = new MessageServiceRequest ( tm ) ; msr . setPriority ( tm . getPriority ( ) ) ; msr . setSequenceid ( tm . getSequenceId ( ) ) ; if ( msr . getPriority ( ) == JetstreamMessage . INTERNAL_MSG_PRIORITY ) { if ( ! m_internalMsgProcessor . processRequest ( msr ) ) { m_totalMsgsDropped . increment ( ) ; throw new MessageServiceException ( MessageServiceException . BUFFER_FULL , "Dispatch Queue Full" ) ; } if ( m_msgsRcvdPerSec . addAndGet ( 1 ) < 0 ) m_msgsRcvdPerSec . set ( 0 ) ; m_totalMsgsRcvd . increment ( ) ; } else { requests . add ( msr ) ; } } if ( ! requests . isEmpty ( ) ) { int batchsize = requests . size ( ) ; if ( ! m_msgProcessor . processBatch ( requests ) ) { m_totalMsgsDropped . addAndGet ( batchsize ) ; throw new MessageServiceException ( MessageServiceException . BUFFER_FULL , "High Priority Dispatch Queue Full - " + " Requested capacity = " + batchsize + " : available capacity = " + m_msgProcessor . getAvailableCapacity ( ) ) ; } m_avgMsgsRcvdPerSec . add ( batchsize ) ; m_totalMsgsRcvd . addAndGet ( batchsize ) ; } if ( stats != null ) { stats . setHighPriorityQueueDepth ( ( int ) m_msgProcessor . getPendingRequests ( ) ) ; stats . setLowPriorityQueueDepth ( ( int ) m_msgProcessor . getPendingRequests ( ) ) ; stats . setMaxQueueDepth ( ( int ) m_msgProcessor . getMaxQueueSz ( ) ) ; } }
post a batch of messages .
