/*
 * 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.collection.impl.cpm.engine;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Hashtable;
import java.util.Map;

import org.apache.uima.UIMAFramework;
import org.apache.uima.UimaContextAdmin;
import org.apache.uima.cas.CAS;
import org.apache.uima.cas.impl.CASImpl;
import org.apache.uima.cas_data.CasData;
import org.apache.uima.collection.CasInitializer;
import org.apache.uima.collection.CollectionException;
import org.apache.uima.collection.CollectionReader;
import org.apache.uima.collection.StatusCallbackListener;
import org.apache.uima.collection.base_cpm.BaseCollectionReader;
import org.apache.uima.collection.base_cpm.CasDataCollectionReader;
import org.apache.uima.collection.base_cpm.SkipCasException;
import org.apache.uima.collection.impl.EntityProcessStatusImpl;
import org.apache.uima.collection.impl.cpm.Constants;
import org.apache.uima.collection.impl.cpm.utils.CPMUtils;
import org.apache.uima.collection.impl.cpm.utils.ChunkMetadata;
import org.apache.uima.collection.impl.cpm.vinci.DATACasUtils;
import org.apache.uima.resource.ResourceProcessException;
import org.apache.uima.resource.metadata.Capability;
import org.apache.uima.util.Level;
import org.apache.uima.util.ProcessTrace;
import org.apache.uima.util.Progress;
import org.apache.uima.util.UimaTimer;
import org.apache.uima.util.impl.ProcessTrace_impl;

/**
 * Component responsible for continuously filling a work queue with bundles containing Cas'es. The
 * queue is shared with a Processing Pipeline that consumes bundles of Cas. As soon as the the
 * bundle is removed from the queue, this component fetches data from configured Collection Reader
 * and enques it onto the queue. This component facilitates asynchronous reading and processing of
 * CAS by seperate threads running in the CPE.
 * 
 * When end of processing is reached due to CPM shutdown or max number of entities are processed a
 * special token, called EOFToken is placed onto a queue. It marks end of processing for Processing
 * Units. No more data is expected to be placed on the work queue. The Processing Threads upon
 * seeing the EOFToken are expected to complete processing and do necessary cleanup.
 * 
 * 
 */
public class ArtifactProducer extends Thread {

  /** The thread state. */
  public int threadState = 0;

  /** The cas pool. */
  private CPECasPool casPool;

  /** The work queue. */
  // Queue shared wit ProcessingUnits
  private BoundedWorkQueue workQueue = null;

  /** The collection reader. */
  // private BoundedWorkQueue outputQueue = null;
  private BaseCollectionReader collectionReader = null;

  /** The reader fetch size. */
  // Number of CAS'es for each fetch from the CollectionReader
  private int readerFetchSize = 1;

  /** The cas list. */
  private CAS[] casList;

  /** The entity count. */
  private long entityCount = 0;

  /** The max to process. */
  private long maxToProcess;

  /** The cpm. */
  private CPMEngine cpm = null;

  /** The cpm stat table. */
  private Map cpmStatTable = null;

  /** The last doc id. */
  private String[] lastDocId = { "" };

  /** The total fetch time. */
  private long totalFetchTime = 0;

  /** The timer. */
  private UimaTimer timer = null;

  /** The callback listeners. */
  private ArrayList callbackListeners = null;

  /** The timedout docs. */
  private Hashtable timedoutDocs = new Hashtable();

  /** The is running. */
  private boolean isRunning = false;

  /** The global shared process trace. */
  private ProcessTrace globalSharedProcessTrace = null;

  /**
   * Instantiates and initializes this instance.
   *
   * @param acpm
   *          the acpm
   */
  public ArtifactProducer(CPMEngine acpm) {
    cpm = acpm;
    if (cpm != null) {
      callbackListeners = cpm.getCallbackListeners();
    }
  }

  /**
   * Construct instance of this class with a reference to the cpe engine and a pool of cas'es.
   * 
   * @param acpm
   *          - reference to the cpe
   * @param aPool
   *          - pool of cases
   */
  public ArtifactProducer(CPMEngine acpm, CPECasPool aPool) {
    cpm = acpm;
    casPool = aPool;
    if (cpm != null) {
      callbackListeners = cpm.getCallbackListeners();
    }
  }

  /**
   * Checks if is running.
   *
   * @return true, if is running
   */
  public boolean isRunning() {
    return isRunning;
  }

