/*
 * 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 org.apache.uima.UIMAFramework;
import org.apache.uima.aae.UIMAEE_Constants;
import org.apache.uima.aae.UimaSerializer;
import org.apache.uima.aae.InProcessCache.CacheEntry;
import org.apache.uima.aae.controller.AggregateAnalysisEngineController;
import org.apache.uima.aae.controller.AggregateAnalysisEngineController_impl;
import org.apache.uima.aae.controller.Endpoint;
import org.apache.uima.aae.controller.Endpoint_impl;
import org.apache.uima.aae.controller.PrimitiveAnalysisEngineController;
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.InvalidMessageException;
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.DelegateStats;
import org.apache.uima.aae.monitor.statistics.LongNumericStatistic;
import org.apache.uima.aae.monitor.statistics.TimerStats;
import org.apache.uima.analysis_engine.asb.impl.FlowContainer;
import org.apache.uima.cas.CAS;
import org.apache.uima.cas.Marker;
import org.apache.uima.cas.impl.XmiSerializationSharedData;
import org.apache.uima.util.Level;

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

  private Object mux = new Object();

  private UimaSerializer uimaSerializer = new UimaSerializer();

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

  private void cacheStats(String aCasReferenceId, long aTimeWaitingForCAS,
          long aTimeToDeserializeCAS) throws Exception {
    CacheEntry entry = getController().getInProcessCache().getCacheEntryForCAS(aCasReferenceId);
    entry.incrementTimeWaitingForCAS(aTimeWaitingForCAS);
    entry.incrementTimeToDeserializeCAS(aTimeToDeserializeCAS);
  }

  private boolean messageContainsXMI(MessageContext aMessageContext, String casReferenceId)
          throws Exception {
    // Fetch serialized CAS from the message
    String xmi = aMessageContext.getStringMessage();
    // *****************************************************************
    // ***** NO XMI In Message. Kick this back to sender with exception
    // *****************************************************************
    if (xmi == null) {
      if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) {
        UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(),
                "handleProcessRequestWithXMI", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
                "UIMAEE_message_has_no_cargo__INFO",
                new Object[] { aMessageContext.getEndpoint().getEndpoint() });
      }
      CasStateEntry stateEntry = null;
      String parentCasReferenceId = null;
      try {
        stateEntry = getController().getLocalCache().lookupEntry(casReferenceId);
        if (stateEntry != null && stateEntry.isSubordinate()) {
          CasStateEntry topParentEntry = getController().getLocalCache().getTopCasAncestor(
                  casReferenceId);
          parentCasReferenceId = topParentEntry.getCasReferenceId();
        }
      } catch (Exception e) {
      }

      getController().getOutputChannel().sendReply(
              new InvalidMessageException("No XMI data in message"), casReferenceId,
              parentCasReferenceId, aMessageContext.getEndpoint(), AsynchAEMessage.Process);
      // Dont process this empty message
      return false;
    }
    return true;
  }

  private synchronized CAS getCAS(boolean fetchCASFromShadowCasPool, String shadowCasPoolKey,
          String casReceivedFrom) {
    CAS cas = null;
    // If this is a new CAS (generated by a CM), fetch a CAS from a Shadow Cas Pool associated with
    // a CM that
    // produced the CAS. Each CM will have its own Shadow Cas Pool
    if (fetchCASFromShadowCasPool) {
      if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINEST)) {
        UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINEST, CLASS_NAME.getName(), "getCAS",
                UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_request_cas_cm__FINE",
                new Object[] { shadowCasPoolKey });
      }
      // Aggregate time spent waiting for a CAS in the shadow cas pool
      ((AggregateAnalysisEngineController) getController()).getDelegateServicePerformance(
              shadowCasPoolKey).beginWaitOnShadowCASPool();
      cas = getController().getCasManagerWrapper().getNewCas(shadowCasPoolKey);
      ((AggregateAnalysisEngineController) getController()).getDelegateServicePerformance(
              shadowCasPoolKey).endWaitOnShadowCASPool();

      if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINE)) {
        UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, CLASS_NAME.getName(), "getCAS",
                UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_request_cas_granted_cm__FINE",
                new Object[] { shadowCasPoolKey });
      }
    } else {
      if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINEST)) {
        UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINEST, CLASS_NAME.getName(), "getCAS",
                UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_request_cas__FINE",
                new Object[] { casReceivedFrom });
      }
      // Aggregate time spent waiting for a CAS in the service cas pool
      getController().getServicePerformance().beginWaitOnCASPool();

      cas = getController().getCasManagerWrapper().getNewCas();
      getController().getServicePerformance().endWaitOnCASPool();
      ServicePerformance sp = getController().getServicePerformance();
      sp.incrementCasPoolWaitTime(sp.getTimeWaitingForCAS());

      if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINE)) {
        UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, CLASS_NAME.getName(), "getCAS",
                UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_request_cas_granted__FINE",
                new Object[] { casReceivedFrom });
      }
    }
    return cas;
  }

  /**
   * 
   * @param casReferenceId
   * @param freeCasEndpoint
   * @param shadowCasPoolKey
   * @param aMessageContext
   * @return
   * @throws Exception
   */
  private CacheEntry deserializeCASandRegisterWithCache(String casReferenceId,
          Endpoint freeCasEndpoint, String shadowCasPoolKey, MessageContext aMessageContext)
          throws Exception {
    long inTime = System.nanoTime();
    boolean casRegistered = false;

    // Time how long we wait on Cas Pool to fetch a new CAS
    long t1 = getController().getCpuTime();
    // *************************************************************************
    // Fetch CAS from a Cas Pool. If the CAS came from a Cas Multiplier
    // fetch the CAS from a shadow CAS pool. Otherwise, fetch the CAS
    // from the service CAS Pool.
    // *************************************************************************
    Endpoint endpoint = aMessageContext.getEndpoint();

    CAS cas = getCAS(aMessageContext.propertyExists(AsynchAEMessage.CasSequence), shadowCasPoolKey,
            endpoint.getEndpoint());
    long timeWaitingForCAS = getController().getCpuTime() - t1;
    // Check if we are still running
    if (getController().isStopped()) {
      // The Controller is in shutdown state.
      getController().dropCAS(cas);
      return null;
    }
    // *************************************************************************
    // Deserialize CAS from the message
    // *************************************************************************
    t1 = getController().getCpuTime();
    String serializationStrategy = endpoint.getSerializer();
    XmiSerializationSharedData deserSharedData = null;
    CacheEntry entry = null;
    if (serializationStrategy.equals("xmi")) {
      // Fetch serialized CAS from the message
      String xmi = aMessageContext.getStringMessage();
      deserSharedData = new XmiSerializationSharedData();
      // UimaSerializer.deserializeCasFromXmi(xmi, cas, deserSharedData, true, -1);
      uimaSerializer.deserializeCasFromXmi(xmi, cas, deserSharedData, true, -1);
    } else if (serializationStrategy.equals("binary")) {
      // *************************************************************************
      // Register the CAS with a local cache
      // *************************************************************************
      // CacheEntry entry = getController().getInProcessCache().register(cas, aMessageContext,
      // deserSharedData, casReferenceId);
      byte[] binarySource = aMessageContext.getByteMessage();
      // UimaSerializer.deserializeCasFromBinary(binarySource, cas);
      uimaSerializer.deserializeCasFromBinary(binarySource, cas);
    }

    // *************************************************************************
    // Check and set up for Delta CAS reply
    // *************************************************************************
    boolean acceptsDeltaCas = false;
    Marker marker = null;
    if (aMessageContext.propertyExists(AsynchAEMessage.AcceptsDeltaCas)) {
      acceptsDeltaCas = aMessageContext.getMessageBooleanProperty(AsynchAEMessage.AcceptsDeltaCas);
      if (acceptsDeltaCas) {
        marker = cas.createMarker();
      }
    }
    // *************************************************************************
    // Register the CAS with a local cache
    // *************************************************************************
    // CacheEntry entry = getController().getInProcessCache().register(cas, aMessageContext,
    // deserSharedData, casReferenceId);
    entry = getController().getInProcessCache().register(cas, aMessageContext, deserSharedData,
            casReferenceId, marker, acceptsDeltaCas);

    long timeToDeserializeCAS = getController().getCpuTime() - t1;
    getController().incrementDeserializationTime(timeToDeserializeCAS);
    LongNumericStatistic statistic;
    if ((statistic = getController().getMonitor().getLongNumericStatistic("",
            Monitor.TotalDeserializeTime)) != null) {
      statistic.increment(timeToDeserializeCAS);
    }
    if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINE)) {
      UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, CLASS_NAME.getName(),
              "handleProcessRequestWithXMI", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
              "UIMAEE_deserialize_cas_time_FINE",
              new Object[] { (double) timeToDeserializeCAS / 1000000.0 });
    }

    // Update Stats
    ServicePerformance casStats = getController().getCasStatistics(casReferenceId);
    casStats.incrementCasDeserializationTime(timeToDeserializeCAS);
    if (getController().isTopLevelComponent()) {
      synchronized (mux) {
        getController().getServicePerformance().incrementCasDeserializationTime(
                timeToDeserializeCAS);
      }
    }
    getController().saveTime(inTime, casReferenceId, getController().getName());

    if (getController() instanceof AggregateAnalysisEngineController) {
      // If the message came from a Cas Multiplier, associate the input/parent CAS id with this CAS
      if (aMessageContext.propertyExists(AsynchAEMessage.CasSequence)) {
        // Fetch parent CAS id
        String inputCasReferenceId = aMessageContext
                .getMessageStringProperty(AsynchAEMessage.InputCasReference);
        if (shadowCasPoolKey != null) {
          // Save the key of the Cas Multiplier in the cache. It will be now known which Cas
          // Multiplier produced this CAS
          entry.setCasProducerKey(shadowCasPoolKey);
        }
        // associate this subordinate CAS with the parent CAS
        entry.setInputCasReferenceId(inputCasReferenceId);
        // Save a Cas Multiplier endpoint where a Free CAS notification will be sent
        entry.setFreeCasEndpoint(freeCasEndpoint);
        cacheStats(inputCasReferenceId, timeWaitingForCAS, timeToDeserializeCAS);
      } else {
        cacheStats(casReferenceId, timeWaitingForCAS, timeToDeserializeCAS);
      }
      DelegateStats stats = new DelegateStats();
      if (entry.getStat() == null) {
        entry.setStat(stats);
        // Add entry for self (this aggregate). MessageContext.getEndpointName()
        // returns the name of the queue receiving the message.
        stats.put(getController().getServiceEndpointName(), new TimerStats());
      } else {
        if (!stats.containsKey(getController().getServiceEndpointName())) {
          stats.put(getController().getServiceEndpointName(), new DelegateStats());
        }
      }
    } else {
      cacheStats(casReferenceId, timeWaitingForCAS, timeToDeserializeCAS);
    }
    if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINE)) {
      UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, CLASS_NAME.getName(),
              "handleProcessRequestWithXMI", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
              "UIMAEE_deserialized_cas_ready_to_process_FINE",
              new Object[] { aMessageContext.getEndpoint().getEndpoint() });
    }
    cacheProcessCommandInClientEndpoint();
    return entry;
  }

  private String getCasReferenceId(MessageContext aMessageContext) throws Exception {
    if (!aMessageContext.propertyExists(AsynchAEMessage.CasReference)) {
      if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) {
        UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(),
                "handleProcessRequestWithCASReference", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
                "UIMAEE_message_has_cas_refid__INFO",
                new Object[] { aMessageContext.getEndpoint().getEndpoint() });
      }
      getController().getOutputChannel().sendReply(
              new InvalidMessageException("No Cas Reference Id Received From Delegate In message"),
              null, null, aMessageContext.getEndpoint(), AsynchAEMessage.Process);
      return null;
    }
    return aMessageContext.getMessageStringProperty(AsynchAEMessage.CasReference);
  }

  /**
   * Handles process request from a remote client
   * 
   * @param aMessageContext
   *          - contains a message from UIMA-AS Client
   * @throws AsynchAEException
   */
  private void handleProcessRequestFromRemoteClient(MessageContext aMessageContext)
          throws AsynchAEException {
    CacheEntry entry = null;
    String casReferenceId = null;
    // Check if there is a cargo in the message
    if (aMessageContext.getMessageIntProperty(AsynchAEMessage.Payload) == AsynchAEMessage.XMIPayload
            && aMessageContext.getStringMessage() == null) {
      return; // No XMI just return
    }

    try {

      String newCASProducedBy = null;
      // Get the CAS Reference Id of the input CAS
      // Fetch id of the CAS from the message. If it doesnt exist the method will create an entry in
      // the log file and return null
      casReferenceId = getCasReferenceId(aMessageContext);
      if (casReferenceId == null) {
        return; // Invalid message. Nothing to do
      }
      // Initially make both equal
      String inputCasReferenceId = casReferenceId;
      // Destination where Free Cas Notification will be sent if the CAS came from a Cas Multiplier
      Endpoint freeCasEndpoint = null;

      CasStateEntry inputCasStateEntry = null;

      // CASes generated by a Cas Multiplier will have a CasSequence property set.
      if (aMessageContext.propertyExists(AsynchAEMessage.CasSequence)) {
        // Fetch the name of the Cas Multiplier's input queue
        // String cmEndpointName = aMessageContext.getEndpoint().getEndpoint();
        String cmEndpointName = aMessageContext
                .getMessageStringProperty(AsynchAEMessage.MessageFrom);
        newCASProducedBy = ((AggregateAnalysisEngineController) getController())
                .lookUpDelegateKey(cmEndpointName);
        // Fetch an ID of the parent CAS
        inputCasReferenceId = aMessageContext
                .getMessageStringProperty(AsynchAEMessage.InputCasReference);
        // Fetch Cache entry for the parent CAS
        CacheEntry inputCasCacheEntry = getController().getInProcessCache().getCacheEntryForCAS(
                inputCasReferenceId);
        // Fetch an endpoint where Free CAS Notification must be sent.
        // This endpoint is unique per CM instance. Meaning, each
        // instance of CM will have an endpoint where it expects Free CAS
        // notifications.
        freeCasEndpoint = aMessageContext.getEndpoint();
        // Clone an endpoint where Free Cas Request will be sent
        freeCasEndpoint = (Endpoint) ((Endpoint_impl) freeCasEndpoint).clone();

        if (getController() instanceof AggregateAnalysisEngineController) {
          inputCasStateEntry = ((AggregateAnalysisEngineController) getController())
                  .getLocalCache().lookupEntry(inputCasReferenceId);

          // Associate Free Cas Notification Endpoint with an input Cas
          inputCasStateEntry.setFreeCasNotificationEndpoint(freeCasEndpoint);
        }

        computeStats(aMessageContext, inputCasReferenceId);
        // Reset the destination
        aMessageContext.getEndpoint().setDestination(null);
        // This CAS came in from a CAS Multiplier. Treat it differently than the
        // input CAS. In case the Aggregate needs to send this CAS to the
        // client, retrieve the client destination by looking up the client endpoint
        // using input CAS reference id. CASes generated by the CAS multiplier will have
        // the same Cas Reference id.
        Endpoint replyToEndpoint = inputCasCacheEntry.getMessageOrigin();
        // The message context contains a Cas Multiplier endpoint. Since
        // we dont want to send a generated CAS back to the CM, override
        // with an endpoint provided by the client of
        // this service. Client endpoint is attached to an input Cas cache entry.
        aMessageContext.getEndpoint().setEndpoint(replyToEndpoint.getEndpoint());
        aMessageContext.getEndpoint().setServerURI(replyToEndpoint.getServerURI());

        // Before sending a CAS to Cas Multiplier, the aggregate has
        // saved the CM key in the CAS cache entry. Fetch the key
        // of the CM so that we can ask the right Shadow Cas Pool for
        // a new CAS. Every Shadow Cas Pool has a unique id which
        // corresponds to a Cas Multiplier key.
        // newCASProducedBy = inputCasCacheEntry.getCasMultiplierKey();
        if (getController() instanceof AggregateAnalysisEngineController) {
          Endpoint casMultiplierEndpoint = ((AggregateAnalysisEngineController) getController())
                  .lookUpEndpoint(newCASProducedBy, false);
          if (casMultiplierEndpoint != null) {
            // Save the URL of the broker managing the Free Cas Notification queue.
            // This is needed when we try to establish a connection to the broker.
            freeCasEndpoint.setServerURI(casMultiplierEndpoint.getServerURI());
          }
        }
      } else if (getController().isTopLevelComponent()) {
        if (getController() instanceof AggregateAnalysisEngineController) {
          ((AggregateAnalysisEngineController) getController()).addMessageOrigin(casReferenceId,
                  aMessageContext.getEndpoint());
        }
        if (getController().isCasMultiplier()) {
          // Send an ack to the client. The ack message will include a FreeCasQueue
          // to enable the client to send messages to the service processing a CAS.
          try {
            getController().getOutputChannel().sendReply(AsynchAEMessage.ServiceInfo,
                    aMessageContext.getEndpoint(), casReferenceId);
          } catch (Exception e) {
            if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) {
              UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, getClass().getName(),
                      "handleProcessRequestFromRemoteClient", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
                      "UIMAEE_exception__WARNING", new Object[] { e });
            }
          }
        }
      }
      // To prevent processing multiple messages with the same CasReferenceId, check the CAS cache
      // to see if the message with a given CasReferenceId is already being processed. It is, the
      // message contains the same request possibly issued by the caller due to timeout. Also this
      // mechanism helps with dealing with scenario when this service is not up when the client
      // sends
      // request. The client can keep re-sending the same request until its timeout thresholds are
      // exceeded. By that time, there may be multiple messages in this service queue with the same
      // CasReferenceId. When the service finally comes back up, it will have multiple messages in
      // its queue possibly from the same client. Only the first message for any given
      // CasReferenceId
      // should be processed.
      if (!getController().getInProcessCache().entryExists(casReferenceId)) {
        CasStateEntry cse = null;
        if (getController().getLocalCache().lookupEntry(casReferenceId) == null) {
          // Create a new entry in the local cache for the CAS received from the remote
          cse = getController().getLocalCache().createCasStateEntry(casReferenceId);
          // Check if this CAS is a child
          if (aMessageContext.propertyExists(AsynchAEMessage.CasSequence)) {
            cse.setInputCasReferenceId(inputCasReferenceId);
          }
        } else {
          cse = getController().getLocalCache().lookupEntry(casReferenceId);
        }

        if (getController() instanceof AggregateAnalysisEngineController
                && aMessageContext.propertyExists(AsynchAEMessage.CasSequence)) {
          String delegateInputQueueName = aMessageContext
                  .getMessageStringProperty(AsynchAEMessage.MessageFrom);
          String delegateKey = ((AggregateAnalysisEngineController) getController())
                  .lookUpDelegateKey(delegateInputQueueName); // aMessageContext.getEndpoint().getEndpoint());
          if (delegateKey != null) {
            Delegate delegate = ((AggregateAnalysisEngineController) getController())
                    .lookupDelegate(delegateKey);
            // Save the last delegate handling this CAS
            cse.setLastDelegate(delegate);
            // If there is one thread receiving messages from Cas Multiplier increment number of
            // child Cases
            // of the parent CAS. If there are more threads (consumers) a special object
            // ConcurrentMessageListener
            // has already incremented the count. This special object enforces order of processing
            // for CASes
            // coming in from the Cas Multiplier.
            if (!delegate.hasConcurrentConsumersOnReplyQueue()) {
              inputCasStateEntry.incrementSubordinateCasInPlayCount();
            }
          }
        }

        entry = deserializeCASandRegisterWithCache(casReferenceId, freeCasEndpoint,
                newCASProducedBy, aMessageContext);
        if (getController().isStopped() || entry == null || entry.getCas() == null) {
          if (entry != null) {
            // The Controller is in shutdown state, release the CAS
            getController().dropCAS(entry.getCasReferenceId(), true);
            entry = null;
          }
          return;
        }
        // *****************************************************************
        // Process the CAS
        // *****************************************************************
        invokeProcess(entry.getCas(), inputCasReferenceId, casReferenceId, aMessageContext,
                newCASProducedBy);
      } else {
        if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) {
          UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(),
                  "handleProcessRequestFromRemoteClient", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
                  "UIMAEE_duplicate_request__INFO", new Object[] { casReferenceId });
        }
      }
    } catch (Exception e) {
      if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) {
        UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, getClass().getName(),
                "handleProcessRequestFromRemoteClient", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
                "UIMAEE_exception__WARNING", e);
      }
      ErrorContext errorContext = new ErrorContext();
      errorContext.add(AsynchAEMessage.Endpoint, aMessageContext.getEndpoint());
      errorContext.add(AsynchAEMessage.Command, AsynchAEMessage.Process);
      errorContext.add(AsynchAEMessage.CasReference, casReferenceId);
      if (entry != null) {
        getController().dropCAS(entry.getCas());
      }
      getController().getErrorHandlerChain().handle(e, errorContext, getController());
    }

  }

  private void handleProcessRequestWithCASReference(MessageContext aMessageContext)
          throws AsynchAEException {
    boolean isNewCAS = false;
    String newCASProducedBy = null;

    try {
      // This is only used when handling CASes produced by CAS Multiplier
      String inputCasReferenceId = null;
      CAS cas = null;
      CasStateEntry cse = null;
      String casReferenceId = getCasReferenceId(aMessageContext);
      if ((cse = getController().getLocalCache().lookupEntry(casReferenceId)) == null) {
        // Create a new entry in the local cache for the CAS received from the remote
        cse = getController().getLocalCache().createCasStateEntry(casReferenceId);
      }

      // Check if this Cas has been sent from a Cas Multiplier. If so, its sequence will be > 0
      if (aMessageContext.propertyExists(AsynchAEMessage.CasSequence)) {
        isNewCAS = true;

        Endpoint casMultiplierEndpoint = aMessageContext.getEndpoint();

        if (casMultiplierEndpoint == null) {
          if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) {
            UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(),
                    "handleProcessRequestWithCASReference",
                    UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_no_endpoint_for_reply__INFO",
                    new Object[] { casReferenceId });
          }
          return;
        }
        // Get the id of the parent Cas
        inputCasReferenceId = aMessageContext
                .getMessageStringProperty(AsynchAEMessage.InputCasReference);
        if (cse.getInputCasReferenceId() == null) {
          cse.setInputCasReferenceId(inputCasReferenceId);
        }

        if (getController() instanceof AggregateAnalysisEngineController) {
          CasStateEntry parentCasEntry = getController().getLocalCache().lookupEntry(
                  inputCasReferenceId);
          // Check if the parent CAS is in a failed state first
          if (parentCasEntry != null && parentCasEntry.isFailed()) {
            // handle CAS release
            getController().process(null, casReferenceId);
            return;
          }

          String delegateKey = ((AggregateAnalysisEngineController) getController())
                  .lookUpDelegateKey(aMessageContext.getEndpoint().getEndpoint());
          Delegate delegate = ((AggregateAnalysisEngineController) getController())
                  .lookupDelegate(delegateKey);
          cse.setLastDelegate(delegate);
          newCASProducedBy = delegate.getKey();
          casMultiplierEndpoint.setIsCasMultiplier(true);
          try {
            // Save the endpoint of the CM which produced the Cas
            getController().getInProcessCache().setCasProducer(casReferenceId, newCASProducedBy);
          } catch (Exception e) {
            if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) {
              UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(),
                      "handleProcessRequestWithCASReference",
                      UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_exception__WARNING",
                      new Object[] { e });
            }
            return;
          }
          // Safety check. The input Cas should not be null here
          if (inputCasReferenceId != null) {
            try {
              Endpoint endp = null;

              // Located the origin of the parent Cas. The produced Cas will inherit the origin from
              // its parent.
              // Once the origin is identified, save the origin using the produced Cas id as a key.
              if (endp == null) {
                boolean gotTheEndpoint = false;
                String parentCasId = inputCasReferenceId;
                // Loop through the parent tree until an origin is found
                while (!gotTheEndpoint) {
                  // Check if the current parent has an associated origin
                  endp = ((AggregateAnalysisEngineController) getController())
                          .getMessageOrigin(parentCasId);
                  // Check if there is an origin. If so, we are done
                  if (endp != null) {
                    break;
                  }
                  // The current parent has no origin, get its parent and try again
                  CacheEntry entry = getController().getInProcessCache().getCacheEntryForCAS(
                          parentCasId);
                  parentCasId = entry.getInputCasReferenceId();
                  // Check if we reached the top of the hierarchy tree. If so, we have no origin.
                  // This should
                  // never be the case. Every Cas must have an origin
                  if (parentCasId == null) {
                    break;
                  }
                }
              }
              // If origin not found log it as this indicates an error
              if (endp == null) {
                System.out.println("Endpoint Not Found For Cas Id:" + inputCasReferenceId);
                if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) {
                  UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(),
                          "handleProcessRequestWithCASReference",
                          UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
                          "UIMAEE_msg_origin_not_found__INFO",
                          new Object[] { getController().getComponentName(), inputCasReferenceId });
                }
              } else {
                ((AggregateAnalysisEngineController) getController()).addMessageOrigin(
                        casReferenceId, endp);
                if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINEST)) {
                  UIMAFramework.getLogger(CLASS_NAME).logrb(
                          Level.FINEST,
                          CLASS_NAME.getName(),
                          "handleProcessRequestWithCASReference",
                          UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
                          "UIMAEE_msg_origin_added__FINEST",
                          new Object[] { getController().getComponentName(), casReferenceId,
                              newCASProducedBy });
                }
              }
            } catch (Exception e) {
             
              if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) {
                UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, getClass().getName(),
                        "handleProcessRequestWithCASReference",
                        UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_exception__WARNING", e);
              }
            }
          } else {
            if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) {
              UIMAFramework.getLogger(CLASS_NAME).logrb(
                      Level.INFO,
                      CLASS_NAME.getName(),
                      "handleProcessRequestWithCASReference",
                      UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
                      "UIMAEE_input_cas_invalid__INFO",
                      new Object[] { getController().getComponentName(), newCASProducedBy,
                          casReferenceId });
            }
          }
        }
        if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINE)) {
          UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, CLASS_NAME.getName(),
                  "handleProcessRequestWithCASReference", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
                  "UIMAEE_new_cas__FINE", new Object[] { casReferenceId, newCASProducedBy });
        }
        aMessageContext.getEndpoint().setEndpoint(casMultiplierEndpoint.getEndpoint());
        aMessageContext.getEndpoint().setServerURI(casMultiplierEndpoint.getServerURI());
      } else {
        if (getController() instanceof AggregateAnalysisEngineController) {
          ((AggregateAnalysisEngineController) getController()).addMessageOrigin(casReferenceId,
                  aMessageContext.getEndpoint());
        }

      }
      cas = getController().getInProcessCache().getCasByReference(casReferenceId);

      long arrivalTime = System.nanoTime();
      getController().saveTime(arrivalTime, casReferenceId, getController().getName());// aMessageContext.getEndpointName());

      if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINE)) {
        UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, CLASS_NAME.getName(),
                "handleProcessRequestWithCASReference", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
                "UIMAEE_analyzing_cas__FINE", new Object[] { casReferenceId });
      }
      // Save Process command in the client endpoint.
      cacheProcessCommandInClientEndpoint();

      if (getController().isStopped()) {
        return;
      }

      if (isNewCAS) {
        invokeProcess(cas, inputCasReferenceId, casReferenceId, aMessageContext, newCASProducedBy);
      } else {
        invokeProcess(cas, casReferenceId, null, aMessageContext, newCASProducedBy);
      }
    } catch (AsynchAEException e) {
      throw e;
    } catch (Exception e) {
      throw new AsynchAEException(e);
    }

  }

  private void handleProcessRequestWithXCAS(MessageContext aMessageContext)
          throws AsynchAEException {

    try {
      // Get the CAS Reference Id of the input CAS
      String casReferenceId = getCasReferenceId(aMessageContext);
      String inputCasReferenceId = casReferenceId;
      // This is only used when handling CASes produced by CAS Multiplier
      String newCASProducedBy = null;

      if (aMessageContext.propertyExists(AsynchAEMessage.CasSequence)) {
        // This CAS came in from the CAS Multiplier. Treat it differently than the
        // input CAS. First, in case the Aggregate needs to send this CAS to the
        // client, retrieve the client destination by looking up the client endpoint
        // using input CAS reference id. CASes generated by the CAS multiplier will have
        // the same Cas Reference id.
        Endpoint replyToEndpoint = getController().getInProcessCache().getCacheEntryForCAS(
                casReferenceId).getMessageOrigin();

        //	
        if (getController() instanceof AggregateAnalysisEngineController) {
          newCASProducedBy = ((AggregateAnalysisEngineController) getController())
                  .lookUpDelegateKey(replyToEndpoint.getEndpoint());
        }
        // MessageContext contains endpoint set by the CAS Multiplier service. Overwrite
        // this with the endpoint of the client who sent the input CAS. In case this
        // aggregate is configured to send new CASes to the client we know where to send them.
        aMessageContext.getEndpoint().setEndpoint(replyToEndpoint.getEndpoint());
        aMessageContext.getEndpoint().setServerURI(replyToEndpoint.getServerURI());
        inputCasReferenceId = String.valueOf(casReferenceId);
        // Set this to null so that the new CAS gets its own Cas Reference Id below
        casReferenceId = null;
      }

      long arrivalTime = System.nanoTime();
      getController().saveTime(arrivalTime, casReferenceId, getController().getName());// aMessageContext.getEndpointName());

      // To prevent processing multiple messages with the same CasReferenceId, check the CAS cache
      // to see if the message with a given CasReferenceId is already being processed. It is, the
      // message contains the same request possibly issued by the caller due to timeout. Also this
      // mechanism helps with dealing with scenario when this service is not up when the client
      // sends
      // request. The client can keep re-sending the same request until its timeout thresholds are
      // exceeded. By that time, there may be multiple messages in this service queue with the same
      // CasReferenceId. When the service finally comes back up, it will have multiple messages in
      // its queue possibly from the same client. Only the first message for any given
      // CasReferenceId
      // should be processed.
      if (casReferenceId == null
              || !getController().getInProcessCache().entryExists(casReferenceId)) {
        String xmi = aMessageContext.getStringMessage();

        // *****************************************************************
        // ***** NO XMI In Message. Kick this back to sender with exception
        // *****************************************************************
        if (xmi == null) {
          if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) {
            UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(),
                    "handleProcessRequestWithXCAS", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
                    "UIMAEE_message_has_no_cargo__INFO",
                    new Object[] { aMessageContext.getEndpoint().getEndpoint() });
          }
          getController().getOutputChannel().sendReply(
                  new InvalidMessageException("No XMI data in message"), casReferenceId, null,
                  aMessageContext.getEndpoint(), AsynchAEMessage.Process);
          // Dont process this empty message
          return;
        }

        if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINE)) {
          UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, CLASS_NAME.getName(),
                  "handleProcessRequestWithXCAS", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
                  "UIMAEE_request_cas__FINE",
                  new Object[] { aMessageContext.getEndpoint().getEndpoint() });
        }
        CAS cas = getController().getCasManagerWrapper().getNewCas();

        if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINE)) {
          UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, CLASS_NAME.getName(),
                  "handleProcessRequestWithXCAS", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
                  "UIMAEE_request_cas_granted__FINE",
                  new Object[] { aMessageContext.getEndpoint().getEndpoint() });
        }
        XmiSerializationSharedData deserSharedData = new XmiSerializationSharedData();
        // UimaSerializer.deserializeCasFromXmi(xmi, cas, deserSharedData, true, -1);
        uimaSerializer.deserializeCasFromXmi(xmi, cas, deserSharedData, true, -1);

        if (casReferenceId == null) {
          CacheEntry entry = getController().getInProcessCache().register(cas, aMessageContext,
                  deserSharedData);
          casReferenceId = entry.getCasReferenceId();
        } else {
          if (getController() instanceof PrimitiveAnalysisEngineController) {
            getController().getInProcessCache().register(cas, aMessageContext, deserSharedData,
                    casReferenceId);
          }
        }
        if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINE)) {
          UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, CLASS_NAME.getName(),
                  "handleProcessRequestWithXCAS", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
                  "UIMAEE_deserialized_cas_ready_to_process_FINE",
                  new Object[] { aMessageContext.getEndpoint().getEndpoint() });
        }
        cacheProcessCommandInClientEndpoint();

        invokeProcess(cas, inputCasReferenceId, casReferenceId, aMessageContext, newCASProducedBy);
      } else {
        if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) {
          UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(),
                  "handleProcessRequestWithXCAS", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
                  "UIMAEE_duplicate_request__INFO", new Object[] { casReferenceId });
        }
      }

    } catch (AsynchAEException e) {
      throw e;
    } catch (Exception e) {
      throw new AsynchAEException(e);
    }

  }

  private void cacheProcessCommandInClientEndpoint() {
    Endpoint clientEndpoint = getController().getClientEndpoint();
    if (clientEndpoint != null) {
      clientEndpoint.setCommand(AsynchAEMessage.Process);
    }
  }

  private void handleCollectionProcessCompleteRequest(MessageContext aMessageContext)
          throws AsynchAEException {
    Endpoint replyToEndpoint = aMessageContext.getEndpoint();
    getController().collectionProcessComplete(replyToEndpoint);
  }

  private void handleReleaseCASRequest(MessageContext aMessageContext) throws AsynchAEException {
    String casReferenceId = aMessageContext.getMessageStringProperty(AsynchAEMessage.CasReference);
    if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINE)) {
      UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, CLASS_NAME.getName(),
              "handleReleaseCASRequest", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
              "UIMAEE_release_cas_req__FINE",
              new Object[] { getController().getName(), casReferenceId });
    }
    getController().releaseNextCas(casReferenceId);
  }

  private void handlePingRequest(MessageContext aMessageContext) {
    try {
      getController().getOutputChannel().sendReply(AsynchAEMessage.Ping,
              aMessageContext.getEndpoint());
    } catch (Exception e) {
      if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) {
        UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, getClass().getName(),
                "handlePingRequest", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
                "UIMAEE_exception__WARNING", new Object[] { e });
      }
    }

  }

  private void handleStopRequest(MessageContext aMessageContext) {
    try {
      String casReferenceId = aMessageContext
              .getMessageStringProperty(AsynchAEMessage.CasReference);
      System.out.println("###################Controller::" + getController().getComponentName()
              + " Received <<<STOP>>> Request For CAS:" + casReferenceId);
      if (getController() instanceof PrimitiveAnalysisEngineController) {
        getController().addAbortedCasReferenceId(casReferenceId);
      } else if (getController() instanceof AggregateAnalysisEngineController_impl) {
        try {
          CasStateEntry casStateEntry = getController().getLocalCache().lookupEntry(casReferenceId);
          // Mark the CAS as if it have failed. In this case we dont associate any
          // exceptions with this CAS so its really not a failure of a CAS or any
          // of its children. We simply use the same logic here as if the CAS failed.
          // The Aggregate replyToClient() method will know that this CAS was stopped
          // as opposed to failed by the fact that the CAS has no exceptions associated
          // with it. In such case the replyToClient() method returns an input CAS as if
          // it has been fully processed.
          casStateEntry.setFailed();
          ((AggregateAnalysisEngineController_impl) getController()).stopCasMultipliers();
        } catch (Exception ex) {
        } // CAS may have already been deleted

      }

    } catch (Exception e) {
    }
  }

  /**
   * Main method called by the predecessor handler.
   * 
   * 
   */
  public void handle(Object anObjectToHandle) // throws AsynchAEException
  {
    try {
      super.validate(anObjectToHandle);

      MessageContext messageContext = (MessageContext) anObjectToHandle;
      if (isHandlerForMessage(messageContext, AsynchAEMessage.Request, AsynchAEMessage.Process)
              || isHandlerForMessage(messageContext, AsynchAEMessage.Request,
                      AsynchAEMessage.CollectionProcessComplete)
              || isHandlerForMessage(messageContext, AsynchAEMessage.Request,
                      AsynchAEMessage.ReleaseCAS)
              || isHandlerForMessage(messageContext, AsynchAEMessage.Request, AsynchAEMessage.Stop)) {
        int payload = messageContext.getMessageIntProperty(AsynchAEMessage.Payload);
        int command = messageContext.getMessageIntProperty(AsynchAEMessage.Command);

        getController().getControllerLatch().waitUntilInitialized();

        // If a Process Request, increment number of CASes processed
        if (messageContext.getMessageIntProperty(AsynchAEMessage.MessageType) == AsynchAEMessage.Request
                && command == AsynchAEMessage.Process
                && !messageContext.propertyExists(AsynchAEMessage.CasSequence)) {
          // Increment number of CASes processed by this service
          getController().getServicePerformance().incrementNumberOfCASesProcessed();
        }
        if (getController().isStopped()) {
          return;
        }

        if (AsynchAEMessage.CASRefID == payload) {
          // Fetch id of the CAS from the message.
          if (getCasReferenceId(messageContext) == null) {
            return; // Invalid message. Nothing to do
          }

          handleProcessRequestWithCASReference(messageContext);
        } else if (AsynchAEMessage.XMIPayload == payload
                || AsynchAEMessage.BinaryPayload == payload) {
          // Fetch id of the CAS from the message.
          if (getCasReferenceId(messageContext) == null) {
            return; // Invalid message. Nothing to do
          }
          handleProcessRequestFromRemoteClient(messageContext);
        } else if (AsynchAEMessage.XCASPayload == payload) {
          // Fetch id of the CAS from the message.
          if (getCasReferenceId(messageContext) == null) {
            return; // Invalid message. Nothing to do
          }
          handleProcessRequestWithXCAS(messageContext);
        } else if (AsynchAEMessage.None == payload
                && AsynchAEMessage.CollectionProcessComplete == command) {
          handleCollectionProcessCompleteRequest(messageContext);
        } else if (AsynchAEMessage.None == payload && AsynchAEMessage.ReleaseCAS == command) {
          handleReleaseCASRequest(messageContext);
        } else if (AsynchAEMessage.None == payload && AsynchAEMessage.Stop == command) {
          handleStopRequest(messageContext);
        } else if (AsynchAEMessage.None == payload && AsynchAEMessage.Ping == command) {
          handlePingRequest(messageContext);
        }
        // Handled Request
        return;
      }
      // Not a Request nor Command. Delegate to the next handler in the chain
      super.delegate(messageContext);
    } catch (Exception e) {
      if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) {
        UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, getClass().getName(),
                "handle", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
                "UIMAEE_exception__WARNING", new Object[] { e });
      }
      getController().getErrorHandlerChain().handle(e,
              HandlerBase.populateErrorContext((MessageContext) anObjectToHandle), getController());
    }
  }

}