@Override public void processUpdates(final DataTagValueUpdate dataTagValueUpdate){
  try {
    activeUpdateThreads.getAndIncrement();
    Collection<SourceDataTagValue> values=dataTagValueUpdate.getValues();
    if (values != null) {
      for (      SourceDataTagValue sourceDataTagValue : values) {
        if (sourceDataTagValue.isControlTag()) {
          processControl(sourceDataTagValue);
        }
 else {
          processDataTag(sourceDataTagValue);
        }
        sourceDataTagValue.log();
      }
    }
  }
  finally {
    activeUpdateThreads.getAndDecrement();
    if (activeUpdateThreads.get() > 100) {
      alarmActive=true;
      switchOffCountDown=new AtomicInteger(SWITCH_OFF_COUNTDOWN);
      if (warningCount.getAndIncrement() % WARNING_FREQUENCY == 0) {
        SMSLOGGER.warn("Over 100 source update threads active.");
      }
    }
 else     if (alarmActive && switchOffCountDown.getAndDecrement() == 0) {
      alarmActive=false;
      SMSLOGGER.warn("Number of active update threads back to normal.");
      warningCount=new AtomicInteger(0);
    }
  }
}

@ Override public void processUpdates ( final DataTagValueUpdate dataTagValueUpdate ) { try { activeUpdateThreads . getAndIncrement ( ) ; Collection < SourceDataTagValue > values = dataTagValueUpdate . getValues ( ) ; if ( values != null ) { for ( SourceDataTagValue sourceDataTagValue : values ) { if ( sourceDataTagValue . isControlTag ( ) ) { processControl ( sourceDataTagValue ) ; } else { processDataTag ( sourceDataTagValue ) ; } sourceDataTagValue . log ( ) ; } } } finally { activeUpdateThreads . getAndDecrement ( ) ; if ( activeUpdateThreads . get ( ) > 100 ) { alarmActive = true ; switchOffCountDown = new AtomicInteger ( SWITCH_OFF_COUNTDOWN ) ; if ( warningCount . getAndIncrement ( ) % WARNING_FREQUENCY == 0 ) { SMSLOGGER . warn ( "Over 100 source update threads active." ) ; } } else if ( alarmActive && switchOffCountDown . getAndDecrement ( ) == 0 ) { alarmActive = false ; SMSLOGGER . warn ( "Number of active update threads back to normal." ) ; warningCount = new AtomicInteger ( 0 ) ; } } }
implementation of the interface of the source update management module . the incoming jms message is only acknowledged once this method has returned successfully ( since this method is specified as listener method in the jca container ; the transaction manager manages the acknowledgment ) .