  /**
   * Plug in Custom Timer to time events.
   *
   * @param aTimer
   *          - custom timer
   */
  public void setUimaTimer(UimaTimer aTimer) {
    timer = aTimer;
  }

  /**
   * Sets the process trace.
   *
   * @param aProcTrace
   *          the new process trace
   */
  public void setProcessTrace(ProcessTrace aProcTrace) {
    globalSharedProcessTrace = aProcTrace;
  }

  /**
   * Returns total time spent when fetching entities from a CollectionReader. This provides a way of
   * gauging throughput of a particular CR.
   * 
   * @return total time spent when fetching entities. -1 when the fetch time is unknown.
   */
  public long getCollectionReaderTotalFetchTime() {
    if (timer != null && totalFetchTime > 0) {
      return totalFetchTime;
    }
    return -1;
  }

  /**
   * Null out fields of this object. Call this only when this object is no longer needed.
   */
  public void cleanup() {
    this.casPool = null;
    this.workQueue = null;
    this.collectionReader = null;
    this.casList = null;
    this.cpm = null;
    if (this.cpmStatTable != null) {
      this.cpmStatTable.clear();
      this.cpmStatTable = null;
    }
    this.lastDocId = null;
  }

  /**
   * Assign total number of entities to process.
   *
   * @param aNumToProcess
   *          - number of entities to read from the Collection Reader
   */
  public void setNumEntitiesToProcess(long aNumToProcess) {
    maxToProcess = aNumToProcess;
  }

  /**
   * Assign CollectionReader to be used for reading.
   *
   * @param aCollectionReader
   *          - collection reader as source of data
   */
  public void setCollectionReader(BaseCollectionReader aCollectionReader) {
    collectionReader = aCollectionReader;
    if (collectionReader.getProcessingResourceMetaData().getConfigurationParameterSettings()
            .getParameterValue("fetchSize") != null) {
      // Determines how many at a time this Collection Reader will return
      // for each fetch
      readerFetchSize = (Integer) collectionReader.getProcessingResourceMetaData()
              .getConfigurationParameterSettings().getParameterValue("fetchSize");
    }
  }

  /**
   * Assigns a queue where the artifacts produced by this component will be deposited.
   *
   * @param aQueue
   *          - queue for the artifacts this class is producing
   */
  public void setWorkQueue(BoundedWorkQueue aQueue) {
    workQueue = aQueue;
  }

  /**
   * Add table that will contain statistics gathered while reading entities from a Collection This
   * table is used for non-uima reports.
   *
   * @param aStatTable
   *          the new CPM stat table
   */
  public void setCPMStatTable(Map aStatTable) {
    cpmStatTable = aStatTable;
  }

  /**
   * Determines if the CPM has processed configured number of entities. Called after each fetch from
   * the Collection Reader.
   * 
   * @return true - all configurted entities processed, false otherwise
   */
  private boolean endOfProcessingReached() {
    // Special case, -1 means all entities in the corpus
    if (maxToProcess == -1) {
      return false;
    } else if (maxToProcess == 0) {
      return true;
    } else {
      // check if exceeded or matched the configured max number of
      // entities
      return (entityCount >= maxToProcess);
    }
  }

  /**
   * Fills the queue up to capacity. This is called before activating ProcessingPipeline as means of
   * optimizing processing. When pipelines start up there are already entities in the work queue to
   * process.
   *
   * @throws Exception
   *           the exception
   */
  public void fillQueue() throws Exception {
    // Create an array holding CAS'es. Configuration of the Reader may
    // include
    // a number of CAS'es to fetch at a time. In this case the array will
    // have
    // size greater than the default (1)
    Object[] casObjectList = new Object[1];
    long capacity = workQueue.getCapacity();
    if (capacity > maxToProcess) {
      capacity = maxToProcess;
      if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) {
        UIMAFramework.getLogger(this.getClass()).logrb(Level.FINEST, this.getClass().getName(),
                "process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE, "UIMA_CPM_reset_queue_size__FINEST",
                new Object[] { Thread.currentThread().getName(), String.valueOf(capacity) });
      }
    }

