/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 * 
 *   http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */

package org.apache.uima.aae.handler.input;

import java.io.ByteArrayInputStream;
import java.util.ArrayList;
import java.util.List;

import org.apache.uima.UIMAFramework;
import org.apache.uima.UIMARuntimeException;
import org.apache.uima.aae.InProcessCache.CacheEntry;
import org.apache.uima.aae.SerializerCache;
import org.apache.uima.aae.UIMAEE_Constants;
import org.apache.uima.aae.UimaSerializer;
import org.apache.uima.aae.controller.AggregateAnalysisEngineController;
import org.apache.uima.aae.controller.AnalysisEngineController;
import org.apache.uima.aae.controller.BaseAnalysisEngineController;
import org.apache.uima.aae.controller.Endpoint;
import org.apache.uima.aae.controller.LocalCache.CasStateEntry;
import org.apache.uima.aae.delegate.Delegate;
import org.apache.uima.aae.error.AsynchAEException;
import org.apache.uima.aae.error.ErrorContext;
import org.apache.uima.aae.error.ExpiredMessageException;
import org.apache.uima.aae.error.ServiceShutdownException;
import org.apache.uima.aae.error.UimaAsDelegateException;
import org.apache.uima.aae.error.UimaEEServiceException;
import org.apache.uima.aae.handler.HandlerBase;
import org.apache.uima.aae.jmx.ServicePerformance;
import org.apache.uima.aae.message.AsynchAEMessage;
import org.apache.uima.aae.message.MessageContext;
import org.apache.uima.aae.monitor.Monitor;
import org.apache.uima.aae.monitor.statistics.AnalysisEnginePerformanceMetrics;
import org.apache.uima.aae.monitor.statistics.LongNumericStatistic;
import org.apache.uima.cas.CAS;
import org.apache.uima.cas.SerialFormat;
import org.apache.uima.cas.impl.AllowPreexistingFS;
import org.apache.uima.cas.impl.BinaryCasSerDes6;
import org.apache.uima.cas.impl.CASImpl;
import org.apache.uima.cas.impl.MarkerImpl;
import org.apache.uima.cas.impl.Serialization;
import org.apache.uima.cas.impl.TypeSystemImpl;
import org.apache.uima.cas.impl.XmiSerializationSharedData;
import org.apache.uima.cas.impl.BinaryCasSerDes6.ReuseInfo;
import org.apache.uima.util.Level;

public class ProcessResponseHandler extends HandlerBase {
  private static final Class CLASS_NAME = ProcessResponseHandler.class;

  public ProcessResponseHandler(String aName) {
    super(aName);
  }

  private Endpoint lookupEndpoint(String anEndpointName, String aCasReferenceId) {
    return getController().getInProcessCache().getEndpoint(anEndpointName, aCasReferenceId);
  }

