org.apache.uima.adapter.jms.activemq

## Class ConcurrentMessageListener

- java.lang.Object
- - org.apache.uima.adapter.jms.activemq.ConcurrentMessageListener

- All Implemented Interfaces:
    org.springframework.jms.listener.SessionAwareMessageListener

      public class ConcurrentMessageListener
      extends Object
      implements org.springframework.jms.listener.SessionAwareMessageListener

  Message listener injected at runtime into Aggregate to handle a race condition when multiple threads simultaneously process messages from a Cas Multiplier. It is only used to process messages from a Cas Multiplier and only if the reply queue has more than one consumer thread configured in a deployment descriptor. The listener creates a pool of threads equal to the number of concurrent consumers defined in the DD for the listener on the reply queue. Once the message is handled in onMessage(), it is than delegated for processing to one of the available threads from the pool. This listener guarantees processing order. It receives messages from Spring in a single thread and if it finds a child CAS in the message, it increments the parent (input) CAS child count and delegates processing to the InputChannel instance. The race condition: The Cas Multiplier sends the last child and the parent almost at the same time. Both are received by the aggregate and are processed in different threads, if a scaleout is used on the reply queue. One thread may start processing the input CAS while the other thread (with the last child) is not yet allowed to run. The first thread takes the input CAS all the way to the final step and since at this time, the input CAS has no children ( the thread processing the last child has not updated the child count yet), it will be prematurely released. When the thread with the last child is allowed to run, it finds that the parent no longer exists in the cache.

- - ### Constructor Summary

    Constructors
    Constructor and Description
    `ConcurrentMessageListener(int concurrentThreads, Object delegateListener, String destination, ThreadGroup threadGroup, String threadPrefix)` Creates a listener with a given number of process threads.

  - ### Method Summary

    All Methods Instance Methods Concrete Methods
    Modifier and Type | Method and Description
    --- | ---
    `ThreadPoolExecutor` | `getTaskExecutor()`
    `void` | `onMessage(javax.jms.Message message, javax.jms.Session session)` Intercept a message to increment a child count of the input CAS.
    `void` | `setAnalysisEngineController(org.apache.uima.aae.controller.AnalysisEngineController controller)`
    `void` | `stop()`

    - ### Methods inherited from class java.lang.Object

      `clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait`

- - ### Constructor Detail

    - #### ConcurrentMessageListener

          public ConcurrentMessageListener(int concurrentThreads,
                                           Object delegateListener,
                                           String destination,
                                           ThreadGroup threadGroup,
                                           String threadPrefix)
                                    throws InvalidClassException

      Creates a listener with a given number of process threads. This listener is injected between Spring and JmsInputChannel to enable orderly processing of CASes. This listener is only used on reply queues that have scale out attribute in DD greater than 1. Its main job is to increment number of child CASes for a given input CAS. It does so in a single thread, and once it completes the update this listener submits the CAS for further processing up to the JmsInputChannel. The CAS is submitted to a queue where the executor assigns a free thread to process the CAS.
      Parameters:
        `concurrentThreads` - - number of threads to use to process CASes
        `delegateListener` - - JmsInputChannel instance to delegate CAS to
      Throws:
        `InvalidClassException`

  - ### Method Detail

    - #### getTaskExecutor

          public ThreadPoolExecutor getTaskExecutor()

    - #### stop

          public void stop()

    - #### setAnalysisEngineController

          public void setAnalysisEngineController(org.apache.uima.aae.controller.AnalysisEngineController controller)

    - #### onMessage

          public void onMessage(javax.jms.Message message,
                                javax.jms.Session session)
                         throws javax.jms.JMSException

      Intercept a message to increment a child count of the input CAS. This method is always called in a single thread, guaranteeing order of processing. The child CAS will always come here first. Once the count is updated, or this CAS is not an child, the message will be delegated to one of the threads in the pool that will eventually call InputChannel object where the actual processing of the message begins.
      Specified by:
        `onMessage` in interface `org.springframework.jms.listener.SessionAwareMessageListener`
      Throws:
        `javax.jms.JMSException`

Copyright © 2006–2018 The Apache Software Foundation. All rights reserved.