    try {
      // Fill the work queue with entities from the CollectionReader
      // capacity=number of slots in the work queue
      for (int i = 0; i < capacity; i++) {
        if (collectionReader.hasNext()) {
          // The CollectionReader returns 1 or more entities at a
          // time. The CollectionReader
          // configuration determines how many it will return for each
          // getNext() call. In case
          // WF Large Store, the 'readerFetchSize' parameter is not
          // used, since the store itself
          // determines how many entities to return for each fetch.
          casObjectList = readNext(readerFetchSize);
          if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) {
            UIMAFramework.getLogger(this.getClass()).logrb(Level.FINEST, this.getClass().getName(),
                    "process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE,
                    "UIMA_CPM_enqueue_cas_bundle__FINEST", new Object[] {
                        Thread.currentThread().getName(), String.valueOf(casObjectList.length) });
          }
          // Count number of entities fetched so far
          entityCount += casObjectList.length;

          // append entities to queue
          workQueue.enqueue(casObjectList);
          // If CollectionReader returns bundles instead of individual
          // entity
          // make the loop termines if exceeding total number of
          // entities to
          // process
          if (entityCount > maxToProcess) {
            break;
          }
        }
      }
      if (cpmStatTable != null) {
        Progress[] progress = collectionReader.getProgress();
        if (progress != null) {
          if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) {
            UIMAFramework.getLogger(this.getClass()).logrb(Level.FINEST, this.getClass().getName(),
                    "process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE,
                    "UIMA_CPM_show_cr_progress__FINEST",
                    new Object[] { Thread.currentThread().getName(),
                        String.valueOf(progress[0].getCompleted()) });
          }
        }
        cpmStatTable.put("COLLECTION_READER_PROGRESS", progress);
      }
    } catch (Exception e) {
      if (casObjectList == null) {
        notifyListeners(null, e);
      } else {
        // Release CAS's back to the cas pool.
        for (int i = 0; casObjectList != null && i < casObjectList.length; i++) {
          if (casObjectList[i] != null && casObjectList[i] instanceof CAS) {
            notifyListeners((CAS) casObjectList[i], e);
            casPool.releaseCas(casList[i]);
            casList[i] = null;
            // synchronized (casPool) { // removed - redundant, because done as part of releaseCas
            // casPool.notifyAll();
            // }

          } else {
            notifyListeners(null, e);
          }
        }

      }
      throw e;
    }

  }

  /**
   * Reads next set of entities from the CollectionReader. This method may return more than one Cas
   * at a time.
   *
   * @param fetchSize
   *          the fetch size
   * @return - The Object returned from the method depends on the type of the CollectionReader.
   *         Either CASData[] or CASObject[] initialized with document metadata and content is
   *         returned. If the CollectionReader has no more entities (EOF), null is returned.
   * @throws IOException
   *           - error while reading corpus
   * @throws CollectionException
   *           -
   * @parma fetchSize - number of entities the CollectionReader should return for each fetch. It is
   *        hint as the Collection Reader ultimately decides how many to return.
   */
  private Object[] readNext(int fetchSize) throws IOException, CollectionException {
    ProcessTrace localTrace = new ProcessTrace_impl(cpm.getPerformanceTuningSettings());

    boolean success = false;
    Object[] casObjects = null;
    threadState = 1000; // Entering hasNext()
    // Checks if the CollectionReader has any documents left
    long start = 0;
    if (timer != null) {
      start = timer.getTimeInMillis();
    }
    boolean eventStarted = false;

    // CasObject based CollectionReader does not support returning more than
    // one CAS at a time. So
    // fake support for this by calling its getNext() until the casList is
    // filled to max capacity.
    // The capacity of casList is equal to the CollectionReader fetchSize,
    // defined in CR descriptor.
    if (collectionReader instanceof CollectionReader) {
      casList = new CAS[fetchSize];
      for (int i = 0; i < fetchSize; i++) {

        if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) {
          UIMAFramework.getLogger(this.getClass()).logrb(Level.FINEST, this.getClass().getName(),
                  "process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE, "UIMA_CPM_cr_fetch_new_cas__FINEST",
                  new Object[] { Thread.currentThread().getName() });
        }

        threadState = 1001; // Waiting for CAS
        // Get the cas from the pool.
        while (cpm.isRunning() && (casList[i] = casPool.getCas(0)) == null)
          ; // intentionally empty while loop

        if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) {
          UIMAFramework.getLogger(this.getClass()).logrb(Level.FINEST, this.getClass().getName(),
                  "process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE,
                  "UIMA_CPM_cr_check_cas_for_null__FINEST", new Object[] {
                      Thread.currentThread().getName(), String.valueOf((casList[i] == null)) });
        }
        if (cpm.isRunning() == false) {
          // CPM is in shutdown stage. No need to enqueue additional
          // documents/CAS'es. Just release
          // those that have been aquired so far back to the pool and
          // return null, indicating
          // end of processing.
          if (timer != null) {
            totalFetchTime += (timer.getTimeInMillis() - start);
          }
          for (int listCounter = 0; casList != null && casList[i] != null
                  && listCounter < casList.length; listCounter++) {
            casPool.releaseCas(casList[listCounter]);
            // synchronized (casPool) { // redundant - releaseCas call does this
            // casPool.notifyAll();
            // }
          }
          if (cpmStatTable != null) {
            Progress[] progress = collectionReader.getProgress();
            cpmStatTable.put("COLLECTION_READER_PROGRESS", progress);
            cpmStatTable.put("COLLECTION_READER_TIME", totalFetchTime);
          }
          if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) {
            UIMAFramework.getLogger(this.getClass()).logrb(Level.FINEST, this.getClass().getName(),
                    "process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE,
                    "UIMA_CPM_in_shutdown_state__FINEST",
                    new Object[] { Thread.currentThread().getName() });
          }
          return null;
        }
        if (casList[i] == null) {
          return null;
        }
        if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) {
          UIMAFramework.getLogger(this.getClass()).logrb(Level.FINEST, this.getClass().getName(),
                  "process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE, "UIMA_CPM_got_new_cas__FINEST",
                  new Object[] { Thread.currentThread().getName() });
        }
        if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) {
          UIMAFramework.getLogger(this.getClass()).logrb(Level.FINEST, this.getClass().getName(),
                  "process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE, "UIMA_CPM_call_cas_reset__FINEST",
                  new Object[] { Thread.currentThread().getName() });
        }
        casList[i].reset();

        // If Collection Reader and CAS Initilaizer do not declare any
        // output SofAs, must be passed the default view (meaning whatever's
        // mapped to _InitialView) for backward compatiblity
        Capability[] capabilities;
        CasInitializer casIni = ((CollectionReader) collectionReader).getCasInitializer();
        if (casIni != null)
          capabilities = casIni.getProcessingResourceMetaData().getCapabilities();
        else
          capabilities = ((CollectionReader) collectionReader).getProcessingResourceMetaData()
                  .getCapabilities();

        boolean sofaUnaware = true;
        for (int j = 0; j < capabilities.length; j++) {
          if (capabilities[j].getOutputSofas().length > 0) {
            sofaUnaware = false;
            break;
          }
        }

        threadState = 1003; // Entering

        // set the current component info of the CAS, so that it knows
        // the sofa
        // mappings for the component that's about to process it
        UimaContextAdmin context = ((CollectionReader) collectionReader).getUimaContextAdmin();
        casList[i].setCurrentComponentInfo(context.getComponentInfo());
        try {
          if (sofaUnaware) {
            // sofa-unaware CR, give it whatever is mapped to the
            // initial view (creating that view first if it's not the default)
            String absSofaName = context.getComponentInfo().mapToSofaID(CAS.NAME_DEFAULT_SOFA);
            if (!CAS.NAME_DEFAULT_SOFA.equals(absSofaName)) {
              casList[i].createView(CAS.NAME_DEFAULT_SOFA);
            }
            CAS view = casList[i].getView(CAS.NAME_DEFAULT_SOFA);

            if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) {
              UIMAFramework.getLogger(this.getClass()).logrb(Level.FINEST,
                      this.getClass().getName(), "process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE,
                      "UIMA_CPM_call_cr_next__FINEST",
                      new Object[] { Thread.currentThread().getName(), "CAS" });
            }
            localTrace.startEvent(collectionReader.getProcessingResourceMetaData().getName(),
                    "Process", "");
            eventStarted = true;
            ((CollectionReader) collectionReader).getNext(view);
            localTrace.endEvent(collectionReader.getProcessingResourceMetaData().getName(),
                    "Process", "success");

            if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) {
              UIMAFramework.getLogger(this.getClass()).logrb(Level.FINEST,
                      this.getClass().getName(), "process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE,
                      "UIMA_CPM_call_cr_next_finished__FINEST",
                      new Object[] { Thread.currentThread().getName(), "CAS" });
            }
          } else
          // sofa-aware CR, give it the base CAS
          {
            CAS baseCas = ((CASImpl) casList[i]).getBaseCAS();
            if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) {
              UIMAFramework.getLogger(this.getClass()).logrb(Level.FINEST,
                      this.getClass().getName(), "process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE,
                      "UIMA_CPM_call_cr_next__FINEST",
                      new Object[] { Thread.currentThread().getName(), "CAS" });

            }
            localTrace.startEvent(collectionReader.getProcessingResourceMetaData().getName(),
                    "Process", "");
            eventStarted = true;
            ((CollectionReader) collectionReader).getNext(baseCas);
            localTrace.endEvent(collectionReader.getProcessingResourceMetaData().getName(),
                    "Process", "success");

            if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) {
              UIMAFramework.getLogger(this.getClass()).logrb(Level.FINEST,
                      this.getClass().getName(), "process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE,
                      "UIMA_CPM_call_cr_next_finished__FINEST",
                      new Object[] { Thread.currentThread().getName(), "CAS" });
            }
          }
          success = true;
        } finally {
          // be sure to unset the component info in the CAS, since the
          // CAS is no longer
          // being processed by the CollectionReader
          casList[i].setCurrentComponentInfo(null);
          if (eventStarted) // use this to make sure we dont end event that has not been explicitely
          // started
          {
            if (!success) {
              localTrace.endEvent(collectionReader.getProcessingResourceMetaData().getName(),
                      "Process", "failure");

            }
            synchronized (globalSharedProcessTrace) {
              globalSharedProcessTrace.aggregate(localTrace);
            }

          }

        }
      }
      casObjects = casList;
      if (casObjects != null && casObjects.length > 0) {
        try {
          if (((CASImpl) casList[0]).isBackwardCompatibleCas()) {
            CAS view = casList[0].getView(CAS.NAME_DEFAULT_SOFA);
            lastDocId[0] = ConsumerCasUtils.getStringFeatValue(view, Constants.METADATA_KEY,
                    Constants.DOC_ID);
          } else {
            lastDocId[0] = "";
          }
        } catch (Exception e) {
          lastDocId[0] = "";
        }
      }
    } else {
      // Retrieve next set of CAS'es. fetchSize is hint to the
      // CollectionReader how many to return
      // Some CollectionReaders return a batch with size different than
      // recommended in
      // fetchSize. Most notably, Large WF Store decides itself how many
      // entities to return for
      // each fetch.

      if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) {
        UIMAFramework.getLogger(this.getClass()).logrb(Level.FINEST, this.getClass().getName(),
                "process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE, "UIMA_CPM_call_cr_next__FINEST",
                new Object[] { Thread.currentThread().getName(), "CasData" });
      }
      localTrace.startEvent(collectionReader.getProcessingResourceMetaData().getName(), "Process",
              "");
      try {
        casObjects = ((CasDataCollectionReader) collectionReader).getNext(fetchSize);
        success = true;
      } finally {
        if (!success) {
          localTrace.endEvent(collectionReader.getProcessingResourceMetaData().getName(), "Process",
                  "failure");

        } else {
          localTrace.endEvent(collectionReader.getProcessingResourceMetaData().getName(), "Process",
                  "success");

        }
        synchronized (globalSharedProcessTrace) {
          globalSharedProcessTrace.aggregate(localTrace);
        }

      }
      if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) {
        UIMAFramework.getLogger(this.getClass()).logrb(Level.FINEST, this.getClass().getName(),
                "process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE,
                "UIMA_CPM_call_cr_next_finished__FINEST",
                new Object[] { Thread.currentThread().getName(), "CasData" });
      }
      if (casObjects != null && casObjects.length > 0) {
        lastDocId = DATACasUtils.getFeatureStructureValues((CasData) casObjects[0],
                Constants.METADATA_KEY, Constants.DOC_ID);
      }
    }
    if (timer != null) {
      totalFetchTime += (timer.getTimeInMillis() - start);
    }

    if (cpmStatTable != null) {
      Progress[] progress = collectionReader.getProgress();
      cpmStatTable.put("COLLECTION_READER_PROGRESS", progress);
      cpmStatTable.put("COLLECTION_READER_TIME", totalFetchTime);
    }
    if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) {
      UIMAFramework.getLogger(this.getClass()).logrb(Level.FINEST, this.getClass().getName(),
              "process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE, "UIMA_CPM_return_cases_from_cr__FINEST",
              new Object[] { Thread.currentThread().getName(), "CAS" });
    }
    return casObjects;
  }

  /**
   * Runs this thread until the CPM halts or the CollectionReader has no more entities. It
   * continuously fills the work queue with entities returned by the CollectionReader.
   */
  @Override
  public void run() {
    boolean crEventCompleted = false; // this flag is used to mark the
    // ProcessTrace event
    if (!cpm.isRunning()) {
      UIMAFramework.getLogger(this.getClass()).logrb(Level.WARNING, this.getClass().getName(),
              "process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE, "UIMA_CPM_cpm_not_running__WARNING",
              new Object[] { Thread.currentThread().getName() });
      return;
    }

    Object[] casObjectList = null;
    // Check if
    if (endOfProcessingReached()) {
      if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) {
        UIMAFramework.getLogger(this.getClass()).logrb(Level.FINEST, this.getClass().getName(),
                "process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE, "UIMA_CPM_cr_done_producing__FINEST",
                new Object[] { Thread.currentThread().getName() });
      }
      placeEOFToken();
      if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) {
        UIMAFramework.getLogger(this.getClass()).logrb(Level.FINEST, this.getClass().getName(),
                "process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE, "UIMA_CPM_eof_marker_enqueued__FINEST",
                new Object[] { Thread.currentThread().getName() });
      }
      return;
    }
    isRunning = true;
    ProcessTrace localTrace = new ProcessTrace_impl(cpm.getPerformanceTuningSettings());
    while (cpm.isRunning()) {

      casList = null;
      casObjectList = null;
      synchronized (cpm.lockForPause) {
        if (cpm.isPaused()) {
          try {
            if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) {
              UIMAFramework.getLogger(this.getClass()).logrb(Level.FINEST,
                      this.getClass().getName(), "process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE,
                      "UIMA_CPM_pausing_cr__FINEST",
                      new Object[] { Thread.currentThread().getName() });
            }
            // Wait until resumed
            cpm.lockForPause.wait();
          } catch (Exception e) {
          }
          if (!cpm.isRunning()) {
            break;
          }
          if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) {
            UIMAFramework.getLogger(this.getClass()).logrb(Level.FINEST, this.getClass().getName(),
                    "process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE, "UIMA_CPM_resume_cr__FINEST",
                    new Object[] { Thread.currentThread().getName() });
          }
        }
      }

      try {
        if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) {
          UIMAFramework.getLogger(this.getClass()).logrb(Level.FINEST, this.getClass().getName(),
                  "process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE, "UIMA_CPM_call_hasnext__FINEST",
                  new Object[] { Thread.currentThread().getName() });
        }
        threadState = 1004; // Entering hasNext()

        // start the CR event
        localTrace.startEvent(collectionReader.getProcessingResourceMetaData().getName(), "Process",
                "");
        crEventCompleted = false;
        if (collectionReader.hasNext()) {
          localTrace.endEvent(collectionReader.getProcessingResourceMetaData().getName(), "Process",
                  "success");
          crEventCompleted = true;

          if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) {
            UIMAFramework.getLogger(this.getClass()).logrb(Level.FINEST, this.getClass().getName(),
                    "process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE, "UIMA_CPM_get_cas_from_cr__FINEST",
                    new Object[] { Thread.currentThread().getName() });
          }
          casObjectList = readNext(readerFetchSize);
          if (casObjectList != null) {
            if (casObjectList instanceof CAS[]) {
              boolean releasedCas = false;
              for (int i = 0; i < casObjectList.length && casObjectList[i] != null; i++) {
                ChunkMetadata meta = CPMUtils.getChunkMetadata((CAS) casObjectList[i]);
                if (meta != null) {
                  if (timedoutDocs.containsKey(meta.getDocId())) {
                    notifyListeners(casList[i],
                            new ResourceProcessException(new SkipCasException(
                                    "Dropping CAS due chunk Timeout. Doc Id::" + meta.getDocId()
                                            + " Sequence:" + meta.getSequence())));

                    casPool.releaseCas((CAS) casObjectList[i]);
                    // synchronized (casPool) { // redundant, releaseCas call does this
                    // casPool.notifyAll();
                    // }
                    releasedCas = true;
                  }
                }
              }
              if (releasedCas) {
                continue;
              }
            }
            if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) {
              UIMAFramework.getLogger(this.getClass()).logrb(Level.FINEST,
                      this.getClass().getName(), "process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE,
                      "UIMA_CPM_place_cas_in_queue__FINEST", new Object[] {
                          Thread.currentThread().getName(), String.valueOf(casObjectList.length) });
            }
            // Prevent processing of new CASes if the CPM has been
            // killed hard. Allow processing of CASes
            // while the CPM is in normal shutdown state.
            // (Moved this code inside if (casObjectList != null)
            // block to avoid NullPointerException. -Adam
            if (cpm.isRunning() == true
                    || (cpm.isRunning() == false && cpm.isHardKilled() == false)) {
              threadState = 1005; // Entering enqueue
              workQueue.enqueue(casObjectList);
              // synchronized (workQueue) { // redundant, enqueue does this
              // workQueue.notifyAll();
              // }
              threadState = 1006; // Done Entering enqueue
              entityCount += casObjectList.length;
              if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) {
                UIMAFramework.getLogger(this.getClass()).logrb(Level.FINEST,
                        this.getClass().getName(), "process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE,
                        "UIMA_CPM_placed_cas_in_queue__FINEST",
                        new Object[] { Thread.currentThread().getName(),
                            String.valueOf(casObjectList.length) });
              }
            } else {
              break; // CPM has been killed
            }
          } else {
            if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) {
              UIMAFramework.getLogger(this.getClass()).logrb(Level.FINEST,
                      this.getClass().getName(), "process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE,
                      "UIMA_CPM_terminate_cr_thread__FINEST",
                      new Object[] { Thread.currentThread().getName() });
            }
            break; // Null should not be returned from getNext
            // unless the CPM is in shutdown mode
          }
        } else {
          if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) {
            UIMAFramework.getLogger(this.getClass()).logrb(Level.FINEST, this.getClass().getName(),
                    "process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE, "UIMA_CPM_processed_all__FINEST",
                    new Object[] { Thread.currentThread().getName() });
          }
          // Stops the CPM and all of the running threads.
          // cpm.stopIt(); APL - don't stop, just terminate this
          // thread, which CPMEngine has joined on
          break;

        }
        // Check if the CollectionReader retrieved expected number of
        // entities
        if (endOfProcessingReached()) {
          threadState = 1010; // End of processing

          if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) {
            UIMAFramework.getLogger(this.getClass()).logrb(Level.FINEST, this.getClass().getName(),
                    "process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE,
                    "UIMA_CPM_end_of_processing__FINEST",
                    new Object[] { Thread.currentThread().getName() });
          }
          break;
        }
      } catch (Exception e) {
        // The following conditional is true if hasNext() has failed
        if (!crEventCompleted) {
          localTrace.endEvent(collectionReader.getProcessingResourceMetaData().getName(), "Process",
                  "failure");
        }
        // e.printStackTrace();
        // changed from FINER to WARNING: https://issues.apache.org/jira/browse/UIMA-2440
        if (UIMAFramework.getLogger().isLoggable(Level.WARNING)) {
          UIMAFramework.getLogger(this.getClass()).logrb(Level.WARNING, this.getClass().getName(),
                  "process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE, "UIMA_CPM_exception__WARNING",
                  new Object[] { Thread.currentThread().getName(), e.getMessage() });

          UIMAFramework.getLogger(this.getClass()).log(Level.WARNING, e.getMessage(), e);
        }
        if (casList == null) {
          notifyListeners(null, e);
        } else {
          // Notify Listeners and release CAS's back to the cas pool.
          for (int i = 0; casList != null && i < casList.length; i++) {
            if (casList[i] != null) {
              notifyListeners(casList[i], e);
              casPool.releaseCas(casList[i]);
              casList[i] = null;
              // synchronized (casPool) { // redundant, releaseCas does this
              // casPool.notifyAll();
              // }

            } else {
              notifyListeners(null, e);
            }
            casList = null;
          }

        }
      } finally {
        // Clear all events
        synchronized (globalSharedProcessTrace) {
          globalSharedProcessTrace.aggregate(localTrace);
        }
        localTrace.clear();
      }
      if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) {
        UIMAFramework.getLogger(this.getClass()).logrb(Level.FINEST, this.getClass().getName(),
                "process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE,
                "UIMA_CPM_show_cpm_running_status__FINEST",
                new Object[] { Thread.currentThread().getName(), String.valueOf(cpm.isRunning()) });
      }
    }

    if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) {
      UIMAFramework.getLogger(this.getClass()).logrb(Level.FINEST, this.getClass().getName(),
              "process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE,
              "UIMA_CPM_show_cpm_running_status__FINEST",
              new Object[] { Thread.currentThread().getName(), String.valueOf(cpm.isRunning()) });
    }
    // Done with processing. Create a "special" EOF token and place it in
    // the queue.
    // Consumers of the queue must interpret this token as End Of File
    // event, meaning
    // end of processing. Such components must do appropriate cleanup and
    // terminate.
    placeEOFToken();
    isRunning = false;
    // Interrupt any waiting threads
    Thread.currentThread().interrupt();
  }

  /**
   * Notify registered callback listeners of a given exception.
   *
   * @param aCas
   *          the a cas
   * @param anException
   *          - exception to propagate to callback listeners
   */
  private void notifyListeners(CAS aCas, Exception anException) {
    for (int i = 0; callbackListeners != null && i < callbackListeners.size(); i++) {
      StatusCallbackListener statCL = (StatusCallbackListener) callbackListeners.get(i);
      if (statCL != null) {
        ProcessTrace prTrace = new ProcessTrace_impl(cpm.getPerformanceTuningSettings());
        EntityProcessStatusImpl aEntityProcStatus = new EntityProcessStatusImpl(prTrace);
        aEntityProcStatus.addEventStatus("Collection Reader Failure", "failed", anException);
        // Notify the listener that the Cas has been processed
        CPMEngine.callEntityProcessCompleteWithCAS(statCL, aCas, aEntityProcStatus);
        // statCL.entityProcessComplete(aCas, aEntityProcStatus);
      }
    }
  }

  /**
   * Place terminating EOFToken into a Work Queue. Any thread reading this token from the queue is
   * responsible for terminating itself.
   * 
   */
  private void placeEOFToken() {
    // Done with processing. Create a "special" EOF token and place it in
    // the queue.
    // Consumers of the queue must interpret this token as End Of File
    // event, meaning
    // end of processing. Such components must do appropriate cleanup and
    // terminate.
    try {
      Object[] eofToken = new Object[1];
      // only need one member in the array
      eofToken[0] = new EOFToken();
      if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) {
        UIMAFramework.getLogger(this.getClass()).logrb(Level.FINEST, this.getClass().getName(),
                "process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE, "UIMA_CPM_enqueue_eof_token__FINEST",
                new Object[] { Thread.currentThread().getName(), String.valueOf(cpm.isRunning()) });
      }
      workQueue.enqueue(eofToken);
      if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) {
        UIMAFramework.getLogger(this.getClass()).logrb(Level.FINEST, this.getClass().getName(),
                "process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE,
                "UIMA_CPM_done_enqueue_eof_token__FINEST",
                new Object[] { Thread.currentThread().getName(), String.valueOf(cpm.isRunning()) });

      }
      // synchronized (workQueue) { // redundant, the enqueue call above does this
      // workQueue.notifyAll();
      // }
    } catch (Exception e) {
      e.printStackTrace();
      if (UIMAFramework.getLogger().isLoggable(Level.SEVERE)) {
        UIMAFramework.getLogger(this.getClass()).logrb(Level.SEVERE, this.getClass().getName(),
                "process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE, "UIMA_CPM_exception__FINER",
                new Object[] { Thread.currentThread().getName(), e.getMessage() });
        UIMAFramework.getLogger(this.getClass()).log(Level.SEVERE, "", e);
      }
    }
  }

  /**
   * Gets the last doc id.
   *
   * @return the last doc id
   */
  public String getLastDocId() {
    if (lastDocId != null && lastDocId.length > 0) {
      return lastDocId[0];
    } else {
      return "N/A";
    }
  }

  /**
   * Invalidate.
   *
   * @param aCasList
   *          the a cas list
   */
  public void invalidate(CAS[] aCasList) {
    for (int i = 0; aCasList != null && i < aCasList.length && aCasList[i] != null; i++) {
      ChunkMetadata meta = CPMUtils.getChunkMetadata(aCasList[i]);
      // Add the docId into a cache of documents that have been dropped
      // due to exception
      // during processing. This is only done for chunked documents
      // (sequence > 0)
      if (meta != null && meta.getSequence() > 0 && !timedoutDocs.containsKey(meta.getDocId())) {
        timedoutDocs.put(meta.getDocId(), meta.getDocId());
      }
    }
  }
}