| /* |
| * 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.adapter.jms.client; |
| |
| import java.util.List; |
| import java.util.concurrent.BlockingQueue; |
| import java.util.concurrent.LinkedBlockingQueue; |
| import java.util.concurrent.Semaphore; |
| |
| import javax.jms.Connection; |
| import javax.jms.DeliveryMode; |
| import javax.jms.Destination; |
| import javax.jms.InvalidDestinationException; |
| import javax.jms.Message; |
| import javax.jms.MessageProducer; |
| import javax.jms.TextMessage; |
| |
| import org.apache.uima.UIMAFramework; |
| import org.apache.uima.aae.UIMAEE_Constants; |
| import org.apache.uima.aae.client.UimaASProcessStatus; |
| import org.apache.uima.aae.client.UimaASProcessStatusImpl; |
| import org.apache.uima.aae.client.UimaAsynchronousEngine; |
| import org.apache.uima.aae.delegate.Delegate; |
| import org.apache.uima.aae.error.ServiceShutdownException; |
| import org.apache.uima.aae.message.AsynchAEMessage; |
| import org.apache.uima.aae.message.UimaMessageValidator; |
| import org.apache.uima.adapter.jms.JmsConstants; |
| import org.apache.uima.adapter.jms.client.BaseUIMAAsynchronousEngineCommon_impl.ClientRequest; |
| import org.apache.uima.adapter.jms.client.BaseUIMAAsynchronousEngineCommon_impl.SharedConnection; |
| import org.apache.uima.adapter.jms.message.PendingMessage; |
| import org.apache.uima.analysis_engine.AnalysisEngineProcessException; |
| import org.apache.uima.cas.CAS; |
| import org.apache.uima.jms.error.handler.BrokerConnectionException; |
| import org.apache.uima.util.Level; |
| import org.apache.uima.util.ProcessTrace; |
| import org.apache.uima.util.impl.ProcessTrace_impl; |
| |
| /** |
| * Creates a worker thread for sending messages. This is an abstract implementation that provides a |
| * thread with run logic. The concrete implementation of the Worker Thread must extend this class. |
| * The application threads share a special in-memory queue with this worker thread. The application |
| * threads add jms messages to the pendingMessageList queue and the worker thread consumes them. The |
| * worker thread terminates when the uima ee client calls doStop() method. |
| */ |
| public abstract class BaseMessageSender implements Runnable, MessageSender { |
| private static final Class CLASS_NAME = BaseMessageSender.class; |
| |
| // A reference to a shared queue where application threads enqueue messages |
| // to be sent |
| protected BlockingQueue<PendingMessage> messageQueue = new LinkedBlockingQueue<PendingMessage>(); |
| |
| // Global flag controlling lifecycle of this thread. It will be set to true |
| // when the |
| // uima ee engine calls doStop() |
| protected volatile boolean done; |
| |
| // A reference to the uima ee client engine |
| protected BaseUIMAAsynchronousEngineCommon_impl engine; |
| |
| // Global flag to indicate failure of the worker thread |
| protected volatile boolean workerThreadFailed; |
| |
| // If the worker thread fails, store the reason for the failure |
| protected Exception exception; |
| |
| // These are required methods to be implemented by a concrete implementation |
| |
| // Returns instance of JMS MessageProducer |
| public abstract MessageProducer getMessageProducer(); |
| |
| // Provides implementation-specific implementation logic. It is expected |
| // that this |
| // message creates an instance of JMS MessageProducer |
| protected abstract void initializeProducer() throws Exception; |
| |
| // Releases resources |
| protected abstract void cleanup() throws Exception; |
| |
| // Returns the name of the destination |
| protected abstract String getDestinationEndpoint() throws Exception; |
| |
| public abstract void setConnection(Connection connection); |
| |
| private MessageProducer producer = null; |
| |
| |
| public BaseMessageSender(BaseUIMAAsynchronousEngineCommon_impl anEngine) { |
| messageQueue = anEngine.pendingMessageQueue; |
| engine = anEngine; |
| try { |
| // Acquire a shared lock. Release it in the run() method once we initialize |
| // the producer. |
| engine.producerSemaphore.acquire(); |
| } catch (InterruptedException e) { |
| } |
| } |
| |
| /** |
| * Stops the worker thread |
| */ |
| public void doStop() { |
| done = true; |
| // Create an empty message to deliver to the queue that is blocking |
| PendingMessage emptyMessage = new PendingMessage(0); |
| messageQueue.add(emptyMessage); |
| } |
| |
| /** |
| * Return the Exception that caused the failure in this worker thread |
| * |
| * @return - Exception |
| */ |
| public Exception getReasonForFailure() { |
| return exception; |
| } |
| |
| /** |
| * The uima ee client should call this method to check if there was a failure. The method returns |
| * true if there was a failure or false otherwise. If true, the uima ee client can call |
| * getReasonForFailure() to get the reason for failure |
| */ |
| public boolean failed() { |
| return workerThreadFailed; |
| } |
| /** |
| * This method determines if a given request message should be rejected or not. Rejected in a sense |
| * that it will not be sent to the destination due to the fact the broker connection is down. |
| */ |
| private boolean reject(PendingMessage pm ) { |
| return reject(pm, new BrokerConnectionException("Unable To Deliver Message To Destination. Connection To Broker "+engine.sharedConnection.getBroker()+" Has Been Lost")); |
| } |
| |
| /** |
| * This method determines if a given request message should be rejected or not. Rejected in a sense |
| * that it will not be sent to the destination due to the fact the broker connection is down. |
| */ |
| private boolean reject( PendingMessage pm, Exception e ) { |
| boolean rejectRequest = false; |
| // If the connection to a broker was lost, notify the client |
| // and reject the request unless this is getMeta Ping request. |
| if ( !engine.sharedConnection.isConnectionValid() ) { |
| String messageKind = ""; |
| if (pm.getMessageType() == AsynchAEMessage.GetMeta ) { |
| messageKind = "GetMeta"; |
| } else if (pm.getMessageType() == AsynchAEMessage.Process ) { |
| messageKind = "Process"; |
| } else if (pm.getMessageType() == AsynchAEMessage.CollectionProcessComplete ) { |
| messageKind = "CollectionProcessComplete"; |
| } |
| rejectRequest = true; |
| try { |
| // Is this is a Process request |
| if (pm.getMessageType() == AsynchAEMessage.Process) { |
| // fetch the cache entry for this CAS |
| ClientRequest cacheEntry = (ClientRequest) engine.getCache().get( |
| pm.get(AsynchAEMessage.CasReference)); |
| if ( cacheEntry != null ) { |
| // We are rejecting any Process requests until connection to broker |
| // is recovered |
| cacheEntry.setProcessException(); |
| // if the request was via synchronous API dont notify listeners |
| // instead the code will throw exception to the client |
| boolean notifyListener = (cacheEntry.isSynchronousInvocation() == false); |
| // handle rejected request |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, getClass().getName(), |
| "reject", JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_client_rejected_process_request_broker_down__INFO", new Object[] { messageKind }); |
| } |
| // Sets the state of a remote delegate to handle subsequent requests |
| engine.serviceDelegate.setState(Delegate.TIMEOUT_STATE); |
| // handle exception but dont rethrow the exception |
| engine.handleException(e, cacheEntry.getCasReferenceId(), null, cacheEntry, notifyListener, false); |
| } |
| } else { |
| // Dont handle GetMeta Ping. Let it flow through. The Ping will be done once the connection is recovered |
| if ( !engine.serviceDelegate.isAwaitingPingReply()) { |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, getClass().getName(), |
| "reject", JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_client_rejected_nonprocess_request_broker_down__INFO", new Object[] { messageKind }); |
| } |
| engine.handleNonProcessException(e); |
| } else if ( pm.getMessageType() == AsynchAEMessage.GetMeta ){ |
| rejectRequest = false; // Don't reject GetMeta Ping |
| } |
| } |
| } catch( Exception ex ) { |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, getClass().getName(), |
| "reject", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAEE_exception__WARNING", new Object[] { ex }); |
| } |
| } |
| } |
| return rejectRequest; |
| } |
| /** |
| * Initializes jms message producer and starts the main thread. This thread waits for messages |
| * enqueued by application threads. The application thread adds a jms message to the |
| * pendingMessageList 'queue' and signals this worker that there is a new message. The worker |
| * thread removes the message and sends it out to a given destination. All messages should be |
| * fully initialized The worker thread does check the message nor adds anything new to the |
| * message. It just sends it out. |
| */ |
| public void run() { |
| String destination = null; |
| // by default, add time to live to each message |
| boolean addTimeToLive = true; |
| // Check the environment for existence of NoTTL tag. If present, |
| // the deployer of the service wants to disable message expiration. |
| if (System.getProperty("NoTTL") != null) { |
| addTimeToLive = false; |
| } |
| |
| // Create and initialize the producer. |
| try { |
| initializeProducer(); |
| destination = getDestinationEndpoint(); |
| if (destination == null) { |
| throw new InvalidDestinationException("Unable to determine the destination"); |
| } |
| } catch (Exception e) { |
| workerThreadFailed = true; |
| exception = e; |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, getClass().getName(), |
| "run", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAEE_exception__WARNING", new Object[] { e }); |
| } |
| return; |
| |
| } finally { |
| engine.producerSemaphore.release(); |
| } |
| |
| engine.onProducerInitialized(); |
| |
| |
| // Wait for messages from application threads. The uima ee client engine |
| // will call doStop() which sets the global flag 'done' to true. |
| PendingMessage pm = null; |
| ClientRequest cacheEntry = null; |
| while (!done) { |
| // Remove the oldest message from the shared 'queue' |
| // // Wait for a new message |
| try { |
| pm = messageQueue.take(); |
| } catch (InterruptedException e) { |
| } |
| if (done) { |
| break; // done in this loop |
| } |
| // Check if the request should be rejected. If the connection to the broker is invalid and the request |
| // is not GetMeta Ping, reject the request after the connection is made. The reject() method created |
| // an exception and notified client of the fact that the request could not continue. The ping is a |
| // special case that we dont reject even though the broker connection has been lost. It is allow to |
| // fall through and will be sent as soon as the connection is recovered. |
| boolean rejectRequest = reject(pm); |
| // blocks until the connection is re-established with a broker |
| engine.recoverSharedConnectionIfClosed(); |
| // get the producer initialized from a valid connection |
| producer = getMessageProducer(); |
| // Check if the request should be rejected. It would be the case if the connection was invalid and |
| // subsequently recovered. If it was invalid, we went through error handling and the request is stale. |
| if ( !rejectRequest && engine.running) { |
| if ( engine.serviceDelegate.isAwaitingPingReply() && |
| pm.getMessageType() == AsynchAEMessage.GetMeta ){ |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, getClass().getName(), |
| "run", JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_client_dispatching_getmeta_ping__INFO", new Object[] { }); |
| } |
| } |
| try { |
| // Request JMS Message from the concrete implementation |
| Message message = null; |
| // Determine if this a CAS Process Request |
| boolean casProcessRequest = isProcessRequest(pm); |
| // Only Process request can be serialized as binary |
| if (casProcessRequest && engine.getSerializationStrategy().equals("binary")) { |
| message = createBytesMessage(); |
| } else { |
| message = createTextMessage(); |
| } |
| |
| initializeMessage(pm, message); |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINE)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb( |
| Level.FINE, |
| CLASS_NAME.getName(), |
| "run", |
| JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_sending_msg_to_endpoint__FINE", |
| new Object[] { |
| UimaMessageValidator.decodeIntToString(AsynchAEMessage.Command, message |
| .getIntProperty(AsynchAEMessage.Command)), |
| UimaMessageValidator.decodeIntToString(AsynchAEMessage.MessageType, message |
| .getIntProperty(AsynchAEMessage.MessageType)), destination }); |
| } |
| if (casProcessRequest) { |
| cacheEntry = (ClientRequest) engine.getCache().get( |
| pm.get(AsynchAEMessage.CasReference)); |
| if (cacheEntry != null) { |
| // Use Process Timeout value for the time-to-live property in the |
| // outgoing JMS message. When this time is exceeded |
| // while the message sits in a queue, the JMS Server will remove it from |
| // the queue. What happens with the expired message depends on the |
| // configuration. Most JMS Providers create a special dead-letter queue |
| // where all expired messages are placed. NOTE: In ActiveMQ expired msgs in the DLQ |
| // are not auto evicted yet and accumulate taking up memory. |
| long timeoutValue = cacheEntry.getProcessTimeout(); |
| |
| if (timeoutValue > 0 && addTimeToLive ) { |
| // Set high time to live value |
| message.setJMSExpiration(10 * timeoutValue); |
| } |
| if (pm.getMessageType() == AsynchAEMessage.Process) { |
| cacheEntry.setCASDepartureTime(System.nanoTime()); |
| } |
| cacheEntry.setCASDepartureTime(System.nanoTime()); |
| UimaASProcessStatus status = new UimaASProcessStatusImpl(new ProcessTrace_impl(), |
| cacheEntry.getCasReferenceId()); |
| // Notify engine before sending a message |
| engine.onBeforeMessageSend(status); |
| } |
| } |
| // start timers |
| if( casProcessRequest ) { |
| // Add the cas to a list of CASes pending reply. Also start the timer if necessary |
| engine.serviceDelegate.addCasToOutstandingList(cacheEntry.getCasReferenceId()); |
| } else if ( pm.getMessageType() == AsynchAEMessage.GetMeta && |
| engine.serviceDelegate.getGetMetaTimeout() > 0 ) { |
| // timer for PING has been started in sendCAS() |
| if ( !engine.serviceDelegate.isAwaitingPingReply()) { |
| engine.serviceDelegate.startGetMetaRequestTimer(); |
| } |
| } |
| // Dispatch asynchronous request to Uima AS service |
| producer.send(message); |
| |
| } catch (Exception e) { |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, getClass().getName(), |
| "run", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAEE_exception__WARNING", new Object[] { e }); |
| } |
| reject(pm,e); |
| } |
| |
| } |
| } |
| try { |
| cleanup(); |
| } catch (Exception e) { |
| handleException(e, destination); |
| } |
| } |
| |
| private void initializeMessage(PendingMessage aPm, Message anOutgoingMessage) throws Exception { |
| // Populate message properties based on outgoing message type |
| switch (aPm.getMessageType()) { |
| case AsynchAEMessage.GetMeta: |
| engine.setMetaRequestMessage(anOutgoingMessage); |
| break; |
| |
| case AsynchAEMessage.Process: |
| String casReferenceId = (String) aPm.get(AsynchAEMessage.CasReference); |
| if (engine.getSerializationStrategy().equals("xmi")) { |
| String serializedCAS = (String) aPm.get(AsynchAEMessage.CAS); |
| engine.setCASMessage(casReferenceId, serializedCAS, anOutgoingMessage); |
| } else { |
| byte[] serializedCAS = (byte[]) aPm.get(AsynchAEMessage.CAS); |
| engine.setCASMessage(casReferenceId, serializedCAS, anOutgoingMessage); |
| |
| } |
| // Message Expiration for Process is added in the main run() loop |
| // right before the message is dispatched to the AS Service |
| break; |
| |
| case AsynchAEMessage.CollectionProcessComplete: |
| engine.setCPCMessage(anOutgoingMessage); |
| break; |
| } |
| } |
| |
| private boolean isProcessRequest(PendingMessage pm) { |
| return pm.getMessageType() == AsynchAEMessage.Process; |
| } |
| |
| private void handleException(Exception e, String aDestination) { |
| workerThreadFailed = true; |
| exception = e; |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, getClass().getName(), |
| "handleException", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAEE_exception__WARNING", new Object[] { e }); |
| } |
| engine.recoverSharedConnectionIfClosed(); |
| // Notify the engine that there was an exception. |
| engine.onException(e, aDestination); |
| |
| } |
| |
| /** |
| * @override |
| */ |
| public MessageProducer getMessageProducer(Destination destination) throws Exception { |
| return null; |
| } |
| |
| } |