  private void cancelTimer(MessageContext aMessageContext, String aCasReferenceId,
          boolean removeEndpoint) throws AsynchAEException {
    if (aMessageContext != null && aMessageContext.getEndpoint() != null) {
      if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINE)) {
        UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, CLASS_NAME.getName(), "cancelTimer",
                UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_cancel_timer__FINE",
                new Object[] { aMessageContext.getEndpoint().getEndpoint(), aCasReferenceId });
      }
      // Retrieve the endpoint from the cache using endpoint name
      // and casRefereceId
      if (aCasReferenceId == null
              && aMessageContext.propertyExists(AsynchAEMessage.Command)
              && aMessageContext.getMessageIntProperty(AsynchAEMessage.Command) == AsynchAEMessage.CollectionProcessComplete) {
        aCasReferenceId = ":CpC";
      }
      if (aMessageContext != null && aMessageContext.getEndpoint() != null) {
        Endpoint endpoint = lookupEndpoint(aMessageContext.getEndpoint().getEndpoint(),
                aCasReferenceId);

        if (endpoint != null) {
          // Received the response within timeout interval so
          // cancel the running timer
          endpoint.cancelTimer();
          if (removeEndpoint) {
            getController().getInProcessCache().removeEndpoint(
                    aMessageContext.getEndpoint().getEndpoint(), aCasReferenceId);
          }
        } else {
          if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) {
            UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(),
                    "cancelTimer", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
                    "UIMAEE_endpoint_not_found__INFO",
                    new Object[] { aMessageContext.getEndpoint().getEndpoint(), aCasReferenceId });
          }
        }
      }
    }
  }

  private void cancelTimerAndProcess(MessageContext aMessageContext, String aCasReferenceId,
          CAS aCAS) throws AsynchAEException {
    computeStats(aMessageContext, aCasReferenceId);

    super.invokeProcess(aCAS, aCasReferenceId, null, aMessageContext, null);

  }

  private boolean isMessageExpected(String aCasReferenceId, Endpoint anEndpointWithTimer)

  {
    if (getController().getInProcessCache().entryExists(aCasReferenceId)
            && anEndpointWithTimer.isWaitingForResponse()) {
      return true;
    }
    return false;
  }

  private void handleUnexpectedMessage(String aCasReferenceId, Endpoint anEndpoint) {
    // Cas does not exist in the CAS Cache. This would be possible if the
    // CAS has been dropped due to timeout and the delegate
    // sends the response later. In asynch communication this scenario is
    // possible. The service may not be up when the client sends
    // the message. Messages accumulate in the service queue until the
    // service becomes available. When this happens, the service
    // will pickup messages from the queue, process them and send respones
    // to an appropriate response queue. Most likely such
    // respones should be thrown away. Well perhaps logged first.
    ErrorContext errorContext = new ErrorContext();
    errorContext.add(AsynchAEMessage.CasReference, aCasReferenceId);
    errorContext.add(AsynchAEMessage.Command, AsynchAEMessage.Process);
    errorContext.add(AsynchAEMessage.Endpoint, anEndpoint);
    AnalysisEngineController controller = getController();
    controller.getErrorHandlerChain().handle(new ExpiredMessageException(), errorContext,
            controller);

  }

  private void handleProcessResponseFromRemote(MessageContext aMessageContext, String aDelegateKey) {
    CAS cas = null;
    String casReferenceId = null;
    Endpoint endpointWithTimer = null;
    try {
      final int payload = aMessageContext.getMessageIntProperty(AsynchAEMessage.Payload);
      casReferenceId = aMessageContext.getMessageStringProperty(AsynchAEMessage.CasReference);
      endpointWithTimer = lookupEndpoint(aMessageContext.getEndpoint().getEndpoint(),
              casReferenceId);

      if (endpointWithTimer == null) {
        if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) {
          UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(),
                  "handleProcessResponseFromRemote", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
                  "UIMAEE_invalid_endpoint__WARNING",
                  new Object[] { aMessageContext.getEndpoint().getEndpoint(), casReferenceId });
        }
        return;
      }
      String delegateKey = ((AggregateAnalysisEngineController) getController())
              .lookUpDelegateKey(aMessageContext.getEndpoint().getEndpoint());
      Delegate delegate = ((AggregateAnalysisEngineController) getController())
              .lookupDelegate(delegateKey);
      boolean casRemovedFromOutstandingList = delegate.removeCasFromOutstandingList(casReferenceId);

      // Check if this process reply message is expected. A message is expected if the Cas Id
      // in the message matches an entry in the delegate's outstanding list. This list stores
      // ids of CASes sent to the remote delegate pending reply.
      if (!casRemovedFromOutstandingList) {
        //handleUnexpectedMessage(casReferenceId, aMessageContext.getEndpoint());
        return;   // out of band reply. Most likely the CAS previously timedout
      }

      // Increment number of CASes processed by this delegate
      if (aDelegateKey != null) {
        ServicePerformance delegateServicePerformance = ((AggregateAnalysisEngineController) getController())
                .getServicePerformance(aDelegateKey);
        if (delegateServicePerformance != null) {
          delegateServicePerformance.incrementNumberOfCASesProcessed();
        }
      }

      String xmi = aMessageContext.getStringMessage();

      // Fetch entry from the cache for a given Cas Id. The entry contains a CAS that will be used
      // during deserialization
      CacheEntry cacheEntry = getController().getInProcessCache().getCacheEntryForCAS(
              casReferenceId);
      // check if the client requested Performance Metrics for the CAS
      if ( aMessageContext.propertyExists(AsynchAEMessage.CASPerComponentMetrics) ) {
        try {
          // find top ancestor of this CAS. All metrics are accumulated there since
          // this is what will be returned to the client
          CacheEntry ancestor = 
                  getController().
                    getInProcessCache().
                      getTopAncestorCasEntry(cacheEntry);
          if ( ancestor != null ) {
        	// fetch Performance Metrics from remote delegate reply
            List<AnalysisEnginePerformanceMetrics> metrics = 
                    UimaSerializer.deserializePerformanceMetrics(aMessageContext.getMessageStringProperty(AsynchAEMessage.CASPerComponentMetrics));
            List<AnalysisEnginePerformanceMetrics> adjustedMetrics =
                    new ArrayList<AnalysisEnginePerformanceMetrics>();
            for(AnalysisEnginePerformanceMetrics delegateMetric : metrics ) {
              String adjustedUniqueName = ((AggregateAnalysisEngineController) getController()).getJmxContext();

              if ( adjustedUniqueName.startsWith("p0=")) {
            	  adjustedUniqueName = adjustedUniqueName.substring(3);  // skip p0=
              }
              adjustedUniqueName = adjustedUniqueName.replaceAll(" Components", "");
              if (!adjustedUniqueName.startsWith("/")) {
            	  adjustedUniqueName = "/"+adjustedUniqueName;
              }
              adjustedUniqueName += delegateMetric.getUniqueName();
              
              boolean found = false;
              AnalysisEnginePerformanceMetrics metric = null;
              for( AnalysisEnginePerformanceMetrics met : ancestor.getDelegateMetrics() ) {
            	  if ( met.getUniqueName().equals(adjustedUniqueName)) {
            		  long at = delegateMetric.getAnalysisTime();
            		  long count = delegateMetric.getNumProcessed();
            		  metric = new AnalysisEnginePerformanceMetrics(delegateMetric.getName(),adjustedUniqueName,at,count);
            		  found = true;
            		  ancestor.getDelegateMetrics().remove(met);
            		  break;
            	  }
              }
              if ( !found ) {
                  metric = new AnalysisEnginePerformanceMetrics(delegateMetric.getName(),adjustedUniqueName,delegateMetric.getAnalysisTime(),delegateMetric.getNumProcessed());
              }
              adjustedMetrics.add(metric);
            }
            
            ancestor.addDelegateMetrics(delegateKey, adjustedMetrics, true);  // true=remote
          }
        } catch (Exception e) {
          // An exception be be thrown here if the service is being stopped.
          // The top level controller may have already cleaned up the cache
          // and the getCacheEntryForCAS() will throw an exception. Ignore it
          // here, we are shutting down.
        }
        
      }
      CasStateEntry casStateEntry = ((AggregateAnalysisEngineController) getController())
              .getLocalCache().lookupEntry(casReferenceId);
      if (casStateEntry != null) {
        casStateEntry.setReplyReceived();
        // Set the key of the delegate that returned the CAS
        casStateEntry.setLastDelegate(delegate);
      } else {
        return; // Cache Entry Not found
      }

      cas = cacheEntry.getCas();
      int totalNumberOfParallelDelegatesProcessingCas = casStateEntry
              .getNumberOfParallelDelegates();
      if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINE)) {
        UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, CLASS_NAME.getName(),
                "handleProcessResponseFromRemote", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
                "UIMAEE_number_parallel_delegates_FINE",
                new Object[] { totalNumberOfParallelDelegatesProcessingCas, Thread.currentThread().getId(), Thread.currentThread().getName() });
      }
      if (totalNumberOfParallelDelegatesProcessingCas > 1) {
    	  // Block this thread until CAS is dispatched to all delegates in parallel step. Fixes race condition where
    	  // a reply comes from one of delegates in parallel step before dispatch sequence completes. Without
    	  // this blocking the result of analysis are merged into a CAS.
    	  casStateEntry.blockIfParallelDispatchNotComplete();
      }
      
      if (cas == null) {
        throw new AsynchAEException(Thread.currentThread().getName()
                + "-Cache Does not contain a CAS. Cas Reference Id::" + casReferenceId);
      }
      if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) {
        UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINEST, CLASS_NAME.getName(),
                "handleProcessResponseFromRemote", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
                "UIMAEE_rcvd_reply_FINEST",
                new Object[] { aMessageContext.getEndpoint().getEndpoint(), casReferenceId, xmi });
      }
      long t1 = getController().getCpuTime();
      /* --------------------- */
      /** DESERIALIZE THE CAS. */
      /* --------------------- */
      //all subsequent serialization must be complete CAS.
      if ( !aMessageContext.getMessageBooleanProperty(AsynchAEMessage.SentDeltaCas))  {
    	cacheEntry.setAcceptsDeltaCas(false);
      }

      SerialFormat serialFormat = endpointWithTimer.getSerialFormat();
      // check if the CAS is part of the Parallel Step
      if (totalNumberOfParallelDelegatesProcessingCas > 1) {
        // Synchronized because replies are merged into the same CAS.
        synchronized (cas) {
          if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINEST)) {
            UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINEST, CLASS_NAME.getName(),
                    "handleProcessResponseFromRemote", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
                    "UIMAEE_delegate_responded_count_FINEST",
                    new Object[] { casStateEntry.howManyDelegatesResponded(), casReferenceId });
          }
          // If a delta CAS, merge it while checking that no pre-existing FSs are modified.
          if (aMessageContext.getMessageBooleanProperty(AsynchAEMessage.SentDeltaCas)) {
            switch (serialFormat) {
            case XMI:
              int highWaterMark = cacheEntry.getHighWaterMark();
              deserialize(xmi, cas, casReferenceId, highWaterMark, AllowPreexistingFS.disallow);
              break;
            case COMPRESSED_FILTERED:
              deserialize(aMessageContext.getByteMessage(), cas, cacheEntry, endpointWithTimer.getTypeSystemImpl(), AllowPreexistingFS.disallow);
              break;
            default:
              throw new UIMARuntimeException(new Exception("Internal error"));
            }
          } else {
            // If not a delta CAS (old service), take all of first reply, and merge in the new
            // entries in the later replies. Ignoring pre-existing FS for 2.2.2 compatibility
            // Note: can't be a compressed binary - that would have returned a delta
            if (casStateEntry.howManyDelegatesResponded() == 0) {
              deserialize(xmi, cas, casReferenceId);
            } else { // process secondary reply from a parallel step
              int highWaterMark = cacheEntry.getHighWaterMark();
              deserialize(xmi, cas, casReferenceId, highWaterMark, AllowPreexistingFS.ignore);
            }
          }
          casStateEntry.incrementHowManyDelegatesResponded();
        }
      } else { // Processing a reply from a non-parallel delegate (binary or delta xmi or xmi)
        byte[] binaryData = aMessageContext.getByteMessage();
        ByteArrayInputStream istream = new ByteArrayInputStream(binaryData);
        switch (serialFormat) {
        case BINARY:
          ((CASImpl)cas).reinit(istream);
          break;
        case COMPRESSED_FILTERED:
          BinaryCasSerDes6 bcs = new BinaryCasSerDes6(cas, (MarkerImpl) cacheEntry.getMarker(), endpointWithTimer.getTypeSystemImpl(), cacheEntry.getCompress6ReuseInfo());          
          bcs.deserialize(istream, AllowPreexistingFS.allow);
          break;
        case XMI:
          if (aMessageContext.getMessageBooleanProperty(AsynchAEMessage.SentDeltaCas)) {
            int highWaterMark = cacheEntry.getHighWaterMark();
            deserialize(xmi, cas, casReferenceId, highWaterMark, AllowPreexistingFS.allow);
          } else {
            deserialize(xmi, cas, casReferenceId);
          }
          break;
        default:
          throw new UIMARuntimeException(new Exception("Internal error"));
        }
      }
      long timeToDeserializeCAS = getController().getCpuTime() - t1;

      getController().getServicePerformance().incrementCasDeserializationTime(timeToDeserializeCAS);

      ServicePerformance casStats = getController().getCasStatistics(casReferenceId);
      casStats.incrementCasDeserializationTime(timeToDeserializeCAS);
      LongNumericStatistic statistic;
      if ((statistic = getController().getMonitor().getLongNumericStatistic("",
              Monitor.TotalDeserializeTime)) != null) {
        statistic.increment(timeToDeserializeCAS);
      }

      computeStats(aMessageContext, casReferenceId);

      // Send CAS for processing when all delegates reply
      // totalNumberOfParallelDelegatesProcessingCas indicates how many delegates are processing CAS
      // in parallel. Default is 1, meaning only one delegate processes the CAS at the same.
      // Otherwise, check if all delegates responded before passing CAS on to the Flow Controller.
      // The idea is that all delegates processing one CAS concurrently must respond, before the CAS
      // is allowed to move on to the next step.
      // HowManyDelegatesResponded is incremented every time a parallel delegate sends response.
      if (totalNumberOfParallelDelegatesProcessingCas == 1
              || receivedAllResponsesFromParallelDelegates(casStateEntry,
                      totalNumberOfParallelDelegatesProcessingCas)) {
        super.invokeProcess(cas, casReferenceId, null, aMessageContext, null);
      }

    } catch (Exception e) {
      // Check if the exception was thrown by the Cache while looking up
      // the CAS. It may be the case if in the parallel step one thread
      // drops the CAS in the Error Handling while another thread processes
      // reply from another delegate in the Parallel Step. A race condition
      // may occur here. If one thread drops the CAS due to excessive exceptions
      // and Flow Controller is configured to drop the CAS, the other thread
      // should not be allowed to move the CAS to process()method. The second
      // thread will find the CAS missing in the cache and the logic below
      // just logs the stale CAS and returns and doesnt attempt to handle
      // the missing CAS exception.
      if (e instanceof AsynchAEException && e.getMessage() != null
              && e.getMessage().startsWith("Cas Not Found")) {
        String key = "N/A";
        if (endpointWithTimer != null) {
          key = ((AggregateAnalysisEngineController) getController())
                  .lookUpDelegateKey(endpointWithTimer.getEndpoint());
        }
        if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) {
          UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINEST, CLASS_NAME.getName(),
                  "handleProcessResponseFromRemote", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
                  "UIMAEE_stale_reply__INFO",
                  new Object[] { getController().getComponentName(), key, casReferenceId });
        }
        // The reply came late. The CAS was removed from the cache.
        return;
      }
      ErrorContext errorContext = new ErrorContext();
      errorContext.add(AsynchAEMessage.Command, AsynchAEMessage.Process);
      errorContext.add(AsynchAEMessage.CasReference, casReferenceId);
      errorContext.add(AsynchAEMessage.Endpoint, aMessageContext.getEndpoint());
      getController().getErrorHandlerChain().handle(e, errorContext, getController());
    } finally {
      incrementDelegateProcessCount(aMessageContext);
    }

  }

  private synchronized boolean receivedAllResponsesFromParallelDelegates(
          CasStateEntry aCasStateEntry, int totalNumberOfParallelDelegatesProcessingCas) {
    if (aCasStateEntry.howManyDelegatesResponded() == totalNumberOfParallelDelegatesProcessingCas) {
      aCasStateEntry.resetDelegateResponded();
      return true;
    }
    return false;
  }

  private void deserialize(String xmi, CAS cas, String casReferenceId, int highWaterMark,
          AllowPreexistingFS allow) throws Exception {
    XmiSerializationSharedData deserSharedData;
    deserSharedData = getController().getInProcessCache().getCacheEntryForCAS(casReferenceId)
            .getDeserSharedData();
    UimaSerializer uimaSerializer = SerializerCache.lookupSerializerByThreadId();
    uimaSerializer.deserializeCasFromXmi(xmi, cas, deserSharedData, true, highWaterMark, allow);
  }
  
  private void deserialize(byte[] bytes, CAS cas, CacheEntry cacheEntry, TypeSystemImpl remoteTs,
          AllowPreexistingFS allow) throws Exception {
    ByteArrayInputStream istream = new ByteArrayInputStream(bytes);
    ReuseInfo reuseInfo = cacheEntry.getCompress6ReuseInfo();

    Serialization.deserializeCAS(cas, istream, remoteTs, reuseInfo, allow);
  }

  private void deserialize(String xmi, CAS cas, String casReferenceId) throws Exception {
    CacheEntry entry = getController().getInProcessCache().getCacheEntryForCAS(casReferenceId);
    // Processing the reply from a standard, non-parallel delegate
    XmiSerializationSharedData deserSharedData;
    deserSharedData = entry.getDeserSharedData();
    if (deserSharedData == null) {
      deserSharedData = new XmiSerializationSharedData();
      entry.setXmiSerializationData(deserSharedData);
    }
    UimaSerializer uimaSerializer = SerializerCache.lookupSerializerByThreadId();
    uimaSerializer.deserializeCasFromXmi(xmi, cas, deserSharedData, true, -1);
  }

  private void handleProcessResponseWithCASReference(MessageContext aMessageContext) {
    String casReferenceId = null;
    CacheEntry cacheEntry = null;

    try {
      casReferenceId = aMessageContext.getMessageStringProperty(AsynchAEMessage.CasReference);
      cacheEntry = getController().getInProcessCache().getCacheEntryForCAS(casReferenceId);
      CasStateEntry casStateEntry = ((AggregateAnalysisEngineController) getController())
              .getLocalCache().lookupEntry(casReferenceId);

      CAS cas = cacheEntry.getCas();
      String delegateKey = ((AggregateAnalysisEngineController) getController())
              .lookUpDelegateKey(aMessageContext.getEndpoint().getEndpoint());
      Delegate delegate = ((AggregateAnalysisEngineController) getController())
              .lookupDelegate(delegateKey);
      if (casStateEntry != null) {
        casStateEntry.setReplyReceived();
        casStateEntry.setLastDelegate(delegate);
      }
      delegate.removeCasFromOutstandingList(casReferenceId);

      if (cas != null) {
        cancelTimerAndProcess(aMessageContext, casReferenceId, cas);
      } else {
        if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) {
          UIMAFramework.getLogger(CLASS_NAME).logrb(
                  Level.INFO,
                  CLASS_NAME.getName(),
                  "handleProcessResponseWithCASReference",
                  UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
                  "UIMAEE_cas_not_in_cache__INFO",
                  new Object[] { getController().getName(), casReferenceId,
                      aMessageContext.getEndpoint().getEndpoint() });
        }
        throw new AsynchAEException("CAS with Reference Id:" + casReferenceId
                + " Not Found in CasManager's CAS Cache");
      }
    } catch (Exception e) {

      ErrorContext errorContext = new ErrorContext();
      errorContext.add(AsynchAEMessage.Command, AsynchAEMessage.Process);
      errorContext.add(AsynchAEMessage.CasReference, casReferenceId);
      errorContext.add(AsynchAEMessage.Endpoint, aMessageContext.getEndpoint());
      getController().getErrorHandlerChain().handle(e, errorContext, getController());
    } finally {
      incrementDelegateProcessCount(aMessageContext);
      if (getController() instanceof AggregateAnalysisEngineController) {
        try {
          String endpointName = aMessageContext.getEndpoint().getEndpoint();
          String delegateKey = ((AggregateAnalysisEngineController) getController())
                  .lookUpDelegateKey(endpointName);
          if (delegateKey != null) {
            Endpoint endpoint = ((AggregateAnalysisEngineController) getController())
                    .lookUpEndpoint(delegateKey, false);

            // Check if the multiplier aborted during processing of this input CAS
            if (endpoint != null && endpoint.isCasMultiplier() && cacheEntry.isAborted()) {
              if (!getController().getInProcessCache().isEmpty()) {
                getController().getInProcessCache().registerCallbackWhenCacheEmpty(
                        getController().getEventListener());
              } else {
                // Callback to notify that the cache is empty
            	  
            	  // !!!!!!!!!!!!!!! WHY DO WE NEED TO CALL onCacheEmpty() IF CAS IS ABORTED?
            	  // !!!!!!!!!!!!!!!!!!!!!! ?????????????????????????????????
//                getController().getEventListener().onCacheEmpty();
              }
            }

          }
        } catch (Exception e) {
          if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) {
            if ( getController() != null ) {
              UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(),
                      "handleProcessResponseWithCASReference", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
                      "UIMAEE_service_exception_WARNING", getController().getComponentName());
            }

            UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, getClass().getName(),
                    "handleProcessResponseWithCASReference", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
                    "UIMAEE_exception__WARNING", e);
          }
        }
      }
    }

  }

  private void incrementDelegateProcessCount(MessageContext aMessageContext) {
    Endpoint endpoint = aMessageContext.getEndpoint();
    if (endpoint != null && getController() instanceof AggregateAnalysisEngineController) {
      try {
        String delegateKey = ((AggregateAnalysisEngineController) getController())
                .lookUpDelegateKey(endpoint.getEndpoint());
        LongNumericStatistic stat = getController().getMonitor().getLongNumericStatistic(
                delegateKey, Monitor.ProcessCount);
        stat.increment();
      } catch (Exception e) {
        if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) {
            UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(),
                    "incrementDelegateProcessCount", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
                    "UIMAEE_delegate_key_for_endpoint_not_found__INFO", new Object[] { getController().getComponentName(), endpoint.getEndpoint() });
        }
      }
    }

  }

  private boolean isException(Object object) {
    return (object instanceof Exception || object instanceof Throwable);
  }

  private boolean isShutdownException(Object object) {
    return (object instanceof Exception && object instanceof UimaEEServiceException
            && ((UimaEEServiceException) object).getCause() != null && ((UimaEEServiceException) object)
            .getCause() instanceof ServiceShutdownException);
  }

  private boolean ignoreException(Object object) {
    if (object != null && isException(object) && !isShutdownException(object)) {
      return false;
    }
    return true;
  }

  private synchronized void handleProcessResponseWithException(MessageContext aMessageContext,
          String delegateKey) {
    if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINE)) {
      UIMAFramework.getLogger(CLASS_NAME)
              .logrb(
                      Level.FINE,
                      CLASS_NAME.getName(),
                      "handleProcessResponseWithException",
                      UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
                      "UIMAEE_handling_exception_from_delegate_FINE",
                      new Object[] { getController().getName(),
                          aMessageContext.getEndpoint().getEndpoint() });
    }
    boolean isCpCError = false;
    String casReferenceId = null;

    try {
      // If a Process Request, increment number of docs processed
      if (aMessageContext.getMessageIntProperty(AsynchAEMessage.MessageType) == AsynchAEMessage.Response
              && aMessageContext.getMessageIntProperty(AsynchAEMessage.Command) == AsynchAEMessage.Process) {
        // Increment number of CASes processed by a delegate
        incrementDelegateProcessCount(aMessageContext);
      }

      Object object = aMessageContext.getObjectMessage();
      if (object == null) {
        // Could be a C++ exception. In this case the exception is just a String in the message
        // cargo
        if (aMessageContext.getStringMessage() != null) {
          object = new UimaEEServiceException(aMessageContext.getStringMessage());
        }
      }
      if (ignoreException(object)) {
        return;
      }

      if (getController() instanceof AggregateAnalysisEngineController
              && aMessageContext.propertyExists(AsynchAEMessage.Command)
              && aMessageContext.getMessageIntProperty(AsynchAEMessage.Command) == AsynchAEMessage.CollectionProcessComplete) {
        isCpCError = true;
        ((AggregateAnalysisEngineController) getController())
                .processCollectionCompleteReplyFromDelegate(delegateKey, false);
      } else {
        casReferenceId = aMessageContext.getMessageStringProperty(AsynchAEMessage.CasReference);
      }

      if (object != null && (object instanceof Exception || object instanceof Throwable)) {
        String casid_msg = (casReferenceId == null) ? "" : " on CAS:" + casReferenceId;
        String controllerName = "/" + getController().getComponentName();
        if (!getController().isTopLevelComponent()) {
          controllerName += ((BaseAnalysisEngineController) getController().getParentController())
                  .getUimaContextAdmin().getQualifiedContextName();
        }
        Exception remoteException = new UimaAsDelegateException("----> Controller:"
                + controllerName + " Received Exception " + casid_msg + " From Delegate:"
                + delegateKey, (Exception) object);
        ErrorContext errorContext = new ErrorContext();
        errorContext.add(AsynchAEMessage.Command, aMessageContext
                .getMessageIntProperty(AsynchAEMessage.Command));
        errorContext.add(AsynchAEMessage.MessageType, aMessageContext
                .getMessageIntProperty(AsynchAEMessage.MessageType));
        if (!isCpCError) {
          errorContext.add(AsynchAEMessage.CasReference, casReferenceId);
        }
        errorContext.add(AsynchAEMessage.Endpoint, aMessageContext.getEndpoint());
        getController().getErrorHandlerChain().handle(remoteException, errorContext,
                getController());
      }
    } catch (Exception e) {
      ErrorContext errorContext = new ErrorContext();
      errorContext.add(AsynchAEMessage.Command, AsynchAEMessage.Process);
      errorContext.add(AsynchAEMessage.CasReference, casReferenceId);
      errorContext.add(AsynchAEMessage.Endpoint, aMessageContext.getEndpoint());
      getController().getErrorHandlerChain().handle(e, errorContext, getController());
    }

  }

  private void handleCollectionProcessCompleteReply(MessageContext aMessageContext,
          String delegateKey) {
    try {
      if (getController() instanceof AggregateAnalysisEngineController) {
        ((AggregateAnalysisEngineController) getController())
                .processCollectionCompleteReplyFromDelegate(delegateKey, true);
      }
    } catch (Exception e) {
      ErrorContext errorContext = new ErrorContext();
      errorContext.add(AsynchAEMessage.Command, AsynchAEMessage.CollectionProcessComplete);
      errorContext.add(AsynchAEMessage.Endpoint, aMessageContext.getEndpoint());
      getController().getErrorHandlerChain().handle(e, errorContext, getController());
    }

  }

  private void resetErrorCounts(String aDelegate) {
    getController().getMonitor().resetCountingStatistic(aDelegate, Monitor.ProcessErrorCount);
    getController().getMonitor().resetCountingStatistic(aDelegate, Monitor.ProcessErrorRetryCount);
  }

  private void handlePingReply(MessageContext aMessageContext) {
    try {

    } catch (Exception e) {
      if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) {
        if ( getController() != null ) {
          UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(),
                  "handlePingReply", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
                  "UIMAEE_service_exception_WARNING", getController().getComponentName());
        }

        UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, getClass().getName(),
                "handlePingReply", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
                "UIMAEE_exception__WARNING", e);
      }
    }
  }

  private void handleServiceInfoReply(MessageContext messageContext) {

    String casReferenceId = null;
    try {
      casReferenceId = messageContext.getMessageStringProperty(AsynchAEMessage.CasReference);
      if ( casReferenceId == null ) {
    	  return;  // nothing to do
      }
      Endpoint freeCasEndpoint = messageContext.getEndpoint();
      CasStateEntry casStateEntry = ((AggregateAnalysisEngineController) getController())
              .getLocalCache().lookupEntry(casReferenceId);
      if (casStateEntry != null) {
        casStateEntry.setFreeCasNotificationEndpoint(freeCasEndpoint);
        //  Fetch host IP where the CAS is being processed. When the UIMA AS service
        //  receives a CAS it immediately sends ServiceInfo Reply message containing 
        //  IP of the host where the service is running.
        String serviceHostIp = messageContext.getMessageStringProperty(AsynchAEMessage.ServerIP);
        if ( serviceHostIp != null ) {
          casStateEntry.setHostIpProcessingCAS(serviceHostIp);
        }
      }
    } catch (Exception e) {
    	UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, getClass().getName(),
                "handleServiceInfoReply", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
                "UIMAEE_exception__WARNING", e);
    	return;
    }

  }

  public void handle(Object anObjectToHandle) throws AsynchAEException {
    super.validate(anObjectToHandle);
    MessageContext messageContext = (MessageContext) anObjectToHandle;

    if (isHandlerForMessage(messageContext, AsynchAEMessage.Response, AsynchAEMessage.Process)
            || isHandlerForMessage(messageContext, AsynchAEMessage.Response, AsynchAEMessage.ACK)
            || isHandlerForMessage(messageContext, AsynchAEMessage.Response,
                    AsynchAEMessage.ServiceInfo)
            || isHandlerForMessage(messageContext, AsynchAEMessage.Response,
                    AsynchAEMessage.CollectionProcessComplete)) {
      int payload = messageContext.getMessageIntProperty(AsynchAEMessage.Payload);
      int command = messageContext.getMessageIntProperty(AsynchAEMessage.Command);
      String delegate = ((Endpoint) messageContext.getEndpoint()).getEndpoint();
      String key = null;
      String fromServer = null;
      if (getController() instanceof AggregateAnalysisEngineController) {
        if (((Endpoint) messageContext.getEndpoint()).isRemote()) {
          if (((MessageContext) anObjectToHandle).propertyExists(AsynchAEMessage.EndpointServer)) {
            fromServer = ((MessageContext) anObjectToHandle)
                    .getMessageStringProperty(AsynchAEMessage.EndpointServer);
          }
          // If old service does not echo back the external broker name then the queue name must be
          // unique.
          // Can't use the ServerURI set by the service as it may be its local name for the broker,
          // e.g. tcp://localhost:61616
        }

        key = ((AggregateAnalysisEngineController) getController()).lookUpDelegateKey(delegate,
                fromServer);
      }
      if (AsynchAEMessage.CASRefID == payload) {

        handleProcessResponseWithCASReference(messageContext);
        if (key != null) {
          resetErrorCounts(key);
        }
      } else if (AsynchAEMessage.XMIPayload == payload || AsynchAEMessage.BinaryPayload == payload) {
        handleProcessResponseFromRemote(messageContext, key);
        if (key != null) {
          resetErrorCounts(key);
        }
      } else if (AsynchAEMessage.Exception == payload) {
        if (key == null) {
          key = ((Endpoint) messageContext.getEndpoint()).getEndpoint();
        }
        handleProcessResponseWithException(messageContext, key);
      } else if (AsynchAEMessage.None == payload
              && AsynchAEMessage.CollectionProcessComplete == command) {
        if (key == null) {
          key = ((Endpoint) messageContext.getEndpoint()).getEndpoint();
        }
        handleCollectionProcessCompleteReply(messageContext, key);
      } else if (AsynchAEMessage.None == payload && AsynchAEMessage.ACK == command) {
        handleACK(messageContext, key);
      } else if (AsynchAEMessage.None == payload && AsynchAEMessage.Ping == command) {
        handlePingReply(messageContext);
      } else if (AsynchAEMessage.None == payload && AsynchAEMessage.ServiceInfo == command) {
        handleServiceInfoReply(messageContext);
      } else {
        throw new AsynchAEException("Invalid Payload. Expected XMI or CasReferenceId Instead Got::"
                + payload);
      }

      // Handled Request to Process with A given Payload
      return;
    }
    // Not a Request nor Command. Delegate to the next handler in the chain
    super.delegate(messageContext);

  }

  private void handleACK(MessageContext aMessageContext, String key) throws AsynchAEException {
  }

}