/*
 * 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.controller;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Semaphore;

import org.apache.uima.UIMAFramework;
import org.apache.uima.aae.AsynchAECasManager;
import org.apache.uima.aae.InProcessCache;
import org.apache.uima.aae.UIMAEE_Constants;
import org.apache.uima.aae.UimaClassFactory;
import org.apache.uima.aae.InProcessCache.CacheEntry;
import org.apache.uima.aae.controller.LocalCache.CasStateEntry;
import org.apache.uima.aae.error.AsynchAEException;
import org.apache.uima.aae.error.ErrorContext;
import org.apache.uima.aae.error.ErrorHandler;
import org.apache.uima.aae.error.ServiceShutdownException;
import org.apache.uima.aae.jmx.JmxManagement;
import org.apache.uima.aae.jmx.PrimitiveServiceInfo;
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.spi.transport.UimaMessage;
import org.apache.uima.aae.spi.transport.UimaTransport;
import org.apache.uima.analysis_engine.AnalysisEngine;
import org.apache.uima.analysis_engine.AnalysisEngineDescription;
import org.apache.uima.analysis_engine.CasIterator;
import org.apache.uima.analysis_engine.metadata.AnalysisEngineMetaData;
import org.apache.uima.cas.CAS;
import org.apache.uima.cas.impl.CASImpl;
import org.apache.uima.cas.impl.OutOfTypeSystemData;
import org.apache.uima.resource.ResourceInitializationException;
import org.apache.uima.resource.ResourceProcessException;
import org.apache.uima.resource.ResourceSpecifier;
import org.apache.uima.resource.metadata.ConfigurationParameter;
import org.apache.uima.resource.metadata.impl.ConfigurationParameter_impl;
import org.apache.uima.util.InvalidXMLException;
import org.apache.uima.util.Level;

public class PrimitiveAnalysisEngineController_impl extends BaseAnalysisEngineController implements
        PrimitiveAnalysisEngineController {
  private static final Class CLASS_NAME = PrimitiveAnalysisEngineController_impl.class;

  // Stores AE metadata
  private AnalysisEngineMetaData analysisEngineMetadata;

  // Number of AE instances
  private int analysisEnginePoolSize;

  // Mutex
  protected Object notifyObj = new Object();

  // Temp list holding instances of AE
  private List aeList = new ArrayList();

  // Stores service info for JMX
  private PrimitiveServiceInfo serviceInfo = null;

  // Pool containing instances of AE. The default implementation provides Thread affinity
  // meaning each thread executes the same AE instance.
  private AnalysisEngineInstancePool aeInstancePool = null;

  private String abortedCASReferenceId = null;
  // Create a shared semaphore to serialize creation of AE instances.
  // There is a single instance of this semaphore per JVM and it
  // guards uima core code that is not thread safe.
  private static Semaphore sharedInitSemaphore = new Semaphore(1);

  public PrimitiveAnalysisEngineController_impl(String anEndpointName,
          String anAnalysisEngineDescriptor, AsynchAECasManager aCasManager,
          InProcessCache anInProcessCache, int aWorkQueueSize, int anAnalysisEnginePoolSize)
          throws Exception {
    this(null, anEndpointName, anAnalysisEngineDescriptor, aCasManager, anInProcessCache,
            aWorkQueueSize, anAnalysisEnginePoolSize, 0);
  }

  public PrimitiveAnalysisEngineController_impl(AnalysisEngineController aParentController,
          String anEndpointName, String anAnalysisEngineDescriptor, AsynchAECasManager aCasManager,
          InProcessCache anInProcessCache, int aWorkQueueSize, int anAnalysisEnginePoolSize)
          throws Exception {
    this(aParentController, anEndpointName, anAnalysisEngineDescriptor, aCasManager,
            anInProcessCache, aWorkQueueSize, anAnalysisEnginePoolSize, 0);
  }

  public PrimitiveAnalysisEngineController_impl(AnalysisEngineController aParentController,
          String anEndpointName, String anAnalysisEngineDescriptor, AsynchAECasManager aCasManager,
          InProcessCache anInProcessCache, int aWorkQueueSize, int anAnalysisEnginePoolSize,
          int aComponentCasPoolSize) throws Exception {
    this(aParentController, anEndpointName, anAnalysisEngineDescriptor, aCasManager,
            anInProcessCache, aWorkQueueSize, anAnalysisEnginePoolSize, aComponentCasPoolSize, null);
  }

  public PrimitiveAnalysisEngineController_impl(AnalysisEngineController aParentController,
          String anEndpointName, String anAnalysisEngineDescriptor, AsynchAECasManager aCasManager,
          InProcessCache anInProcessCache, int aWorkQueueSize, int anAnalysisEnginePoolSize,
          int aComponentCasPoolSize, long anInitialCasHeapSize) throws Exception {
    this(aParentController, anEndpointName, anAnalysisEngineDescriptor, aCasManager,
            anInProcessCache, aWorkQueueSize, anAnalysisEnginePoolSize, aComponentCasPoolSize,
            anInitialCasHeapSize, null);
  }

  public PrimitiveAnalysisEngineController_impl(AnalysisEngineController aParentController,
          String anEndpointName, String anAnalysisEngineDescriptor, AsynchAECasManager aCasManager,
          InProcessCache anInProcessCache, int aWorkQueueSize, int anAnalysisEnginePoolSize,
          int aComponentCasPoolSize, JmxManagement aJmxManagement) throws Exception {
    this(aParentController, anEndpointName, anAnalysisEngineDescriptor, aCasManager,
            anInProcessCache, aWorkQueueSize, anAnalysisEnginePoolSize, aComponentCasPoolSize, 0,
            aJmxManagement);
  }

  public PrimitiveAnalysisEngineController_impl(AnalysisEngineController aParentController,
          String anEndpointName, String anAnalysisEngineDescriptor, AsynchAECasManager aCasManager,
          InProcessCache anInProcessCache, int aWorkQueueSize, int anAnalysisEnginePoolSize,
          int aComponentCasPoolSize, long anInitialCasHeapSize, JmxManagement aJmxManagement)
          throws Exception {
    super(aParentController, aComponentCasPoolSize, anInitialCasHeapSize, anEndpointName,
            anAnalysisEngineDescriptor, aCasManager, anInProcessCache, null, aJmxManagement);
    analysisEnginePoolSize = anAnalysisEnginePoolSize;
  }

  public PrimitiveAnalysisEngineController_impl(AnalysisEngineController aParentController,
          String anEndpointName, String anAnalysisEngineDescriptor, AsynchAECasManager aCasManager,
          InProcessCache anInProcessCache, int aWorkQueueSize, int anAnalysisEnginePoolSize,
          JmxManagement aJmxManagement) throws Exception {
    this(aParentController, anEndpointName, anAnalysisEngineDescriptor, aCasManager,
            anInProcessCache, aWorkQueueSize, anAnalysisEnginePoolSize, 0, aJmxManagement);
  }

  public int getAEInstanceCount() {
    return analysisEnginePoolSize;
  }

  public void initializeAnalysisEngine() throws ResourceInitializationException {
    ResourceSpecifier rSpecifier = null;
    try {
      //  Acquire single-permit semaphore to serialize instantiation of AEs.
      //  This is done to control access to non-thread safe structures in the
      //  core. The sharedInitSemaphore is a static and is shared by all instances
      //  of this class. 
      sharedInitSemaphore.acquire();
      // Parse the descriptor in the calling thread.
      rSpecifier = UimaClassFactory.produceResourceSpecifier(super.aeDescriptor);
        AnalysisEngine ae = UIMAFramework.produceAnalysisEngine(rSpecifier, paramsMap);
        if (aeInstancePool == null) {
          aeInstancePool = new AnalysisEngineInstancePoolWithThreadAffinity(analysisEnginePoolSize);
        }
        if (analysisEngineMetadata == null) {
          analysisEngineMetadata = ae.getAnalysisEngineMetaData();
        }
        //  Check if OperationalProperties allow replication of this AE. Throw exception if
        //  the deployment descriptor says to scale the service *but* the AE descriptor's 
        //  OperationalProperties disallow it.
        if ( !analysisEngineMetadata.getOperationalProperties().isMultipleDeploymentAllowed() &&
             aeInstancePool.size() >= 1 ) {
          throw new ResourceInitializationException( UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
                  "UIMAEE_multiple_deployment_not_allowed__WARNING", new Object[] {this.getComponentName(), ae.getMetaData().getName()});
        }
        aeInstancePool.checkin(ae);
        if (aeInstancePool.size() == analysisEnginePoolSize) {
          try {
            postInitialize();
          } catch (Exception e) {
            if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) {
              UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, getClass().getName(),
                      "initializeAnalysisEngine", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
                      "UIMAEE_exception__WARNING", new Object[] { e });
            }
            throw new ResourceInitializationException(e);
          }
        }
    } catch (Exception e) {
      if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) {
        UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, getClass().getName(),
                "initializeAnalysisEngine", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
                "UIMAEE_exception__WARNING", new Object[] { e });
      }
      super.notifyListenersWithInitializationStatus(e);
      if (isTopLevelComponent()) {
        super.notifyListenersWithInitializationStatus(e);
      } else {
        // get the top level controller to notify
        AnalysisEngineController controller = this.getParentController();
        while (!controller.isTopLevelComponent()) {
          controller = controller.getParentController();
        }
        getParentController().notifyListenersWithInitializationStatus(e);
      }

      throw new ResourceInitializationException(e);
    } finally {
      //  Release the shared semaphore so that another instance of this class can instantiate
      //  an Analysis Engine.
      sharedInitSemaphore.release();
    }
    
  }

  public boolean threadAssignedToAE() {
    if (aeInstancePool == null) {
      return false;
    }

    return aeInstancePool.exists();
  }

  public void initialize() throws AsynchAEException {
  }

  /**
   * This method is called after all AE instances initialize. It is called once. It initializes
   * service Cas Pool, notifies the deployer that initialization completed and finally loweres a
   * semaphore allowing messages to be processed.
   * 
   * @throws AsynchAEException
   */
  private void postInitialize() throws AsynchAEException {
    try {
      if (errorHandlerChain == null) {
        super.plugInDefaultErrorHandlerChain();
      }
      if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.CONFIG)) {
        UIMAFramework.getLogger(CLASS_NAME).logrb(Level.CONFIG, getClass().getName(), "initialize",
                UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
                "UIMAEE_primitive_ctrl_init_info__CONFIG", new Object[] { analysisEnginePoolSize });
      }

      if (serviceInfo == null) {
        serviceInfo = new PrimitiveServiceInfo(isCasMultiplier(), this);
      }
      serviceInfo.setServiceKey(delegateKey);
      serviceInfo.setAnalysisEngineInstanceCount(analysisEnginePoolSize);

      if (!isStopped()) {
        getMonitor().setThresholds(getErrorHandlerChain().getThresholds());
        // Initialize Cas Manager
        if (getCasManagerWrapper() != null) {
          try {
            if (getCasManagerWrapper().isInitialized()) {
              getCasManagerWrapper().addMetadata(getAnalysisEngineMetadata());
              if (isTopLevelComponent()) {
                getCasManagerWrapper().initialize("PrimitiveAEService");
                CAS cas = getCasManagerWrapper().getNewCas("PrimitiveAEService");
                cas.release();
              }
            }
            if (isTopLevelComponent()) {
              super.notifyListenersWithInitializationStatus(null);
            }

            // All internal components of this Primitive have been initialized. Open the latch
            // so that this service can start processing requests.
            latch.openLatch(getName(), isTopLevelComponent(), true);

          } catch (Exception e) {
            if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) {
              UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, getClass().getName(),
                      "postInitialize", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
                      "UIMAEE_exception__WARNING", new Object[] { e });
            }
            throw new AsynchAEException(e);
          }
        } else {
          if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.CONFIG)) {
            UIMAFramework.getLogger(CLASS_NAME).logrb(Level.CONFIG, getClass().getName(),
                    "postInitialize", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
                    "UIMAEE_cas_manager_wrapper_notdefined__CONFIG", new Object[] {});
          }
        }
      }
      ((BaseAnalysisEngineController) this).startServiceCleanupThread(30000); // sleep for 30 secs
    } catch (AsynchAEException e) {
      if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) {
        UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, getClass().getName(),
                "postInitialize", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
                "UIMAEE_exception__WARNING", new Object[] { e });
      }
      throw e;
    } catch (Exception e) {
      if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) {
        UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, getClass().getName(),
                "postInitialize", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
                "UIMAEE_exception__WARNING", new Object[] { e });
      }
      throw new AsynchAEException(e);
    }

    if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) {
      UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(), "postInitialize",
              UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_initialized_controller__INFO",
              new Object[] { getComponentName() });
    }
    super.serviceInitialized = true;
  }

  /**
   * Forces initialization of a Cas Pool if this is a Cas Multiplier delegate collocated with an
   * aggregate. The parent aggregate calls this method when all type systems have been merged.
   */
  public synchronized void onInitialize() {
    // Component's Cas Pool is registered lazily, when the process() is called for
    // the first time. For monitoring purposes, we need the comoponent's Cas Pool
    // MBeans to register during initialization of the service. For a Cas Multiplier
    // force creation of the Cas Pool and registration of a Cas Pool with the JMX Server.
    // Just get the CAS and release it back to the component's Cas Pool.
    if (isCasMultiplier() && !isTopLevelComponent()) {
      CAS cas = (CAS) getUimaContext().getEmptyCas(CAS.class);
      cas.release();
    }
  }

  /**
	 * 
	 */
  public void collectionProcessComplete(Endpoint anEndpoint)// throws AsynchAEException
  {
    AnalysisEngine ae = null;
    long start = super.getCpuTime();
    localCache.dumpContents();
    try {
      ae = aeInstancePool.checkout();
      if (ae != null) {
        ae.collectionProcessComplete();
      }
      if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINEST)) {
        UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINEST, getClass().getName(),
                "collectionProcessComplete", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
                "UIMAEE_cpc_all_cases_processed__FINEST", new Object[] { getComponentName() });
      }
      getServicePerformance().incrementAnalysisTime(super.getCpuTime() - start);
      if (!anEndpoint.isRemote()) {
        UimaTransport transport = getTransport(anEndpoint.getEndpoint());
        UimaMessage message = transport.produceMessage(AsynchAEMessage.CollectionProcessComplete,
                AsynchAEMessage.Response, getName());
        // Send CPC completion reply back to the client. Use internal (non-jms) transport
        transport.getUimaMessageDispatcher(anEndpoint.getEndpoint()).dispatch(message);
      } else {
        getOutputChannel().sendReply(AsynchAEMessage.CollectionProcessComplete, anEndpoint);
      }

      if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINE)) {
        UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, getClass().getName(),
                "collectionProcessComplete", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
                "UIMAEE_cpc_completed__FINE", new Object[] { getComponentName() });
      }

    } catch (Exception e) {
      ErrorContext errorContext = new ErrorContext();
      errorContext.add(AsynchAEMessage.Command, AsynchAEMessage.CollectionProcessComplete);
      errorContext.add(AsynchAEMessage.Endpoint, anEndpoint);
      getErrorHandlerChain().handle(e, errorContext, this);
    } finally {
      clearStats();
      if (ae != null) {
        try {
          aeInstancePool.checkin(ae);
        } catch (Exception ex) {
          if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) {
            UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, getClass().getName(),
                    "collectionProcessComplete", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
                    "UIMAEE_unable_to_check_ae_back_to_pool__WARNING",
                    new Object[] { getComponentName(), ex });
          }
        }
      }
    }
  }

  /**
   * This is called when a Stop request is received from a client. Add the provided Cas id to the
   * list of aborted CASes. The process() method checks this list to determine if it should continue
   * generating children.
   * 
   * @param aCasReferenceId
   *          - Id of an input CAS. The client wants to stop generation of child CASes from this
   *          CAS.
   * 
   * @return
   */

  public void process(CAS aCAS, String aCasReferenceId, Endpoint anEndpoint) {

    if (stopped) {
      return;
    }
    CasStateEntry parentCasStateEntry = null;
    try {
      parentCasStateEntry = getLocalCache().lookupEntry(aCasReferenceId);
    } catch (Exception e) {
      if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) {
        UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, getClass().getName(),
                "process", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
                "UIMAEE_exception__WARNING", new Object[] { e });
      }
      return;
    }

    long totalProcessTime = 0; // stored total time spent producing ALL CASes

    boolean inputCASReturned = false;
    boolean processingFailed = false;
    // This is a primitive controller. No more processing is to be done on the Cas. Mark the
    // destination as final and return CAS in reply.
    anEndpoint.setFinal(true);
    AnalysisEngine ae = null;
    try {
      // Checkout an instance of AE from the pool
      ae = aeInstancePool.checkout();
      // Get input CAS entry from the InProcess cache
      long time = super.getCpuTime();
      CasIterator casIterator = ae.processAndOutputNewCASes(aCAS);
      // Store how long it took to call processAndOutputNewCASes()
      totalProcessTime = (super.getCpuTime() - time);
      long sequence = 1;
      long hasNextTime = 0; // stores time in hasNext()
      long getNextTime = 0; // stores time in next();
      boolean moreCASesToProcess = true;
      boolean casAbortedDueToExternalRequest = false;
      while (moreCASesToProcess) {
        long timeToProcessCAS = 0; // stores time in hasNext() and next() for each CAS
        hasNextTime = super.getCpuTime();
        if (!casIterator.hasNext()) {
          moreCASesToProcess = false;
          // Measure how long it took to call hasNext()
          timeToProcessCAS = (super.getCpuTime() - hasNextTime);
          totalProcessTime += timeToProcessCAS;
          break; // from while
        }
        // Measure how long it took to call hasNext()
        timeToProcessCAS = (super.getCpuTime() - hasNextTime);
        getNextTime = super.getCpuTime();
        CAS casProduced = casIterator.next();
        // Add how long it took to call next()
        timeToProcessCAS += (super.getCpuTime() - getNextTime);
        // Add time to call hasNext() and next() to the running total
        totalProcessTime += timeToProcessCAS;
        casAbortedDueToExternalRequest = abortGeneratingCASes(aCasReferenceId);
        // If the service is stopped or aborted, stop generating new CASes and just return the input
        // CAS
        if (stopped || casAbortedDueToExternalRequest) {
          if (getInProcessCache() != null && getInProcessCache().getSize() > 0
                  && getInProcessCache().entryExists(aCasReferenceId)) {
            try {
              // Set a flag on the input CAS to indicate that the processing was aborted
              getInProcessCache().getCacheEntryForCAS(aCasReferenceId).setAborted(true);
            } 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.
            } finally {
              // !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
              // We are terminating the iterator here, release the internal CAS lock
              // so that we can release the CAS. This approach may need to be changed
              // as there may potentially be a problem with a Class Loader.
              // !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
              ((CASImpl) aCAS).enableReset(true);
              try {
                // We are either stopping the service or aborting input CAS due to explicit STOP
                // request
                // from a client. If a new CAS was produced, release it back to the pool.
                if (casProduced != null) {
                  casProduced.release();
                }
              } catch (Exception e) {
                System.out.println("Controller:" + getComponentName()
                        + " Attempt to release CAS Failed");
              }
              if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) {
                UIMAFramework.getLogger(CLASS_NAME).logrb(
                        Level.INFO,
                        getClass().getName(),
                        "process",
                        UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
                        "UIMAEE_stopped_producing_new_cases__INFO",
                        new Object[] { Thread.currentThread().getId(), getComponentName(),
                            aCasReferenceId });
              }
              System.out.println(">>>> Cas Multiplier:" + getComponentName()
                      + " Stopped Generating CASes from Input CAS:" + aCasReferenceId);
            }
          }
          if (casAbortedDueToExternalRequest) {

            // The controller was told to stop generating new CASes. Just return the input CAS to
            // the
            // client
            // throw new ResourceProcessException(new
            // InterruptedException("Cas Multiplier:"+getComponentName()+" Aborted CAS:"+aCasReferenceId));

            break; // break out of the cas producing loop and return an input CAS to the client
          } else {
            // The controller is stopping
            return;
          }
        }
        OutOfTypeSystemData otsd = getInProcessCache().getOutOfTypeSystemData(aCasReferenceId);
        MessageContext mContext = getInProcessCache()
                .getMessageAccessorByReference(aCasReferenceId);
        CacheEntry newEntry = getInProcessCache().register(casProduced, mContext, otsd);
        // if this Cas Multiplier is not Top Level service, add new Cas Id to the private
        // cache of the parent aggregate controller. The Aggregate needs to know about
        // all CASes it has in play that were generated from the input CAS.
        CasStateEntry childCasStateEntry = null;
        if (!isTopLevelComponent()) {
          newEntry.setNewCas(true, parentController.getComponentName());
          // Create CAS state entry in the aggregate's local cache
          childCasStateEntry = parentController.getLocalCache().createCasStateEntry(
                  newEntry.getCasReferenceId());
          // Fetch the parent CAS state entry from the aggregate's local cache. We need to increment
          // number of child CASes associated with it.
          parentCasStateEntry = parentController.getLocalCache().lookupEntry(aCasReferenceId);
        } else {
          childCasStateEntry = getLocalCache().createCasStateEntry(newEntry.getCasReferenceId());
        }
        // Associate parent CAS (input CAS) with the new CAS.
        childCasStateEntry.setInputCasReferenceId(aCasReferenceId);
        // Increment number of child CASes generated from the input CAS
        parentCasStateEntry.incrementSubordinateCasInPlayCount();

        // Associate input CAS with the new CAS
        newEntry.setInputCasReferenceId(aCasReferenceId);
        newEntry.setCasSequence(sequence);
        // Add to the cache how long it took to process the generated (subordinate) CAS
        getCasStatistics(newEntry.getCasReferenceId()).incrementAnalysisTime(timeToProcessCAS);
        if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINE)) {
          UIMAFramework.getLogger(CLASS_NAME).logrb(
                  Level.FINE,
                  getClass().getName(),
                  "process",
                  UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
                  "UIMAEE_produced_new_cas__FINE",
                  new Object[] { Thread.currentThread().getName(),
                      getUimaContextAdmin().getQualifiedContextName(),
                      newEntry.getCasReferenceId(), aCasReferenceId });
        }
        // Add the generated CAS to the outstanding CAS Map. Client notification will release
        // this CAS back to its pool
        synchronized (syncObject) {
          if (isTopLevelComponent()) {
            // Add the id of the generated CAS to the map holding outstanding CASes. This
            // map will be referenced when a client sends Free CAS Notification. The map
            // stores the id of the CAS both as a key and a value. Map is used to facilitate
            // quick lookup
            cmOutstandingCASes.put(newEntry.getCasReferenceId(), newEntry.getCasReferenceId());
          }
          // Increment number of CASes processed by this service
          sequence++;
        }
        if (!anEndpoint.isRemote()) {
          UimaTransport transport = getTransport(anEndpoint.getEndpoint());
          UimaMessage message = transport.produceMessage(AsynchAEMessage.Process,
                  AsynchAEMessage.Request, getName());
          message.addStringProperty(AsynchAEMessage.CasReference, newEntry.getCasReferenceId());
          message.addStringProperty(AsynchAEMessage.InputCasReference, aCasReferenceId);
          message.addLongProperty(AsynchAEMessage.CasSequence, sequence);
          ServicePerformance casStats = getCasStatistics(aCasReferenceId);

          message.addLongProperty(AsynchAEMessage.TimeToSerializeCAS, casStats
                  .getRawCasSerializationTime());
          message.addLongProperty(AsynchAEMessage.TimeToDeserializeCAS, casStats
                  .getRawCasDeserializationTime());
          message.addLongProperty(AsynchAEMessage.TimeInProcessCAS, casStats.getRawAnalysisTime());
          long iT = getIdleTimeBetweenProcessCalls(AsynchAEMessage.Process);
          message.addLongProperty(AsynchAEMessage.IdleTime, iT);
          if (!stopped) {
            transport.getUimaMessageDispatcher(anEndpoint.getEndpoint()).dispatch(message);
          }
        } else {
          // Send generated CAS to the client
          if (!stopped) {
            getOutputChannel().sendReply(newEntry, anEndpoint);
          }
        }
        // Remove the new CAS state entry from the local cache if this a top level primitive.
        // If not top level, the client (an Aggregate) will remove this entry when this new
        // generated CAS reaches Final State.
        if (isTopLevelComponent()) {
          try {
            localCache.lookupEntry(newEntry.getCasReferenceId()).setDropped(true);
          } catch (Exception e) {
          }
          localCache.remove(newEntry.getCasReferenceId());
        }

        // Remove Stats from the global Map associated with the new CAS
        // These stats for this CAS were added to the response message
        // and are no longer needed
        dropCasStatistics(newEntry.getCasReferenceId());
      } // while

      if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINEST)) {
        UIMAFramework.getLogger(CLASS_NAME).logrb(
                Level.FINEST,
                getClass().getName(),
                "process",
                UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
                "UIMAEE_completed_analysis__FINEST",
                new Object[] { Thread.currentThread().getName(), getComponentName(),
                    aCasReferenceId, (double) (super.getCpuTime() - time) / (double) 1000000 });
      }
      getMonitor().resetCountingStatistic("", Monitor.ProcessErrorCount);
      // Set total number of children generated from this CAS
      // Store total time spent processing this input CAS
      getCasStatistics(aCasReferenceId).incrementAnalysisTime(totalProcessTime);
      if (!anEndpoint.isRemote()) {
        inputCASReturned = true;
        UimaTransport transport = getTransport(anEndpoint.getEndpoint());

        UimaMessage message = transport.produceMessage(AsynchAEMessage.Process,
                AsynchAEMessage.Response, getName());
        message.addStringProperty(AsynchAEMessage.CasReference, aCasReferenceId);
        ServicePerformance casStats = getCasStatistics(aCasReferenceId);

        message.addLongProperty(AsynchAEMessage.TimeToSerializeCAS, casStats
                .getRawCasSerializationTime());
        message.addLongProperty(AsynchAEMessage.TimeToDeserializeCAS, casStats
                .getRawCasDeserializationTime());
        message.addLongProperty(AsynchAEMessage.TimeInProcessCAS, casStats.getRawAnalysisTime());
        long iT = getIdleTimeBetweenProcessCalls(AsynchAEMessage.Process);
        message.addLongProperty(AsynchAEMessage.IdleTime, iT);
        // Send reply back to the client. Use internal (non-jms) transport
        if (!stopped) {
          transport.getUimaMessageDispatcher(anEndpoint.getEndpoint()).dispatch(message);
        }
      } else {
        if (!stopped) {
          getOutputChannel().sendReply(aCasReferenceId, anEndpoint);
        }

        inputCASReturned = true;
      }
      // Remove input CAS state entry from the local cache
      if (!isTopLevelComponent()) {
        localCache.lookupEntry(aCasReferenceId).setDropped(true);
        localCache.remove(aCasReferenceId);
      }
    } catch (Throwable e) {
      processingFailed = true;
      ErrorContext errorContext = new ErrorContext();
      errorContext.add(AsynchAEMessage.CasReference, aCasReferenceId);
      errorContext.add(AsynchAEMessage.Command, AsynchAEMessage.Process);
      errorContext.add(AsynchAEMessage.MessageType, AsynchAEMessage.Request);
      errorContext.add(AsynchAEMessage.Endpoint, anEndpoint);
      // Handle the exception. Pass reference to the PrimitiveController instance
      getErrorHandlerChain().handle(e, errorContext, this);
    } finally {
      dropCasStatistics(aCasReferenceId);

      if (ae != null) {
        try {
          aeInstancePool.checkin(ae);
        } catch (Exception e) {
          if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) {
            UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, getClass().getName(),
                    "process", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
                    "UIMAEE_exception__WARNING", new Object[] { e });
          }
        }
      }
      // drop the CAS if it has been successfully processed. If there was a failure, the Error
      // Handler
      // will drop the CAS
      if (isTopLevelComponent() && !processingFailed) {
        // Release CASes produced from the input CAS if the input CAS has been aborted
        if (abortGeneratingCASes(aCasReferenceId)) {

          if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) {
            UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(), "process",
                    UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_remove_cache_entry__INFO",
                    new Object[] { getComponentName(), aCasReferenceId });
          }
          getInProcessCache().releaseCASesProducedFromInputCAS(aCasReferenceId);
        } else if (inputCASReturned && isTopLevelComponent()) {
          // Remove input CAS cache entry if the CAS has been sent to the client
          dropCAS(aCasReferenceId, true);
          localCache.dumpContents();
        }
      }
    }
  }

  private void addConfigIntParameter(String aParamName, int aParamValue) {
    ConfigurationParameter cp = new ConfigurationParameter_impl();
    cp.setMandatory(false);
    cp.setMultiValued(false);
    cp.setName(aParamName);
    cp.setType("Integer");
    getAnalysisEngineMetadata().getConfigurationParameterDeclarations().addConfigurationParameter(
            cp);
    getAnalysisEngineMetadata().getConfigurationParameterSettings().setParameterValue(aParamName,
            aParamValue);

  }

  // Return metadata
  public void sendMetadata(Endpoint anEndpoint) throws AsynchAEException {
    addConfigIntParameter(AnalysisEngineController.AEInstanceCount, analysisEnginePoolSize);

    if (getAnalysisEngineMetadata().getOperationalProperties().getOutputsNewCASes()) {
      addConfigIntParameter(AnalysisEngineController.CasPoolSize, super.componentCasPoolSize);
    }
    super.sendMetadata(anEndpoint, getAnalysisEngineMetadata());
  }

  private AnalysisEngineMetaData getAnalysisEngineMetadata() {
    return analysisEngineMetadata;
  }

  /**
   * Executes action on error. Primitive Controller allows two types of actions TERMINATE and
   * DROPCAS.
   */
  public void takeAction(String anAction, String anEndpointName, ErrorContext anErrorContext) {
    try {
      if (ErrorHandler.TERMINATE.equalsIgnoreCase(anAction)
              || ErrorHandler.DROPCAS.equalsIgnoreCase(anAction)) {
        super.handleAction(anAction, anEndpointName, anErrorContext);
      }
    } catch (Exception e) {
      if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) {
        UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, getClass().getName(),
                "takeAction", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
                "UIMAEE_exception__WARNING", new Object[] { e });
      }
    }
  }

  public String getServiceEndpointName() {
    return getName();
  }

  public synchronized ControllerLatch getControllerLatch() {
    return latch;
  }

  public boolean isPrimitive() {
    return true;
  }

  public Monitor getMonitor() {
    return super.monitor;
  }

  public void setMonitor(Monitor monitor) {
    this.monitor = monitor;
  }

  public void handleDelegateLifeCycleEvent(String anEndpoint, int aDelegateCount) {
    if (aDelegateCount == 0) {
      // tbi
    }
  }

  protected String getNameFromMetadata() {
    return super.getMetaData().getName();
  }

  public void setAnalysisEngineInstancePool(AnalysisEngineInstancePool aPool) {
    aeInstancePool = aPool;
  }

  public PrimitiveServiceInfo getServiceInfo() {
    if (serviceInfo == null) {
      serviceInfo = new PrimitiveServiceInfo(isCasMultiplier(), this);
      serviceInfo.setServiceKey(delegateKey);
    }
    if (isTopLevelComponent() && getInputChannel() != null) {
      serviceInfo.setInputQueueName(getInputChannel().getServiceInfo().getInputQueueName());
      serviceInfo.setBrokerURL(super.getBrokerURL());
      serviceInfo.setDeploymentDescriptorPath(super.aeDescriptor);
    }

    return serviceInfo;
  }

  public void stop() {
    super.stop();
    if (aeInstancePool != null) {
      try {
        aeInstancePool.destroy();
      } catch (Exception e) {
        if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) {
          UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, getClass().getName(),
                  "stop", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
                  "UIMAEE_exception__WARNING", new Object[] { e });
        }
      }
    }
    try {
      for (Entry<String, UimaTransport> transport : transports.entrySet()) {
        transport.getValue().stopIt();
      }
    } catch (Exception e) {
      if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) {
        UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, getClass().getName(),
                "stop", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
                "UIMAEE_exception__WARNING", new Object[] { e });
      }
    }

    if (cmOutstandingCASes != null) {
      if (!cmOutstandingCASes.isEmpty()) {
        // If there are outstanding CASes, force them to be released
        // If the CM is blocking on getCAS() this will unblock it and
        // enable termination. Otherwise, a hang may occur
        Iterator<String> it = cmOutstandingCASes.keySet().iterator();
        while (it.hasNext()) {
          String casId = it.next();
          try {
            CacheEntry entry = getInProcessCache().getCacheEntryForCAS(casId);
            if (entry != null && entry.getCas() != null) {
              System.out.println("Primitive:" + getComponentName() + " Forcing Release of CAS:"
                      + casId + " in stop()");
              // Force CAS release to unblock CM thread
              entry.getCas().release();
            }
          } catch (Exception e) {
            System.out.println("Controller:" + getComponentName() + " CAS:" + casId
                    + " Not Found In Cache");

          }
        }

      }
      cmOutstandingCASes.clear();
    }
    if (aeList != null) {
      aeList.clear();
      aeList = null;
    }
    System.out.println("Service:" + getComponentName() + " Stopped");
  }

}