| /* |
| * 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.activemq; |
| |
| import java.io.BufferedOutputStream; |
| import java.io.BufferedWriter; |
| import java.io.ByteArrayOutputStream; |
| import java.io.NotSerializableException; |
| import java.io.PrintWriter; |
| import java.io.StringWriter; |
| import java.net.ConnectException; |
| import java.net.InetAddress; |
| import java.util.Date; |
| import java.util.Iterator; |
| import java.util.Map; |
| import java.util.Set; |
| import java.util.StringTokenizer; |
| import java.util.Timer; |
| import java.util.TimerTask; |
| import java.util.Map.Entry; |
| import java.util.concurrent.ConcurrentHashMap; |
| import java.util.concurrent.CountDownLatch; |
| |
| import javax.jms.BytesMessage; |
| import javax.jms.Connection; |
| import javax.jms.Destination; |
| import javax.jms.JMSException; |
| import javax.jms.Message; |
| import javax.jms.MessageProducer; |
| import javax.jms.ObjectMessage; |
| import javax.jms.TextMessage; |
| import javax.management.ServiceNotFoundException; |
| |
| import org.apache.activemq.ActiveMQConnection; |
| import org.apache.activemq.ActiveMQConnectionFactory; |
| import org.apache.activemq.command.ActiveMQDestination; |
| import org.apache.uima.UIMAFramework; |
| import org.apache.uima.aae.Channel; |
| import org.apache.uima.aae.InputChannel; |
| import org.apache.uima.aae.OutputChannel; |
| import org.apache.uima.aae.UIMAEE_Constants; |
| import org.apache.uima.aae.UimaSerializer; |
| import org.apache.uima.aae.InProcessCache.CacheEntry; |
| import org.apache.uima.aae.controller.AggregateAnalysisEngineController; |
| import org.apache.uima.aae.controller.AnalysisEngineController; |
| import org.apache.uima.aae.controller.Endpoint; |
| import org.apache.uima.aae.controller.PrimitiveAnalysisEngineController; |
| import org.apache.uima.aae.controller.LocalCache.CasStateEntry; |
| import org.apache.uima.aae.delegate.Delegate; |
| import org.apache.uima.aae.error.AsynchAEException; |
| import org.apache.uima.aae.error.ErrorContext; |
| import org.apache.uima.aae.error.MessageTimeoutException; |
| import org.apache.uima.aae.error.ServiceShutdownException; |
| import org.apache.uima.aae.error.UimaEEServiceException; |
| import org.apache.uima.aae.jmx.ServiceInfo; |
| import org.apache.uima.aae.jmx.ServicePerformance; |
| import org.apache.uima.aae.message.AsynchAEMessage; |
| import org.apache.uima.aae.message.UIMAMessage; |
| import org.apache.uima.aae.monitor.Monitor; |
| import org.apache.uima.aae.monitor.statistics.LongNumericStatistic; |
| import org.apache.uima.adapter.jms.JmsConstants; |
| import org.apache.uima.cas.CAS; |
| import org.apache.uima.cas.impl.XmiSerializationSharedData; |
| import org.apache.uima.resource.metadata.ProcessingResourceMetaData; |
| import org.apache.uima.util.Level; |
| |
| public class JmsOutputChannel implements OutputChannel { |
| |
| private static final Class CLASS_NAME = JmsOutputChannel.class; |
| |
| private static final long INACTIVITY_TIMEOUT = 1800000; // 30 minutes in term of millis |
| |
| private CountDownLatch controllerLatch = new CountDownLatch(1); |
| |
| private ActiveMQConnectionFactory connectionFactory; |
| |
| // Name of the external queue this service uses to receive messages |
| private String serviceInputEndpoint; |
| |
| // Name of the internal queue this services uses to receive messages from delegates |
| private String controllerInputEndpoint; |
| |
| // Name of the queue used by Cas Multiplier to receive requests to free CASes |
| private String secondaryInputEndpoint; |
| |
| // The service controller |
| private AnalysisEngineController analysisEngineController; |
| |
| // Cache containing connections to destinations this service interacts with |
| // Each entry in this cache has an inactivity timer that times amount of time |
| // elapsed since the last time a message was sent to the destination. |
| private ConcurrentHashMap connectionMap = new ConcurrentHashMap(); |
| |
| private String serverURI; |
| |
| private String serviceProtocolList = ""; |
| |
| private volatile boolean aborting = false; |
| |
| private Destination freeCASTempQueue; |
| |
| private String hostIP = null; |
| |
| private UimaSerializer uimaSerializer = new UimaSerializer(); |
| |
| // By default every message will have expiration time added |
| private volatile boolean addTimeToLive = true; |
| |
| public JmsOutputChannel() { |
| try { |
| hostIP = InetAddress.getLocalHost().getHostAddress(); |
| } catch (Exception e) { /* silently deal with this error */ |
| } |
| // Check the environment for existence of NoTTL tag. If present, |
| // the deployer of the service wants to avoid message expiration. |
| if (System.getProperty("NoTTL") != null) { |
| addTimeToLive = false; |
| } |
| |
| } |
| |
| /** |
| * Sets the ActiveMQ Broker URI |
| */ |
| public void setServerURI(String aServerURI) { |
| serverURI = aServerURI; |
| } |
| |
| protected void setFreeCasQueue(Destination destination) { |
| freeCASTempQueue = destination; |
| } |
| |
| public String getServerURI() { |
| return serverURI; |
| } |
| |
| public String getName() { |
| return ""; |
| } |
| |
| /** |
| * |
| * @param connectionFactory |
| */ |
| public void setConnectionFactory(ActiveMQConnectionFactory connectionFactory) { |
| this.connectionFactory = connectionFactory; |
| } |
| |
| public void setServiceInputEndpoint(String anEnpoint) { |
| serviceInputEndpoint = anEnpoint; |
| } |
| |
| public void setSecondaryInputQueue(String anEndpoint) { |
| secondaryInputEndpoint = anEndpoint; |
| } |
| |
| public ActiveMQConnectionFactory getConnectionFactory() { |
| return this.connectionFactory; |
| } |
| |
| public void initialize() throws AsynchAEException { |
| if (getAnalysisEngineController() instanceof AggregateAnalysisEngineController) { |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINE)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, CLASS_NAME.getName(), "initialize", |
| JmsConstants.JMS_LOG_RESOURCE_BUNDLE, "UIMAJMS_connector_list__FINE", |
| new Object[] { System.getProperty("ActiveMQConnectors") }); |
| } |
| // Aggregate controller set this System property at startup in |
| serviceProtocolList = System.getProperty("ActiveMQConnectors"); |
| } |
| } |
| |
| /** |
| * Serializes CAS using indicated Serializer. |
| * |
| * @param aCAS |
| * - CAS instance to serialize |
| * @param aSerializerKey |
| * - a key identifying which serializer to use |
| * @return - String - serialized CAS as String |
| * @throws Exception |
| */ |
| public String serializeCAS(boolean isReply, CAS aCAS, String aCasReferenceId, |
| String aSerializerKey) throws Exception { |
| |
| long start = getAnalysisEngineController().getCpuTime(); |
| |
| String serializedCas = null; |
| |
| if (isReply || "xmi".equalsIgnoreCase(aSerializerKey)) { |
| CacheEntry cacheEntry = getAnalysisEngineController().getInProcessCache() |
| .getCacheEntryForCAS(aCasReferenceId); |
| |
| XmiSerializationSharedData serSharedData; |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINE)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, CLASS_NAME.getName(), "serializeCAS", |
| JmsConstants.JMS_LOG_RESOURCE_BUNDLE, "UIMAJMS_serialize_cas__FINE", |
| new Object[] { aCasReferenceId }); |
| } |
| if (isReply) { |
| serSharedData = cacheEntry.getDeserSharedData(); |
| if (cacheEntry.acceptsDeltaCas() |
| && (cacheEntry.getMarker() != null && cacheEntry.getMarker().isValid())) { |
| serializedCas = uimaSerializer.serializeCasToXmi(aCAS, serSharedData, cacheEntry |
| .getMarker()); |
| cacheEntry.setSentDeltaCas(true); |
| } else { |
| serializedCas = uimaSerializer.serializeCasToXmi(aCAS, serSharedData); |
| cacheEntry.setSentDeltaCas(false); |
| } |
| // if market is invalid, create a fresh marker. |
| if (cacheEntry.getMarker() != null && !cacheEntry.getMarker().isValid()) { |
| cacheEntry.setMarker(aCAS.createMarker()); |
| } |
| } else { |
| serSharedData = cacheEntry.getDeserSharedData(); |
| if (serSharedData == null) { |
| serSharedData = new XmiSerializationSharedData(); |
| cacheEntry.setXmiSerializationData(serSharedData); |
| } |
| serializedCas = uimaSerializer.serializeCasToXmi(aCAS, serSharedData); |
| int maxOutgoingXmiId = serSharedData.getMaxXmiId(); |
| // Save High Water Mark in case a merge is needed |
| getAnalysisEngineController().getInProcessCache().getCacheEntryForCAS(aCasReferenceId) |
| .setHighWaterMark(maxOutgoingXmiId); |
| } |
| |
| } else if ("xcas".equalsIgnoreCase(aSerializerKey)) { |
| // Default is XCAS |
| ByteArrayOutputStream bos = new ByteArrayOutputStream(); |
| try { |
| uimaSerializer.serializeToXCAS(bos, aCAS, null, null, null); |
| serializedCas = bos.toString(); |
| } catch (Exception e) { |
| throw e; |
| } finally { |
| bos.close(); |
| } |
| } |
| |
| LongNumericStatistic statistic; |
| if ((statistic = getAnalysisEngineController().getMonitor().getLongNumericStatistic("", |
| Monitor.TotalSerializeTime)) != null) { |
| statistic.increment(getAnalysisEngineController().getCpuTime() - start); |
| } |
| |
| return serializedCas; |
| } |
| |
| /** |
| * This method verifies that the destination (queue) exists. It opens a connection the a broker, |
| * creates a session and a message producer. Finally, using the message producer, sends an empty |
| * message to a queue. This API support enables checking for existence of the reply (temp) queue |
| * before any processing of a cas is done. This is an optimization to prevent expensive processing |
| * if the client destination is no longer available. |
| */ |
| public void bindWithClientEndpoint(Endpoint anEndpoint) throws Exception { |
| // check if the reply endpoint is a temp destination |
| if (anEndpoint.getDestination() != null) { |
| // create message producer if one doesnt exist for this destination |
| JmsEndpointConnection_impl endpointConnection = getEndpointConnection(anEndpoint); |
| // Create empty message |
| TextMessage tm = endpointConnection.produceTextMessage(""); |
| // test sending a message to reply endpoint. This tests existence of |
| // a temp queue. If the client has been shutdown, this will fail |
| // with an exception. |
| endpointConnection.send(tm, 0, false); |
| } |
| } |
| |
| private long getInactivityTimeout(String destination, String brokerURL) { |
| if (System.getProperty(JmsConstants.SessionTimeoutOverride) != null) { |
| try { |
| long overrideTimeoutValue = Long.parseLong(System |
| .getProperty(JmsConstants.SessionTimeoutOverride)); |
| // endpointConnection.setInactivityTimeout(overrideTimeoutValue); // If the connection is |
| // not used within this interval it will be removed |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINE)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb( |
| Level.FINE, |
| CLASS_NAME.getName(), |
| "getEndpointConnection", |
| JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_override_connection_timeout__FINE", |
| new Object[] { analysisEngineController, overrideTimeoutValue, destination, |
| brokerURL }); |
| } |
| return overrideTimeoutValue; |
| } catch (NumberFormatException e) { |
| /* ignore. use the default */} |
| } else { |
| // endpointConnection.setInactivityTimeout(INACTIVITY_TIMEOUT); // If the connection is not |
| // used within this interval it will be removed |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINE)) { |
| UIMAFramework.getLogger(CLASS_NAME) |
| .logrb( |
| Level.FINE, |
| CLASS_NAME.getName(), |
| "getEndpointConnection", |
| JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_connection_timeout__FINE", |
| new Object[] { analysisEngineController, INACTIVITY_TIMEOUT, destination, |
| brokerURL }); |
| } |
| } |
| return (int) INACTIVITY_TIMEOUT; // default |
| } |
| /** |
| * Stop JMS connection and close all sessions associated with this connection |
| * |
| * @param brokerConnectionEntry |
| */ |
| private void invalidateConnectionAndEndpoints(BrokerConnectionEntry brokerConnectionEntry ) { |
| Connection conn = brokerConnectionEntry.getConnection(); |
| try { |
| if ( conn != null && ((ActiveMQConnection)conn).isClosed()) { |
| brokerConnectionEntry.getConnection().stop(); |
| brokerConnectionEntry.getConnection().close(); |
| brokerConnectionEntry.setConnection(null); |
| for (Entry<Object, JmsEndpointConnection_impl> endpoints : brokerConnectionEntry.endpointMap |
| .entrySet()) { |
| endpoints.getValue().close(); // close session and producer |
| } |
| } |
| } catch (Exception e) { |
| // Ignore this for now. Attempting to close connection that has been closed |
| // Ignore we are shutting down |
| } finally { |
| brokerConnectionEntry.endpointMap.clear(); |
| connectionMap.remove(brokerConnectionEntry.getBrokerURL()); |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb( |
| Level.INFO, |
| CLASS_NAME.getName(), |
| "invalidateConnectionAndEndpoints", |
| JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_service_closing_connection__INFO", |
| new Object[] { getAnalysisEngineController().getComponentName(), |
| brokerConnectionEntry.getBrokerURL() }); |
| } |
| } |
| brokerConnectionEntry.setConnection(null); |
| |
| } |
| private String getDestinationName(Endpoint anEndpoint) { |
| String destination = anEndpoint.getEndpoint(); |
| if (anEndpoint.getDestination() != null |
| && anEndpoint.getDestination() instanceof ActiveMQDestination) { |
| destination = ((ActiveMQDestination) anEndpoint.getDestination()).getPhysicalName(); |
| } |
| return destination; |
| } |
| private String getLookupKey(Endpoint anEndpoint) { |
| String key = anEndpoint.getEndpoint() + anEndpoint.getServerURI(); |
| String destination = getDestinationName(anEndpoint); |
| if ( anEndpoint.getDelegateKey() != null ) { |
| key = anEndpoint.getDelegateKey() + "-"+destination; |
| } else { |
| key = "Client-"+destination; |
| } |
| return key; |
| } |
| private BrokerConnectionEntry createConnectionEntry(String brokerURL) { |
| BrokerConnectionEntry brokerConnectionEntry = new BrokerConnectionEntry(); |
| connectionMap.put(brokerURL, brokerConnectionEntry); |
| ConnectionTimer connectionTimer = new ConnectionTimer(brokerConnectionEntry); |
| connectionTimer.setAnalysisEngineController(getAnalysisEngineController()); |
| brokerConnectionEntry.setConnectionTimer(connectionTimer); |
| return brokerConnectionEntry; |
| } |
| /** |
| * Returns {@link JmsEndpointConnection_impl} instance bound to a destination defined in the |
| * {@link Endpoint} The endpoint identifies the destination that should receive the message. This |
| * method refrences a cache that stores active connections. Active connections are those that are |
| * fully bound and being used for communication. The key to locate the entry in the connection |
| * cache is the queue name + broker URI. This uniquely identifies the destination. If an entry |
| * does not exist in the cache, this routine will create a new connection, initialize it, and |
| * cache it for future use. The cache is purely for optimization, to prevent openinig a connection |
| * for every message which is a costly operation. Instead the connection is open, cached and |
| * reused. The {@link JmsEndpointConnection_impl} instance is stored in the cache, and uses a |
| * timer to make sure stale connection are removed. If a connection is not used in a given time |
| * interval, the connection is considered stale and is dropped from the cache. |
| * |
| * @param anEndpoint |
| * - endpoint configuration containing connection information to a destination |
| * @return - |
| * @throws AsynchAEException |
| */ |
| private synchronized JmsEndpointConnection_impl getEndpointConnection(Endpoint anEndpoint) |
| throws AsynchAEException, ServiceShutdownException, ConnectException { |
| |
| try { |
| controllerLatch.await(); |
| } catch (InterruptedException e) { |
| } |
| |
| JmsEndpointConnection_impl endpointConnection = null; |
| |
| // First get a Map containing destinations managed by a broker provided by the client |
| BrokerConnectionEntry brokerConnectionEntry = null; |
| if (connectionMap.containsKey(anEndpoint.getServerURI())) { |
| brokerConnectionEntry = (BrokerConnectionEntry) connectionMap.get(anEndpoint.getServerURI()); |
| // Findbugs thinks that the above may return null, perhaps due to a race condition. Add |
| // the null check just in case |
| if (brokerConnectionEntry == null) { |
| throw new AsynchAEException("Controller:" |
| + getAnalysisEngineController().getComponentName() |
| + " Unable to Lookup Broker Connection For URL:" + anEndpoint.getServerURI()); |
| } |
| brokerConnectionEntry.setBrokerURL(anEndpoint.getServerURI()); |
| if ( JmsEndpointConnection_impl.connectionClosedOrFailed(brokerConnectionEntry) ) { |
| invalidateConnectionAndEndpoints(brokerConnectionEntry); |
| brokerConnectionEntry = createConnectionEntry(anEndpoint.getServerURI()); |
| } |
| } else { |
| brokerConnectionEntry = createConnectionEntry(anEndpoint.getServerURI()); |
| } |
| String key = getLookupKey(anEndpoint); |
| String destination = getDestinationName(anEndpoint); |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINE)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb( |
| Level.FINE, |
| CLASS_NAME.getName(), |
| "getEndpointConnection", |
| JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_acquiring_connection_to_endpoint__FINE", |
| new Object[] { getAnalysisEngineController().getComponentName(), destination, |
| anEndpoint.getServerURI() }); |
| } |
| |
| // check the cache first |
| if (!brokerConnectionEntry.endpointExists(key)) { |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINE)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb( |
| Level.FINE, |
| CLASS_NAME.getName(), |
| "getEndpointConnection", |
| JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_create_new_connection__FINE", |
| new Object[] { getAnalysisEngineController().getComponentName(), destination, |
| anEndpoint.getServerURI() }); |
| } |
| endpointConnection = new JmsEndpointConnection_impl(brokerConnectionEntry, anEndpoint, |
| getAnalysisEngineController()); |
| brokerConnectionEntry.addEndpointConnection(key, endpointConnection); |
| long replyQueueInactivityTimeout = getInactivityTimeout(destination, anEndpoint |
| .getServerURI()); |
| brokerConnectionEntry.getConnectionTimer().setInactivityTimeout(replyQueueInactivityTimeout); |
| |
| // Connection is not in the cache, create a new connection, initialize it and cache it |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINE)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, CLASS_NAME.getName(), |
| "getEndpointConnection", JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_open_new_connection_to_endpoint__FINE", |
| new Object[] { getDestinationName(anEndpoint), anEndpoint.getServerURI() }); |
| } |
| |
| /** |
| * Open connection to a broker, create JMS session and MessageProducer |
| */ |
| endpointConnection.open(); |
| |
| brokerConnectionEntry.getConnectionTimer().setConnectionCreationTimestamp( |
| endpointConnection.connectionCreationTimestamp); |
| |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINE)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb( |
| Level.FINE, |
| CLASS_NAME.getName(), |
| "getEndpointConnection", |
| JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_connection_opened_to_endpoint__FINE", |
| new Object[] { getAnalysisEngineController().getComponentName(), destination, |
| anEndpoint.getServerURI() }); |
| } |
| } else { |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINE)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb( |
| Level.FINE, |
| CLASS_NAME.getName(), |
| "getEndpointConnection", |
| JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_reusing_existing_connection__FINE", |
| new Object[] { getAnalysisEngineController().getComponentName(), destination, |
| anEndpoint.getServerURI() }); |
| } |
| // Retrieve connection from the connection cache |
| endpointConnection = brokerConnectionEntry.getEndpointConnection(key); |
| // check the state of the connection and re-open it if necessary |
| if (endpointConnection != null && !endpointConnection.isOpen()) { |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINE)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, CLASS_NAME.getName(), |
| "getEndpointConnection", JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_connection_closed_reopening_endpoint__FINE", |
| new Object[] { destination }); |
| } |
| endpointConnection.open(); |
| brokerConnectionEntry.getConnectionTimer() |
| .setConnectionCreationTimestamp(System.nanoTime()); |
| } |
| } |
| return endpointConnection; |
| } |
| |
| /** |
| * Sends request message to a delegate. |
| * |
| * @param aCommand |
| * - the type of request [Process|GetMeta] |
| * @param anEndpoint |
| * - the destination where the delegate receives messages |
| * |
| * @throws AsynchAEException |
| */ |
| public void sendRequest(int aCommand, String aCasReferenceId, Endpoint anEndpoint) |
| throws AsynchAEException { |
| try { |
| |
| JmsEndpointConnection_impl endpointConnection = getEndpointConnection(anEndpoint); |
| |
| TextMessage tm = endpointConnection.produceTextMessage(""); |
| tm.setIntProperty(AsynchAEMessage.Payload, AsynchAEMessage.None); |
| tm.setStringProperty(AsynchAEMessage.CasReference, aCasReferenceId); |
| populateHeaderWithRequestContext(tm, anEndpoint, aCommand); |
| if (aCommand == AsynchAEMessage.ReleaseCAS || aCommand == AsynchAEMessage.Stop) { |
| |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINE)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb( |
| Level.FINE, |
| CLASS_NAME.getName(), |
| "sendRequest", |
| JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_release_cas_req__FINE", |
| new Object[] { getAnalysisEngineController().getName(), anEndpoint.getEndpoint(), |
| aCasReferenceId }); |
| } |
| } |
| // Only used to send a Stop or ReleaseCas request so probably no need to start a connection |
| // timer ? |
| endpointConnection.send(tm, 0, true); |
| } catch (JMSException e) { |
| // Unable to establish connection to the endpoint. Logit and continue |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINE)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(), "sendRequest", |
| JmsConstants.JMS_LOG_RESOURCE_BUNDLE, "UIMAJMS_unable_to_connect__INFO", |
| new Object[] { getAnalysisEngineController().getName(), anEndpoint.getEndpoint() }); |
| } |
| } catch (ServiceShutdownException e) { |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(), "sendRequest", |
| JmsConstants.JMS_LOG_RESOURCE_BUNDLE, "UIMAJMS_jms_shutdown__INFO", |
| new Object[] { getAnalysisEngineController().getName(), anEndpoint.getEndpoint() }); |
| } |
| } catch (Exception e) { |
| throw new AsynchAEException(e); |
| } |
| } |
| |
| /** |
| * Sends request message to a delegate. |
| * |
| * @param aCommand |
| * - the type of request [Process|GetMeta] |
| * @param anEndpoint |
| * - the destination where the delegate receives messages |
| * |
| * @throws AsynchAEException |
| */ |
| public void sendRequest(int aCommand, Endpoint anEndpoint) { |
| Delegate delegate = null; |
| try { |
| JmsEndpointConnection_impl endpointConnection = getEndpointConnection(anEndpoint); |
| |
| TextMessage tm = endpointConnection.produceTextMessage(""); |
| tm.setIntProperty(AsynchAEMessage.Payload, AsynchAEMessage.None); |
| tm.setText(""); // Need this to prevent the Broker from throwing an exception when sending a |
| // message to C++ service |
| |
| populateHeaderWithRequestContext(tm, anEndpoint, aCommand); |
| |
| // For remotes add a special property to the message. This property |
| // will be echoed back by the service. This property enables matching |
| // the reply with the right endpoint object managed by the aggregate. |
| if (anEndpoint.isRemote()) { |
| tm.setStringProperty(AsynchAEMessage.EndpointServer, anEndpoint.getServerURI()); |
| } |
| boolean startTimer = false; |
| // Start timer for endpoints that are remote and are managed by a different broker |
| // than this service. If an endpoint contains a destination object, the outgoing |
| // request will contain a JMSReplyTo object which will point to a temp queue |
| if (anEndpoint.isRemote() && anEndpoint.getDestination() == null) { |
| startTimer = true; |
| } |
| if (aCommand == AsynchAEMessage.CollectionProcessComplete) { |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINE)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, CLASS_NAME.getName(), |
| "sendRequest", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAEE_send_cpc_req__FINE", new Object[] { anEndpoint.getEndpoint() }); |
| } |
| } else if (aCommand == AsynchAEMessage.ReleaseCAS) { |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINEST)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb( |
| Level.FINEST, |
| CLASS_NAME.getName(), |
| "sendRequest", |
| JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_releasecas_request__endpoint__FINEST", |
| new Object[] { getAnalysisEngineController().getName(), |
| endpointConnection.getEndpoint() }); |
| } |
| } else if (aCommand == AsynchAEMessage.GetMeta) { |
| if (anEndpoint.getDestination() != null) { |
| String replyQueueName = ((ActiveMQDestination) anEndpoint.getDestination()) |
| .getPhysicalName().replaceAll(":", "_"); |
| if (getAnalysisEngineController() instanceof AggregateAnalysisEngineController) { |
| String delegateKey = ((AggregateAnalysisEngineController) getAnalysisEngineController()) |
| .lookUpDelegateKey(anEndpoint.getEndpoint()); |
| ServiceInfo serviceInfo = ((AggregateAnalysisEngineController) getAnalysisEngineController()) |
| .getDelegateServiceInfo(delegateKey); |
| if (serviceInfo != null) { |
| serviceInfo.setReplyQueueName(replyQueueName); |
| serviceInfo.setServiceKey(delegateKey); |
| } |
| delegate = lookupDelegate(delegateKey); |
| if (delegate.getGetMetaTimeout() > 0) { |
| delegate.startGetMetaRequestTimer(); |
| } |
| } |
| } else if (!anEndpoint.isRemote()) { |
| ServiceInfo serviceInfo = ((AggregateAnalysisEngineController) getAnalysisEngineController()) |
| .getServiceInfo(); |
| if (serviceInfo != null) { |
| serviceInfo.setReplyQueueName(controllerInputEndpoint); |
| } |
| } |
| } else { |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINEST)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb( |
| Level.FINEST, |
| CLASS_NAME.getName(), |
| "sendRequest", |
| JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_metadata_request__endpoint__FINEST", |
| new Object[] { endpointConnection.getEndpoint(), |
| endpointConnection.getServerUri() }); |
| } |
| } |
| if (endpointConnection.send(tm, 0, startTimer) != true) { |
| throw new ServiceNotFoundException(); |
| } |
| |
| } catch (Exception e) { |
| if (delegate != null && aCommand == AsynchAEMessage.GetMeta) { |
| delegate.cancelDelegateTimer(); |
| } |
| // Handle the error |
| ErrorContext errorContext = new ErrorContext(); |
| errorContext.add(AsynchAEMessage.Command, aCommand); |
| errorContext.add(AsynchAEMessage.Endpoint, anEndpoint); |
| getAnalysisEngineController().getErrorHandlerChain().handle(e, errorContext, |
| getAnalysisEngineController()); |
| } |
| } |
| |
| public void sendRequest(String aCasReferenceId, Endpoint anEndpoint) throws AsynchAEException { |
| |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINE)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, CLASS_NAME.getName(), "sendRequest", |
| JmsConstants.JMS_LOG_RESOURCE_BUNDLE, "UIMAJMS_sending_msg_to_endpoint__FINE", |
| new Object[] { anEndpoint.getEndpoint() }); |
| } |
| try { |
| if (anEndpoint.isRemote()) { |
| if (anEndpoint.getSerializer().equals("xmi")) { |
| String serializedCAS = getSerializedCasAndReleaseIt(false, aCasReferenceId, anEndpoint, |
| anEndpoint.isRetryEnabled()); |
| if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb( |
| Level.FINEST, |
| CLASS_NAME.getName(), |
| "sendRequest", |
| JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_sending_serialized_cas__FINEST", |
| new Object[] { getAnalysisEngineController().getComponentName(), |
| anEndpoint.getEndpoint(), aCasReferenceId, serializedCAS }); |
| } |
| // Send process request to remote delegate and start timeout timer |
| sendCasToRemoteEndpoint(true, serializedCAS, null, aCasReferenceId, anEndpoint, true, 0); |
| } else { |
| byte[] serializedCAS = getBinaryCasAndReleaseIt(false, aCasReferenceId, anEndpoint, |
| anEndpoint.isRetryEnabled()); |
| if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb( |
| Level.FINEST, |
| CLASS_NAME.getName(), |
| "sendRequest", |
| JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_sending_binary_cas__FINEST", |
| new Object[] { getAnalysisEngineController().getComponentName(), |
| anEndpoint.getEndpoint(), aCasReferenceId, serializedCAS }); |
| } |
| |
| // Send process request to remote delegate and start timeout timer |
| sendCasToRemoteEndpoint(true, serializedCAS, null, aCasReferenceId, anEndpoint, true, 0); |
| |
| } |
| |
| } else { |
| // Not supported |
| } |
| } catch (ServiceShutdownException e) { |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(), |
| "sendRequest", JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_exception__WARNING", new Object[] { JmsConstants.threadName(), e }); |
| } |
| } |
| |
| catch (AsynchAEException e) { |
| throw e; |
| } catch (Exception e) { |
| throw new AsynchAEException(e); |
| } |
| } |
| |
| /** |
| * Sends request message to process CAS to the given destinations. This method enables processing |
| * of the same CAS in multiple Analysis Engines at the same time. |
| * |
| * @param aCasReferenceId |
| * @param anEndpoint |
| * @throws AsynchAEException |
| */ |
| public void sendRequest(String aCasReferenceId, Endpoint[] endpoints) throws AsynchAEException { |
| Endpoint currentEndpoint = null; |
| try { |
| boolean cacheSerializedCas = endpointRetryEnabled(endpoints); |
| // The default serialization strategy for parallel step is xmi. |
| // Binary serialization doesnt support merge. |
| endpoints[0].setSerializer("xmi"); |
| |
| // Serialize CAS using serializer defined in the first endpoint. All endpoints in the parallel |
| // Flow |
| // must use the same format (either XCAS or XMI) |
| String serializedCAS = getSerializedCasAndReleaseIt(false, aCasReferenceId, endpoints[0], |
| cacheSerializedCas); |
| // Using provided endpoints, create JMS message for each destination and sent the serialized |
| // CAS to it. |
| for (int i = 0; i < endpoints.length; i++) { |
| // For remote delegates, optionally cache serialized CAS in case a retry on timeout is |
| // required. |
| if (endpoints[i].isRemote()) { |
| |
| currentEndpoint = endpoints[i]; |
| if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb( |
| Level.FINEST, |
| CLASS_NAME.getName(), |
| "sendRequest", |
| JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_sending_serialized_cas__FINEST", |
| new Object[] { getAnalysisEngineController().getComponentName(), |
| endpoints[i].getEndpoint(), aCasReferenceId, serializedCAS }); |
| } |
| |
| // The default serialization strategy for parallel step is xmi. |
| // Binary serialization doesnt support merge. |
| endpoints[i].setSerializer("xmi"); |
| |
| sendCasToRemoteEndpoint(true, serializedCAS, null, aCasReferenceId, endpoints[i], true, 0); |
| } else { |
| // Currently this use case is not supported. Parallel processing of CAS is only supported |
| // with remote Delegates |
| } |
| } |
| } catch (Exception e) { |
| // Handle the error |
| ErrorContext errorContext = new ErrorContext(); |
| errorContext.add(AsynchAEMessage.Command, AsynchAEMessage.Process); |
| errorContext.add(AsynchAEMessage.Endpoint, currentEndpoint); |
| errorContext.add(AsynchAEMessage.CasReference, aCasReferenceId); |
| getAnalysisEngineController().getErrorHandlerChain().handle(e, errorContext, |
| getAnalysisEngineController()); |
| } |
| } |
| |
| public void sendReply(CAS aCas, String anInputCasReferenceId, String aNewCasReferenceId, |
| Endpoint anEndpoint, long sequence) throws AsynchAEException { |
| try { |
| anEndpoint.setReplyEndpoint(true); |
| if (anEndpoint.isRemote()) { |
| // Serializes CAS and releases it back to CAS Pool |
| String serializedCAS = getSerializedCas(true, aNewCasReferenceId, anEndpoint, anEndpoint |
| .isRetryEnabled()); |
| sendCasToRemoteEndpoint(false, serializedCAS, anInputCasReferenceId, aNewCasReferenceId, |
| anEndpoint, false, sequence); |
| } else { |
| // Not supported |
| } |
| } catch (ServiceShutdownException e) { |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(), |
| "sendReply", JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_exception__WARNING", new Object[] { JmsConstants.threadName(), e }); |
| } |
| } catch (AsynchAEException e) { |
| throw e; |
| } |
| |
| catch (Exception e) { |
| throw new AsynchAEException(e); |
| } |
| |
| } |
| |
| public void sendReply(CacheEntry entry, Endpoint anEndpoint) throws AsynchAEException { |
| try { |
| anEndpoint.setReplyEndpoint(true); |
| if (anEndpoint.isRemote()) { |
| if (anEndpoint.getSerializer().equals("xmi")) { |
| // Serializes CAS and releases it back to CAS Pool |
| String serializedCAS = getSerializedCas(true, entry.getCasReferenceId(), anEndpoint, |
| anEndpoint.isRetryEnabled()); |
| sendCasToRemoteEndpoint(false, serializedCAS, entry, anEndpoint, false); |
| } else { |
| byte[] binaryCas = getBinaryCas(true, entry.getCasReferenceId(), anEndpoint, anEndpoint |
| .isRetryEnabled()); |
| if (binaryCas == null) { |
| return; |
| } |
| sendCasToRemoteEndpoint(false, binaryCas, entry, anEndpoint, false); |
| } |
| |
| } else { |
| // Not supported |
| } |
| } catch (ServiceShutdownException e) { |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(), |
| "sendReply", JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_exception__WARNING", new Object[] { JmsConstants.threadName(), e }); |
| } |
| } catch (AsynchAEException e) { |
| throw e; |
| } |
| |
| catch (Exception e) { |
| throw new AsynchAEException(e); |
| } |
| |
| } |
| |
| public void sendReply(int aCommand, Endpoint anEndpoint, String aCasReferenceId) |
| throws AsynchAEException { |
| try { |
| if (aborting) { |
| return; |
| } |
| anEndpoint.setReplyEndpoint(true); |
| JmsEndpointConnection_impl endpointConnection = getEndpointConnection(anEndpoint); |
| |
| TextMessage tm = endpointConnection.produceTextMessage(""); |
| tm.setIntProperty(AsynchAEMessage.Payload, AsynchAEMessage.None); |
| populateHeaderWithResponseContext(tm, anEndpoint, aCommand); |
| tm.setStringProperty(AsynchAEMessage.CasReference, aCasReferenceId); |
| |
| // If this service is a Cas Multiplier add to the message a FreeCasQueue. |
| // The client may need send Stop request to that queue. |
| if (aCommand == AsynchAEMessage.ServiceInfo |
| && getAnalysisEngineController().isCasMultiplier() && freeCASTempQueue != null) { |
| // Attach a temp queue to the outgoing message. This a queue where |
| // Free CAS notifications need to be sent from the client |
| tm.setJMSReplyTo(freeCASTempQueue); |
| } |
| endpointConnection.send(tm, 0, false); |
| addIdleTime(tm); |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINE)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb( |
| Level.FINE, |
| CLASS_NAME.getName(), |
| "sendReply", |
| JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_cpc_reply_sent__FINE", |
| new Object[] { getAnalysisEngineController().getComponentName(), |
| anEndpoint.getEndpoint() }); |
| } |
| } catch (JMSException e) { |
| // Unable to establish connection to the endpoint. Logit and continue |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(), "sendReply", |
| JmsConstants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_exception__WARNING", |
| new Object[] { e }); |
| } |
| } |
| |
| catch (ServiceShutdownException e) { |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(), |
| "sendReply", JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_exception__WARNING", new Object[] { JmsConstants.threadName(), e }); |
| } |
| } |
| |
| catch (AsynchAEException e) { |
| throw e; |
| } catch (Exception e) { |
| throw new AsynchAEException(e); |
| } |
| |
| } |
| |
| public void sendReply(int aCommand, Endpoint anEndpoint) throws AsynchAEException { |
| anEndpoint.setReplyEndpoint(true); |
| try { |
| if (aborting) { |
| return; |
| } |
| JmsEndpointConnection_impl endpointConnection = getEndpointConnection(anEndpoint); |
| |
| TextMessage tm = endpointConnection.produceTextMessage(""); |
| tm.setIntProperty(AsynchAEMessage.Payload, AsynchAEMessage.None); |
| populateHeaderWithResponseContext(tm, anEndpoint, aCommand); |
| |
| endpointConnection.send(tm, 0, false); |
| addIdleTime(tm); |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINE)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb( |
| Level.FINE, |
| CLASS_NAME.getName(), |
| "sendReply", |
| JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_cpc_reply_sent__FINE", |
| new Object[] { getAnalysisEngineController().getComponentName(), |
| anEndpoint.getEndpoint() }); |
| } |
| } catch (JMSException e) { |
| // Unable to establish connection to the endpoint. Logit and continue |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(), "sendReply", |
| JmsConstants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_exception__WARNING", |
| new Object[] { e }); |
| } |
| } |
| |
| catch (ServiceShutdownException e) { |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(), |
| "sendReply", JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_exception__WARNING", new Object[] { JmsConstants.threadName(), e }); |
| } |
| } |
| |
| catch (AsynchAEException e) { |
| throw e; |
| } catch (Exception e) { |
| throw new AsynchAEException(e); |
| } |
| } |
| |
| /** |
| * Sends JMS Reply Message to a given endpoint. The reply message contains given Throwable (with |
| * full stack) |
| * |
| * @param t |
| * - Throwable to include in the reply message |
| * @param anEndpoint |
| * - an endpoint to receive the reply message |
| * @param aCasReferenceId |
| * - a unique CAS reference id |
| * |
| * @throws AsynchAEException |
| */ |
| public void sendReply(String aCasReferenceId, Endpoint anEndpoint) throws AsynchAEException { |
| anEndpoint.setReplyEndpoint(true); |
| try { |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINE)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, CLASS_NAME.getName(), "sendReply", |
| JmsConstants.JMS_LOG_RESOURCE_BUNDLE, "UIMAJMS_sending_replyto_endpoint__FINE", |
| new Object[] { anEndpoint.getEndpoint(), aCasReferenceId }); |
| } |
| if (anEndpoint.isRemote()) { |
| CacheEntry entry = null; |
| try { |
| entry = getAnalysisEngineController().getInProcessCache().getCacheEntryForCAS( |
| aCasReferenceId); |
| |
| } catch (Exception e) { |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb( |
| Level.FINE, |
| CLASS_NAME.getName(), |
| "sendReply", |
| JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_cas_not_found__INFO", |
| new Object[] { getAnalysisEngineController().getComponentName(), |
| anEndpoint.getEndpoint(), aCasReferenceId }); |
| } |
| return; |
| } |
| if (anEndpoint.getSerializer().equals("xmi")) { |
| // Serializes CAS and releases it back to CAS Pool |
| String serializedCAS = getSerializedCas(true, aCasReferenceId, anEndpoint, false); |
| sendCasToRemoteEndpoint(false, serializedCAS, null, aCasReferenceId, anEndpoint, false, 0); |
| } else { |
| byte[] binaryCas = getBinaryCas(true, entry.getCasReferenceId(), anEndpoint, anEndpoint |
| .isRetryEnabled()); |
| if (binaryCas == null) { |
| return; |
| } |
| sendCasToRemoteEndpoint(false, binaryCas, entry, anEndpoint, false); |
| } |
| } else { |
| // Not supported |
| } |
| } catch (ServiceShutdownException e) { |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(), |
| "sendReply", JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_exception__WARNING", new Object[] { JmsConstants.threadName(), e }); |
| } |
| } |
| |
| catch (AsynchAEException e) { |
| throw e; |
| } catch (Exception e) { |
| throw new AsynchAEException(e); |
| } |
| } |
| |
| /** |
| * Sends JMS Reply Message to a given endpoint. The reply message contains given Throwable (with |
| * full stack) |
| * |
| * @param t |
| * - Throwable to include in the reply message |
| * @param anEndpoint |
| * - an endpoint to receive the reply message |
| * @param aCasReferenceId |
| * - a unique CAS reference id |
| * |
| * @throws AsynchAEException |
| */ |
| public void sendReply(Throwable t, String aCasReferenceId, String aParentCasReferenceId, |
| Endpoint anEndpoint, int aCommand) throws AsynchAEException { |
| anEndpoint.setReplyEndpoint(true); |
| try { |
| Throwable wrapper = null; |
| if (!(t instanceof UimaEEServiceException)) { |
| // Strip off AsyncAEException and replace with UimaEEServiceException |
| if (t instanceof AsynchAEException && t.getCause() != null) { |
| wrapper = new UimaEEServiceException(t.getCause()); |
| } else { |
| wrapper = new UimaEEServiceException(t); |
| } |
| } |
| if (aborting) { |
| return; |
| } |
| anEndpoint.setReplyEndpoint(true); |
| JmsEndpointConnection_impl endpointConnection = getEndpointConnection(anEndpoint); |
| // Create Message that will contain serialized Exception with stack |
| ObjectMessage om = endpointConnection.produceObjectMessage(); |
| // Now try to catch non-serializable exception. The Throwable passed into this method may |
| // not be serializable. Catch the exception, and create a wrapper containing stringified |
| // stack trace. |
| try { |
| // serialize the Throwable |
| if (wrapper == null) { |
| om.setObject(t); |
| } else { |
| om.setObject(wrapper); |
| } |
| } catch( RuntimeException e) { |
| // Check if we failed due to non-serializable object in the Throwable |
| if ( e.getCause() != null && e.getCause() instanceof NotSerializableException ) { |
| // stringify the stack trace |
| StringWriter sw = new StringWriter(); |
| t.printStackTrace(new PrintWriter(sw)); |
| wrapper = new UimaEEServiceException(sw.toString()); |
| // serialize the new wrapper |
| om.setObject(wrapper); |
| } else { |
| throw e; // rethrow |
| } |
| } catch( Exception e) { |
| throw e; // rethrow |
| } |
| // Add common header properties |
| populateHeaderWithResponseContext(om, anEndpoint, aCommand); // AsynchAEMessage.Process); |
| |
| om.setIntProperty(AsynchAEMessage.Payload, AsynchAEMessage.Exception); |
| if (aCasReferenceId != null) { |
| om.setStringProperty(AsynchAEMessage.CasReference, aCasReferenceId); |
| if (aParentCasReferenceId != null) { |
| om.setStringProperty(AsynchAEMessage.InputCasReference, aParentCasReferenceId); |
| } |
| } |
| |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINE)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, CLASS_NAME.getName(), "sendReply", |
| JmsConstants.JMS_LOG_RESOURCE_BUNDLE, "UIMAJMS_sending_exception__FINE", |
| new Object[] { getAnalysisEngineController().getName(), anEndpoint.getEndpoint() }); |
| } |
| // Dispatch Message to destination |
| endpointConnection.send(om, 0, false); |
| addIdleTime(om); |
| } catch (JMSException e) { |
| // Unable to establish connection to the endpoint. Logit and continue |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(), "sendReply", |
| JmsConstants.JMS_LOG_RESOURCE_BUNDLE, "UIMAJMS_unable_to_connect__INFO", |
| new Object[] { getAnalysisEngineController().getName(), anEndpoint.getEndpoint() }); |
| } |
| } catch (ServiceShutdownException e) { |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(), |
| "sendReply", JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_exception__WARNING", new Object[] { JmsConstants.threadName(), e }); |
| } |
| } catch (AsynchAEException e) { |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(), "sendReply", |
| JmsConstants.JMS_LOG_RESOURCE_BUNDLE, "UIMAJMS_unable_to_connect__INFO", |
| new Object[] { getAnalysisEngineController().getName(), anEndpoint.getEndpoint() }); |
| } |
| } catch (Exception e) { |
| throw new AsynchAEException(e); |
| } |
| } |
| |
| /** |
| * |
| * @param aMetadata |
| * @param anEndpoint |
| * @throws AsynchAEException |
| */ |
| public void sendReply(ProcessingResourceMetaData aProcessingResourceMetadata, |
| Endpoint anEndpoint, boolean serialize) throws AsynchAEException { |
| if (aborting) { |
| return; |
| } |
| long msgSize = 0; |
| |
| ByteArrayOutputStream bos = new ByteArrayOutputStream(); |
| try { |
| anEndpoint.setReplyEndpoint(true); |
| // Initialize JMS connection to given endpoint |
| JmsEndpointConnection_impl endpointConnection = getEndpointConnection(anEndpoint); |
| |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINEST)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINEST, CLASS_NAME.getName(), "sendReply", |
| JmsConstants.JMS_LOG_RESOURCE_BUNDLE, "UIMAJMS_produce_txt_msg__FINE", |
| new Object[] {}); |
| } |
| TextMessage tm = endpointConnection.produceTextMessage(""); |
| |
| // Collocated Aggregate components dont send metadata just empty reply |
| // Such aggregate has merged its typesystem already since it shares |
| // CasManager with its parent |
| if (serialize) { |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINEST)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINEST, CLASS_NAME.getName(), |
| "sendReply", JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_serializing_meta__FINE", new Object[] {}); |
| } |
| // Serialize metadata |
| aProcessingResourceMetadata.toXML(bos); |
| tm.setText(bos.toString()); |
| msgSize = bos.toString().length(); |
| } |
| |
| tm.setIntProperty(AsynchAEMessage.Payload, AsynchAEMessage.Metadata); |
| // This service supports Binary Serialization |
| tm.setIntProperty(AsynchAEMessage.Serialization, AsynchAEMessage.BinarySerialization); |
| |
| populateHeaderWithResponseContext(tm, anEndpoint, AsynchAEMessage.GetMeta); |
| if (freeCASTempQueue != null) { |
| // Attach a temp queue to the outgoing message. This a queue where |
| // Free CAS notifications need to be sent from the client |
| tm.setJMSReplyTo(freeCASTempQueue); |
| } |
| |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINEST)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINEST, CLASS_NAME.getName(), "sendReply", |
| JmsConstants.JMS_LOG_RESOURCE_BUNDLE, "UIMAJMS_metadata_reply__endpoint__FINEST", |
| new Object[] { serviceInputEndpoint, anEndpoint.getEndpoint() }); |
| } |
| endpointConnection.send(tm, msgSize, false); |
| } catch (JMSException e) { |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(), |
| "sendReply", JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_exception__WARNING", new Object[] { JmsConstants.threadName(), e }); |
| } |
| // Unable to establish connection to the endpoint. Log it and continue |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(), "sendReply", |
| JmsConstants.JMS_LOG_RESOURCE_BUNDLE, "UIMAJMS_unable_to_connect__INFO", |
| new Object[] { getAnalysisEngineController().getName(), anEndpoint.getEndpoint() }); |
| } |
| } |
| |
| catch (ServiceShutdownException e) { |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(), |
| "sendReply", JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_exception__WARNING", new Object[] { JmsConstants.threadName(), e }); |
| } |
| } catch (Exception e) { |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(), |
| "sendReply", JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_exception__WARNING", new Object[] { JmsConstants.threadName(), e }); |
| } |
| throw new AsynchAEException(e); |
| } finally { |
| try { |
| bos.close(); |
| } catch (Exception e) { |
| } |
| } |
| } |
| |
| private byte[] getBinaryCas(boolean isReply, String aCasReferenceId, Endpoint anEndpoint, |
| boolean cacheSerializedCas) throws Exception { |
| CAS cas = null; |
| try { |
| byte[] serializedCAS = null; |
| // Using Cas reference Id retrieve CAS from the shared Cash |
| cas = getAnalysisEngineController().getInProcessCache().getCasByReference(aCasReferenceId); |
| ServicePerformance casStats = getAnalysisEngineController().getCasStatistics(aCasReferenceId); |
| CacheEntry entry = getAnalysisEngineController().getInProcessCache().getCacheEntryForCAS( |
| aCasReferenceId); |
| long t1 = getAnalysisEngineController().getCpuTime(); |
| // Serialize CAS for remote Delegates |
| String serializer = anEndpoint.getSerializer(); |
| if (cas == null || entry == null) { |
| return null; |
| } |
| if (serializer.equals("binary")) { |
| if (entry.acceptsDeltaCas() && isReply) { |
| if (entry.getMarker() != null && entry.getMarker().isValid()) { |
| serializedCAS = uimaSerializer.serializeCasToBinary(cas, entry.getMarker()); |
| entry.setSentDeltaCas(true); |
| } else { |
| serializedCAS = uimaSerializer.serializeCasToBinary(cas); |
| entry.setSentDeltaCas(false); |
| } |
| } else { |
| serializedCAS = uimaSerializer.serializeCasToBinary(cas); |
| entry.setSentDeltaCas(false); |
| } |
| // create a fresh marker |
| if (entry.getMarker() != null && !entry.getMarker().isValid()) { |
| entry.setMarker(cas.createMarker()); |
| } |
| } else { |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb( |
| Level.INFO, |
| CLASS_NAME.getName(), |
| "getBinaryCas", |
| JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_invalid_serializer__WARNING", |
| new Object[] { getAnalysisEngineController().getName(), serializer, |
| anEndpoint.getEndpoint() }); |
| } |
| throw new UimaEEServiceException("Invalid Serializer:" + serializer + " For Endpoint:" |
| + anEndpoint.getEndpoint()); |
| } |
| long timeToSerializeCas = getAnalysisEngineController().getCpuTime() - t1; |
| |
| getAnalysisEngineController().incrementSerializationTime(timeToSerializeCas); |
| |
| entry.incrementTimeToSerializeCAS(timeToSerializeCas); |
| casStats.incrementCasSerializationTime(timeToSerializeCas); |
| getAnalysisEngineController().getServicePerformance().incrementCasSerializationTime( |
| timeToSerializeCas); |
| return serializedCAS; |
| } catch (Exception e) { |
| throw new AsynchAEException(e); |
| } |
| |
| } |
| |
| private String getSerializedCas(boolean isReply, String aCasReferenceId, Endpoint anEndpoint, |
| boolean cacheSerializedCas) throws Exception { |
| CAS cas = null; |
| try { |
| String serializedCAS = null; |
| // Using Cas reference Id retrieve CAS from the shared Cash |
| cas = getAnalysisEngineController().getInProcessCache().getCasByReference(aCasReferenceId); |
| ServicePerformance casStats = getAnalysisEngineController().getCasStatistics(aCasReferenceId); |
| if (cas == null) { |
| serializedCAS = getAnalysisEngineController().getInProcessCache().getSerializedCAS( |
| aCasReferenceId); |
| } else { |
| CacheEntry entry = getAnalysisEngineController().getInProcessCache().getCacheEntryForCAS( |
| aCasReferenceId); |
| long t1 = getAnalysisEngineController().getCpuTime(); |
| // Serialize CAS for remote Delegates |
| String serializer = anEndpoint.getSerializer(); |
| if (serializer == null || serializer.trim().length() == 0) { |
| serializer = "xmi"; |
| } |
| serializedCAS = serializeCAS(isReply, cas, aCasReferenceId, serializer); |
| long timeToSerializeCas = getAnalysisEngineController().getCpuTime() - t1; |
| getAnalysisEngineController().incrementSerializationTime(timeToSerializeCas); |
| |
| entry.incrementTimeToSerializeCAS(timeToSerializeCas); |
| casStats.incrementCasSerializationTime(timeToSerializeCas); |
| getAnalysisEngineController().getServicePerformance().incrementCasSerializationTime( |
| timeToSerializeCas); |
| if (cacheSerializedCas) { |
| getAnalysisEngineController().getInProcessCache().saveSerializedCAS(aCasReferenceId, |
| serializedCAS); |
| } |
| } |
| return serializedCAS; |
| } catch (Exception e) { |
| throw new AsynchAEException(e); |
| } |
| } |
| |
| private byte[] getBinaryCasAndReleaseIt(boolean isReply, String aCasReferenceId, |
| Endpoint anEndpoint, boolean cacheSerializedCas) throws Exception { |
| try { |
| return getBinaryCas(isReply, aCasReferenceId, anEndpoint, cacheSerializedCas); |
| } catch (Exception e) { |
| throw new AsynchAEException(e); |
| } finally { |
| if (getAnalysisEngineController() instanceof PrimitiveAnalysisEngineController |
| && anEndpoint.isRemote()) { |
| getAnalysisEngineController().dropCAS(aCasReferenceId, true); |
| } |
| } |
| } |
| |
| private String getSerializedCasAndReleaseIt(boolean isReply, String aCasReferenceId, |
| Endpoint anEndpoint, boolean cacheSerializedCas) throws Exception { |
| try { |
| return getSerializedCas(isReply, aCasReferenceId, anEndpoint, cacheSerializedCas); |
| } catch (Exception e) { |
| throw new AsynchAEException(e); |
| } finally { |
| if (getAnalysisEngineController() instanceof PrimitiveAnalysisEngineController |
| && anEndpoint.isRemote()) { |
| getAnalysisEngineController().dropCAS(aCasReferenceId, true); |
| } |
| } |
| } |
| |
| private boolean endpointRetryEnabled(Endpoint[] endpoints) { |
| for (int i = 0; i < endpoints.length; i++) { |
| if (endpoints[i].isRetryEnabled()) { |
| return true; |
| } |
| } |
| return false; |
| } |
| |
| private void populateStats(Message aTextMessage, Endpoint anEndpoint, String aCasReferenceId, |
| int anAdminCommand, boolean isRequest) throws Exception { |
| if (anEndpoint.isFinal()) { |
| aTextMessage.setLongProperty("SENT-TIME", System.nanoTime()); |
| } |
| |
| if (anAdminCommand == AsynchAEMessage.Process) { |
| if (isRequest) { |
| long departureTime = System.nanoTime(); |
| getAnalysisEngineController().saveTime(departureTime, aCasReferenceId, |
| anEndpoint.getEndpoint()); |
| } else { |
| |
| ServicePerformance casStats = getAnalysisEngineController().getCasStatistics( |
| aCasReferenceId); |
| |
| aTextMessage.setLongProperty(AsynchAEMessage.TimeToSerializeCAS, casStats |
| .getRawCasSerializationTime()); |
| aTextMessage.setLongProperty(AsynchAEMessage.TimeToDeserializeCAS, casStats |
| .getRawCasDeserializationTime()); |
| aTextMessage.setLongProperty(AsynchAEMessage.TimeInProcessCAS, casStats |
| .getRawAnalysisTime()); |
| aTextMessage.setLongProperty(AsynchAEMessage.TimeWaitingForCAS, |
| getAnalysisEngineController().getServicePerformance().getTimeWaitingForCAS()); |
| long iT = getAnalysisEngineController().getIdleTimeBetweenProcessCalls( |
| AsynchAEMessage.Process); |
| aTextMessage.setLongProperty(AsynchAEMessage.IdleTime, iT); |
| String lookupKey = getAnalysisEngineController().getName(); |
| long arrivalTime = getAnalysisEngineController().getTime(aCasReferenceId, lookupKey); // serviceInputEndpoint); |
| long timeInService = getAnalysisEngineController().getCpuTime() - arrivalTime; |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINEST)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINEST, CLASS_NAME.getName(), |
| "populateStats", JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_timein_service__FINEST", |
| new Object[] { serviceInputEndpoint, (double) timeInService / (double) 1000000 }); |
| } |
| } |
| } |
| } |
| |
| private long getCommandTimeoutValue(Endpoint anEndpoint, int aCommand) { |
| switch (aCommand) { |
| case AsynchAEMessage.GetMeta: |
| return anEndpoint.getMetadataRequestTimeout(); |
| case AsynchAEMessage.Process: |
| return anEndpoint.getProcessRequestTimeout(); |
| } |
| return 0; // no match for the command |
| } |
| |
| /** |
| * Adds Request specific properties to the JMS Header. |
| * |
| * @param aMessage |
| * @param anEndpoint |
| * @param aCommand |
| * @throws Exception |
| */ |
| private void populateHeaderWithRequestContext(Message aMessage, Endpoint anEndpoint, int aCommand) |
| throws Exception { |
| aMessage.setIntProperty(AsynchAEMessage.MessageType, AsynchAEMessage.Request); |
| aMessage.setIntProperty(AsynchAEMessage.Command, aCommand); |
| // TODO override default based on system property |
| aMessage.setBooleanProperty(AsynchAEMessage.AcceptsDeltaCas, true); |
| long timeout = getCommandTimeoutValue(anEndpoint, aCommand); |
| // If the timeout is defined in the Deployment Descriptor and |
| // the service is configured to use time to live (TTL), add |
| // JMS message expiration time. The TTL is by default always |
| // added to the message. To override this add "-DNoTTL" to the |
| // command line. |
| if (timeout > 0 && addTimeToLive) { |
| Delegate delegate = lookupDelegate(anEndpoint.getDelegateKey()); |
| long ttl = timeout; |
| // How many CASes are in the list of CASes pending reply for this delegate |
| int currentOutstandingCasListSize = delegate.getCasPendingReplyListSize(); |
| if (currentOutstandingCasListSize > 0) { |
| // increase the time-to-live |
| ttl *= currentOutstandingCasListSize; |
| } |
| aMessage.setJMSExpiration(ttl); |
| } |
| if (getAnalysisEngineController() instanceof AggregateAnalysisEngineController) { |
| aMessage.setStringProperty(AsynchAEMessage.MessageFrom, controllerInputEndpoint); |
| if (anEndpoint.isRemote()) { |
| String protocol = serviceProtocolList; |
| if (anEndpoint.getServerURI().trim().toLowerCase().startsWith("http") |
| || (anEndpoint.getReplyToEndpoint() != null && anEndpoint.getReplyToEndpoint() |
| .trim().length() > 0)) { |
| protocol = anEndpoint.getServerURI().trim(); |
| // protocol = extractURLWithProtocol(serviceProtocolList, "http"); |
| |
| // get the replyto endpoint name |
| String replyTo = anEndpoint.getReplyToEndpoint(); |
| if (replyTo == null && anEndpoint.getDestination() == null) { |
| throw new AsynchAEException( |
| "replyTo endpoint name not specified for HTTP-based endpoint:" |
| + anEndpoint.getEndpoint()); |
| } |
| if (replyTo == null) { |
| replyTo = ""; |
| } |
| aMessage.setStringProperty(AsynchAEMessage.MessageFrom, replyTo); |
| |
| } |
| Object destination; |
| |
| if ((destination = anEndpoint.getDestination()) != null) { |
| aMessage.setJMSReplyTo((Destination) destination); |
| aMessage.setStringProperty(UIMAMessage.ServerURI, anEndpoint.getServerURI()); |
| } else { |
| aMessage.setStringProperty(UIMAMessage.ServerURI, protocol); |
| } |
| |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINE)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb( |
| Level.FINE, |
| CLASS_NAME.getName(), |
| "populateHeaderWithRequestContext", |
| JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_sending_new_msg_to_remote_FINE", |
| new Object[] { getAnalysisEngineController().getComponentName(), |
| anEndpoint.getServerURI(), anEndpoint.getEndpoint() }); |
| } |
| } else // collocated |
| { |
| aMessage.setStringProperty(UIMAMessage.ServerURI, anEndpoint.getServerURI()); |
| } |
| } |
| } |
| |
| /** |
| * Adds Response specific properties to the JMS Header |
| * |
| * @param aMessage |
| * @param anEndpoint |
| * @param aCommand |
| * @throws Exception |
| */ |
| private void populateHeaderWithResponseContext(Message aMessage, Endpoint anEndpoint, int aCommand) |
| throws Exception { |
| aMessage.setIntProperty(AsynchAEMessage.MessageType, AsynchAEMessage.Response); |
| aMessage.setIntProperty(AsynchAEMessage.Command, aCommand); |
| aMessage.setStringProperty(AsynchAEMessage.MessageFrom, serviceInputEndpoint); |
| |
| if (anEndpoint.isRemote()) { |
| aMessage.setStringProperty(UIMAMessage.ServerURI, getServerURI()); |
| if (hostIP != null) { |
| aMessage.setStringProperty(AsynchAEMessage.ServerIP, hostIP); |
| } |
| if (anEndpoint.getEndpointServer() != null) { |
| aMessage.setStringProperty(AsynchAEMessage.EndpointServer, anEndpoint.getEndpointServer()); |
| } |
| } else { |
| aMessage.setStringProperty(UIMAMessage.ServerURI, anEndpoint.getServerURI()); |
| } |
| } |
| |
| public AnalysisEngineController getAnalysisEngineController() { |
| return analysisEngineController; |
| } |
| |
| public void setController(AnalysisEngineController analysisEngineController) { |
| this.analysisEngineController = analysisEngineController; |
| controllerLatch.countDown(); |
| } |
| |
| public String getControllerInputEndpoint() { |
| return controllerInputEndpoint; |
| } |
| |
| public void setControllerInputEndpoint(String controllerInputEndpoint) { |
| this.controllerInputEndpoint = controllerInputEndpoint; |
| } |
| |
| private void dispatch(Message aMessage, Endpoint anEndpoint, CacheEntry entry, boolean isRequest, |
| JmsEndpointConnection_impl endpointConnection, long msgSize) throws Exception { |
| // Add stats |
| populateStats(aMessage, anEndpoint, entry.getCasReferenceId(), AsynchAEMessage.Process, |
| isRequest); |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINE)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb( |
| Level.FINE, |
| CLASS_NAME.getName(), |
| "dispatch", |
| JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_sending_new_msg_to_remote_FINE", |
| new Object[] { getAnalysisEngineController().getName(), |
| endpointConnection.getServerUri(), endpointConnection.getEndpoint() }); |
| } |
| // By default start a timer associated with a connection to the endpoint. Once a connection is |
| // established with an |
| // endpoint it is cached and reused for subsequent messaging. If the connection is not used |
| // within a given interval |
| // the timer silently expires and closes the connection. This mechanism is similar to what Web |
| // Server does when |
| // managing sessions. In case when we want the remote delegate to respond to a temporary queue, |
| // which is implied |
| // by anEndpoint.getDestination != null, we dont start the timer. |
| boolean startConnectionTimer = isRequest ? false : true; // connection time is for replies |
| // ---------------------------------------------------- |
| // Send Request Messsage to the Endpoint |
| // ---------------------------------------------------- |
| // Add the CAS to the delegate's list of CASes pending reply. Do the add before |
| // the send to eliminate a race condition where the reply is received (on different |
| // thread) *before* the CAS is added to the list. |
| if (isRequest) { |
| anEndpoint.setWaitingForResponse(true); |
| // Add CAS to the list of CASes pending reply |
| addCasToOutstandingList(entry, isRequest, anEndpoint.getDelegateKey()); |
| } else { |
| addIdleTime(aMessage); |
| } |
| // If the send fails it returns false. |
| if (endpointConnection.send(aMessage, msgSize, startConnectionTimer) == false) { |
| // Failure on sending a request requires cleanup that includes stopping a listener |
| // on the delegate that we were unable to send a message to. The delegate state is |
| // set to FAILED. If there are retries or more CASes to send to this delegate the |
| // connection will be retried. |
| if (isRequest) { |
| // Spin recovery thread to handle send error. After the recovery thread |
| // is started the current (process) thread goes back to a thread pool in |
| // ThreadPoolExecutor. The recovery thread can than stop the listener and the |
| // ThreadPoolExecutor since all threads are back in the pool. Any retries will |
| // be done in the recovery thread. |
| RecoveryThread recoveryThread = new RecoveryThread(this, anEndpoint, entry, isRequest, |
| getAnalysisEngineController()); |
| Thread t = new Thread(Thread.currentThread().getThreadGroup().getParent(), recoveryThread); |
| t.start(); |
| } else { |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb( |
| Level.INFO, |
| CLASS_NAME.getName(), |
| "dispatch", |
| JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_send_reply_failed__INFO", |
| new Object[] { getAnalysisEngineController().getComponentName(), |
| endpointConnection.getServerUri(), endpointConnection.getEndpoint() }); |
| } |
| } |
| } |
| } |
| |
| private void sendCasToRemoteEndpoint(boolean isRequest, String aSerializedCAS, |
| String anInputCasReferenceId, String aCasReferenceId, Endpoint anEndpoint, |
| boolean startTimer, long sequence) throws AsynchAEException, ServiceShutdownException { |
| long msgSize = 0; |
| try { |
| if (aborting) { |
| return; |
| } |
| CacheEntry entry = this.getCacheEntry(aCasReferenceId); |
| if (entry == null) { |
| throw new AsynchAEException("Controller:" |
| + getAnalysisEngineController().getComponentName() |
| + " Unable to Send Message To Remote Endpoint: " + anEndpoint.getEndpoint() |
| + " CAS:" + aCasReferenceId + " Not In The Cache"); |
| } |
| |
| // Get the connection object for a given endpoint |
| JmsEndpointConnection_impl endpointConnection = getEndpointConnection(anEndpoint); |
| if (endpointConnection == null) { |
| throw new AsynchAEException("Controller:" |
| + getAnalysisEngineController().getComponentName() |
| + " Unable to Send Message To Remote Endpoint: " + anEndpoint.getEndpoint() |
| + " Connection is Invalid. InputCasReferenceId:" + anInputCasReferenceId |
| + " CasReferenceId:" + aCasReferenceId + " Sequece:" + sequence); |
| } |
| if (!endpointConnection.isOpen()) { |
| if (!isRequest) { |
| return; |
| } |
| } |
| TextMessage tm = null; |
| try { |
| // Create empty JMS Text Message |
| tm = endpointConnection.produceTextMessage(""); |
| } catch (AsynchAEException ex) { |
| System.out.println("UIMA AS Service:" + getAnalysisEngineController().getComponentName() |
| + " Unable to Send Reply Message To Remote Endpoint: " |
| + anEndpoint.getDestination() + ". Broker:" + anEndpoint.getServerURI() |
| + " is Unavailable. InputCasReferenceId:" + anInputCasReferenceId |
| + " CasReferenceId:" + aCasReferenceId + " Sequece:" + sequence); |
| UIMAFramework.getLogger(CLASS_NAME).logrb( |
| Level.INFO, |
| CLASS_NAME.getName(), |
| "sendCasToRemoteDelegate", |
| JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_unable_to_connect__INFO", |
| new Object[] { getAnalysisEngineController().getComponentName(), |
| anEndpoint.getEndpoint() }); |
| return; |
| } |
| // Save Serialized CAS in case we need to re-send it for analysis |
| if (anEndpoint.isRetryEnabled() |
| && getAnalysisEngineController().getInProcessCache() |
| .getSerializedCAS(aCasReferenceId) == null) { |
| getAnalysisEngineController().getInProcessCache().saveSerializedCAS(aCasReferenceId, |
| aSerializedCAS); |
| } |
| if (aSerializedCAS != null) { |
| msgSize = aSerializedCAS.length(); |
| } |
| tm.setText(aSerializedCAS); |
| tm.setIntProperty(AsynchAEMessage.Payload, AsynchAEMessage.XMIPayload); |
| // Add Cas Reference Id to the outgoing JMS Header |
| tm.setStringProperty(AsynchAEMessage.CasReference, aCasReferenceId); |
| |
| // Add common properties to the JMS Header |
| if (isRequest == true) { |
| populateHeaderWithRequestContext(tm, anEndpoint, AsynchAEMessage.Process); |
| } else { |
| populateHeaderWithResponseContext(tm, anEndpoint, AsynchAEMessage.Process); |
| tm.setBooleanProperty(AsynchAEMessage.SentDeltaCas, entry.sentDeltaCas()); |
| } |
| // The following is true when the analytic is a CAS Multiplier |
| if (sequence > 0 && !isRequest) { |
| // Override MessageType set in the populateHeaderWithContext above. |
| // Make the reply message look like a request. This message will contain a new CAS |
| // produced by the CAS Multiplier. The client will treat this CAS |
| // differently from the input CAS. |
| tm.setIntProperty(AsynchAEMessage.MessageType, AsynchAEMessage.Request); |
| tm.setStringProperty(AsynchAEMessage.InputCasReference, anInputCasReferenceId); |
| // Add a sequence number assigned to this CAS by the controller |
| tm.setLongProperty(AsynchAEMessage.CasSequence, sequence); |
| isRequest = true; |
| // Add the name of the FreeCas Queue |
| if (freeCASTempQueue != null) { |
| // Attach a temp queue to the outgoing message. This a queue where |
| // Free CAS notifications need to be sent from the client |
| tm.setJMSReplyTo(freeCASTempQueue); |
| } |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINE)) { |
| if (entry != null) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb( |
| Level.FINE, |
| CLASS_NAME.getName(), |
| "sendCasToRemoteEndpoint", |
| JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_send_cas_to_collocated_service_detail__FINE", |
| new Object[] { getAnalysisEngineController().getComponentName(), "Remote", |
| anEndpoint.getEndpoint(), aCasReferenceId, anInputCasReferenceId, |
| entry.getInputCasReferenceId() }); |
| } |
| } |
| } |
| dispatch(tm, anEndpoint, entry, isRequest, endpointConnection, msgSize); |
| } catch (JMSException e) { |
| // Unable to establish connection to the endpoint. Log it and continue |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(), |
| "sendCasToRemoteDelegate", JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_unable_to_connect__INFO", |
| new Object[] { getAnalysisEngineController().getName(), anEndpoint.getEndpoint() }); |
| } |
| } |
| |
| catch (ServiceShutdownException e) { |
| throw e; |
| } catch (AsynchAEException e) { |
| throw e; |
| } catch (Exception e) { |
| throw new AsynchAEException(e); |
| } |
| |
| } |
| |
| private void sendCasToRemoteEndpoint(boolean isRequest, byte[] aSerializedCAS, |
| String anInputCasReferenceId, String aCasReferenceId, Endpoint anEndpoint, |
| boolean startTimer, long sequence) throws AsynchAEException, ServiceShutdownException { |
| long msgSize = 0; |
| try { |
| if (aborting) { |
| return; |
| } |
| if (aSerializedCAS != null) { |
| msgSize = aSerializedCAS.length; |
| } |
| CacheEntry entry = this.getCacheEntry(aCasReferenceId); |
| if (entry == null) { |
| throw new AsynchAEException("Controller:" |
| + getAnalysisEngineController().getComponentName() |
| + " Unable to Send Message To Remote Endpoint: " + anEndpoint.getEndpoint() |
| + " CAS:" + aCasReferenceId + " Not In The Cache"); |
| } |
| // Get the connection object for a given endpoint |
| JmsEndpointConnection_impl endpointConnection = getEndpointConnection(anEndpoint); |
| if (endpointConnection == null) { |
| throw new AsynchAEException("Controller:" |
| + getAnalysisEngineController().getComponentName() |
| + " Unable to Send Message To Remote Endpoint: " + anEndpoint.getEndpoint() |
| + " Connection is Invalid. InputCasReferenceId:" + anInputCasReferenceId |
| + " CasReferenceId:" + aCasReferenceId + " Sequece:" + sequence); |
| } |
| if (!endpointConnection.isOpen()) { |
| if (!isRequest) { |
| return; |
| } |
| } |
| |
| BytesMessage tm = null; |
| try { |
| // Create empty JMS Text Message |
| tm = endpointConnection.produceByteMessage(); |
| } catch (AsynchAEException ex) { |
| System.out.println("UIMA AS Service:" + getAnalysisEngineController().getComponentName() |
| + " Unable to Send Reply Message To Remote Endpoint: " |
| + anEndpoint.getDestination() + ". Broker:" + anEndpoint.getServerURI() |
| + " is Unavailable. InputCasReferenceId:" + anInputCasReferenceId |
| + " CasReferenceId:" + aCasReferenceId + " Sequece:" + sequence); |
| UIMAFramework.getLogger(CLASS_NAME).logrb( |
| Level.INFO, |
| CLASS_NAME.getName(), |
| "sendCasToRemoteDelegate", |
| JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_unable_to_connect__INFO", |
| new Object[] { getAnalysisEngineController().getComponentName(), |
| anEndpoint.getEndpoint() }); |
| return; |
| } |
| |
| tm.writeBytes(aSerializedCAS); |
| tm.setIntProperty(AsynchAEMessage.Payload, AsynchAEMessage.BinaryPayload); |
| // Add Cas Reference Id to the outgoing JMS Header |
| tm.setStringProperty(AsynchAEMessage.CasReference, aCasReferenceId); |
| // Add common properties to the JMS Header |
| if (isRequest == true) { |
| populateHeaderWithRequestContext(tm, anEndpoint, AsynchAEMessage.Process); |
| } else { |
| populateHeaderWithResponseContext(tm, anEndpoint, AsynchAEMessage.Process); |
| tm.setBooleanProperty(AsynchAEMessage.SentDeltaCas, entry.sentDeltaCas()); |
| } |
| // The following is true when the analytic is a CAS Multiplier |
| if (sequence > 0 && !isRequest) { |
| // Override MessageType set in the populateHeaderWithContext above. |
| // Make the reply message look like a request. This message will contain a new CAS |
| // produced by the CAS Multiplier. The client will treat this CAS |
| // differently from the input CAS. |
| tm.setIntProperty(AsynchAEMessage.MessageType, AsynchAEMessage.Request); |
| tm.setStringProperty(AsynchAEMessage.InputCasReference, anInputCasReferenceId); |
| // Add a sequence number assigned to this CAS by the controller |
| tm.setLongProperty(AsynchAEMessage.CasSequence, sequence); |
| isRequest = true; |
| if (freeCASTempQueue != null) { |
| // Attach a temp queue to the outgoing message. This a queue where |
| // Free CAS notifications need to be sent from the client |
| tm.setJMSReplyTo(freeCASTempQueue); |
| } |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINE)) { |
| if (entry != null) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb( |
| Level.FINE, |
| CLASS_NAME.getName(), |
| "sendCasToRemoteEndpoint", |
| JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_send_cas_to_collocated_service_detail__FINE", |
| new Object[] { getAnalysisEngineController().getComponentName(), "Remote", |
| anEndpoint.getEndpoint(), aCasReferenceId, anInputCasReferenceId, |
| entry.getInputCasReferenceId() }); |
| } |
| } |
| } |
| dispatch(tm, anEndpoint, entry, isRequest, endpointConnection, msgSize); |
| } catch (JMSException e) { |
| // Unable to establish connection to the endpoint. Logit and continue |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(), |
| "sendCasToRemoteDelegate", JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_unable_to_connect__INFO", |
| new Object[] { getAnalysisEngineController().getName(), anEndpoint.getEndpoint() }); |
| } |
| } |
| |
| catch (ServiceShutdownException e) { |
| throw e; |
| } catch (AsynchAEException e) { |
| throw e; |
| } catch (Exception e) { |
| throw new AsynchAEException(e); |
| } |
| |
| } |
| |
| private void sendCasToRemoteEndpoint(boolean isRequest, String aSerializedCAS, CacheEntry entry, |
| Endpoint anEndpoint, boolean startTimer) throws AsynchAEException, |
| ServiceShutdownException { |
| CasStateEntry casStateEntry = null; |
| long msgSize = 0; |
| try { |
| if (aborting) { |
| return; |
| } |
| casStateEntry = getAnalysisEngineController().getLocalCache().lookupEntry( |
| entry.getCasReferenceId()); |
| // Get the connection object for a given endpoint |
| JmsEndpointConnection_impl endpointConnection = getEndpointConnection(anEndpoint); |
| if (!endpointConnection.isOpen()) { |
| if (!isRequest) { |
| return; |
| } |
| } |
| // Create empty JMS Text Message |
| TextMessage tm = null; |
| try { |
| // Create empty JMS Text Message |
| tm = endpointConnection.produceTextMessage(""); |
| } catch (AsynchAEException ex) { |
| System.out.println("UIMA AS Service:" + getAnalysisEngineController().getComponentName() |
| + " Unable to Send Reply Message To Remote Endpoint: " |
| + anEndpoint.getDestination() + ". Broker:" + anEndpoint.getServerURI() |
| + " is Unavailable. CasReferenceId:" + casStateEntry.getCasReferenceId()); |
| UIMAFramework.getLogger(CLASS_NAME).logrb( |
| Level.INFO, |
| CLASS_NAME.getName(), |
| "sendCasToRemoteDelegate", |
| JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_unable_to_connect__INFO", |
| new Object[] { getAnalysisEngineController().getComponentName(), |
| anEndpoint.getEndpoint() }); |
| return; |
| } |
| |
| // Save Serialized CAS in case we need to re-send it for analysis |
| if (anEndpoint.isRetryEnabled() |
| && getAnalysisEngineController().getInProcessCache().getSerializedCAS( |
| entry.getCasReferenceId()) == null) { |
| getAnalysisEngineController().getInProcessCache().saveSerializedCAS( |
| entry.getCasReferenceId(), aSerializedCAS); |
| } |
| if (aSerializedCAS != null) { |
| msgSize = aSerializedCAS.length(); |
| } |
| |
| tm.setText(aSerializedCAS); |
| tm.setIntProperty(AsynchAEMessage.Payload, AsynchAEMessage.XMIPayload); |
| // Add Cas Reference Id to the outgoing JMS Header |
| tm.setStringProperty(AsynchAEMessage.CasReference, entry.getCasReferenceId()); |
| // Add common properties to the JMS Header |
| if (isRequest == true) { |
| populateHeaderWithRequestContext(tm, anEndpoint, AsynchAEMessage.Process); |
| } else { |
| populateHeaderWithResponseContext(tm, anEndpoint, AsynchAEMessage.Process); |
| tm.setBooleanProperty(AsynchAEMessage.SentDeltaCas, entry.sentDeltaCas()); |
| } |
| // The following is true when the analytic is a CAS Multiplier |
| if (casStateEntry.isSubordinate() && !isRequest) { |
| // Override MessageType set in the populateHeaderWithContext above. |
| // Make the reply message look like a request. This message will contain a new CAS |
| // produced by the CAS Multiplier. The client will treat this CAS |
| // differently from the input CAS. |
| tm.setIntProperty(AsynchAEMessage.MessageType, AsynchAEMessage.Request); |
| |
| isRequest = true; |
| // Save the id of the parent CAS |
| tm.setStringProperty(AsynchAEMessage.InputCasReference, getTopParentCasReferenceId(entry |
| .getCasReferenceId())); |
| // Add a sequence number assigned to this CAS by the controller |
| tm.setLongProperty(AsynchAEMessage.CasSequence, entry.getCasSequence()); |
| // If this is a Cas Multiplier, add a reference to a special queue where |
| // the client sends Free Cas Notifications |
| if (freeCASTempQueue != null) { |
| // Attach a temp queue to the outgoing message. This is a queue where |
| // Free CAS notifications need to be sent from the client |
| tm.setJMSReplyTo(freeCASTempQueue); |
| } |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINE)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb( |
| Level.FINE, |
| CLASS_NAME.getName(), |
| "sendCasToRemoteEndpoint", |
| JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_send_cas_to_collocated_service_detail__FINE", |
| new Object[] { getAnalysisEngineController().getComponentName(), "Remote", |
| anEndpoint.getEndpoint(), entry.getCasReferenceId(), |
| entry.getInputCasReferenceId(), entry.getInputCasReferenceId() }); |
| } |
| } |
| |
| dispatch(tm, anEndpoint, entry, isRequest, endpointConnection, msgSize); |
| |
| } catch (JMSException e) { |
| // Unable to establish connection to the endpoint. Logit and continue |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(), |
| "sendCasToRemoteDelegate", JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_unable_to_connect__INFO", |
| new Object[] { getAnalysisEngineController().getName(), anEndpoint.getEndpoint() }); |
| } |
| } |
| |
| catch (ServiceShutdownException e) { |
| throw e; |
| } catch (AsynchAEException e) { |
| throw e; |
| } catch (Exception e) { |
| throw new AsynchAEException(e); |
| } |
| |
| } |
| |
| private void sendCasToRemoteEndpoint(boolean isRequest, byte[] aSerializedCAS, CacheEntry entry, |
| Endpoint anEndpoint, boolean startTimer) throws AsynchAEException, |
| ServiceShutdownException { |
| CasStateEntry casStateEntry = null; |
| long msgSize = 0; |
| try { |
| if (aborting) { |
| return; |
| } |
| casStateEntry = getAnalysisEngineController().getLocalCache().lookupEntry( |
| entry.getCasReferenceId()); |
| // Get the connection object for a given endpoint |
| JmsEndpointConnection_impl endpointConnection = getEndpointConnection(anEndpoint); |
| |
| if (aSerializedCAS != null) { |
| msgSize = aSerializedCAS.length; |
| } |
| if (!endpointConnection.isOpen()) { |
| if (!isRequest) { |
| return; |
| } |
| } |
| // Create empty JMS Text Message |
| BytesMessage tm = null; |
| try { |
| // Create empty JMS Text Message |
| tm = endpointConnection.produceByteMessage(); |
| } catch (AsynchAEException ex) { |
| System.out.println("UIMA AS Service:" + getAnalysisEngineController().getComponentName() |
| + " Unable to Send Reply Message To Remote Endpoint: " |
| + anEndpoint.getDestination() + ". Broker:" + anEndpoint.getServerURI() |
| + " is Unavailable. CasReferenceId:" + casStateEntry.getCasReferenceId()); |
| return; |
| } |
| tm.writeBytes(aSerializedCAS); |
| tm.setIntProperty(AsynchAEMessage.Payload, AsynchAEMessage.BinaryPayload); |
| // Add Cas Reference Id to the outgoing JMS Header |
| tm.setStringProperty(AsynchAEMessage.CasReference, entry.getCasReferenceId()); |
| // Add common properties to the JMS Header |
| if (isRequest == true) { |
| populateHeaderWithRequestContext(tm, anEndpoint, AsynchAEMessage.Process); |
| } else { |
| populateHeaderWithResponseContext(tm, anEndpoint, AsynchAEMessage.Process); |
| } |
| if (casStateEntry == null) { |
| return; |
| } |
| // The following is true when the analytic is a CAS Multiplier |
| if (casStateEntry.isSubordinate() && !isRequest) { |
| // Override MessageType set in the populateHeaderWithContext above. |
| // Make the reply message look like a request. This message will contain a new CAS |
| // produced by the CAS Multiplier. The client will treat this CAS |
| // differently from the input CAS. |
| tm.setIntProperty(AsynchAEMessage.MessageType, AsynchAEMessage.Request); |
| |
| isRequest = true; |
| // Save the id of the parent CAS |
| tm.setStringProperty(AsynchAEMessage.InputCasReference, getTopParentCasReferenceId(entry |
| .getCasReferenceId())); |
| // Add a sequence number assigned to this CAS by the controller |
| tm.setLongProperty(AsynchAEMessage.CasSequence, entry.getCasSequence()); |
| // If this is a Cas Multiplier, add a reference to a special queue where |
| // the client sends Free Cas Notifications |
| if (freeCASTempQueue != null) { |
| // Attach a temp queue to the outgoing message. This is a queue where |
| // Free CAS notifications need to be sent from the client |
| tm.setJMSReplyTo(freeCASTempQueue); |
| } |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINE)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb( |
| Level.FINE, |
| CLASS_NAME.getName(), |
| "sendCasToRemoteEndpoint", |
| JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_send_cas_to_collocated_service_detail__FINE", |
| new Object[] { getAnalysisEngineController().getComponentName(), "Remote", |
| anEndpoint.getEndpoint(), entry.getCasReferenceId(), |
| entry.getInputCasReferenceId(), entry.getInputCasReferenceId() }); |
| } |
| } |
| dispatch(tm, anEndpoint, entry, isRequest, endpointConnection, msgSize); |
| |
| } catch (JMSException e) { |
| // Unable to establish connection to the endpoint. Logit and continue |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(), |
| "sendCasToRemoteDelegate", JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_unable_to_connect__INFO", |
| new Object[] { getAnalysisEngineController().getName(), anEndpoint.getEndpoint() }); |
| } |
| } catch (ServiceShutdownException e) { |
| throw e; |
| } catch (AsynchAEException e) { |
| throw e; |
| } catch (Exception e) { |
| throw new AsynchAEException(e); |
| } |
| |
| } |
| |
| private Delegate lookupDelegate(String aDelegateKey) { |
| if (getAnalysisEngineController() instanceof AggregateAnalysisEngineController) { |
| Delegate delegate = ((AggregateAnalysisEngineController) getAnalysisEngineController()) |
| .lookupDelegate(aDelegateKey); |
| return delegate; |
| } |
| return null; |
| } |
| |
| private void addCasToOutstandingList(CacheEntry entry, boolean isRequest, String aDelegateKey) { |
| Delegate delegate = null; |
| if (isRequest && (delegate = lookupDelegate(aDelegateKey)) != null) { |
| delegate.addCasToOutstandingList(entry.getCasReferenceId()); |
| } |
| } |
| |
| private void removeCasFromOutstandingList(CacheEntry entry, boolean isRequest, String aDelegateKey) { |
| Delegate delegate = null; |
| if (isRequest && (delegate = lookupDelegate(aDelegateKey)) != null) { |
| delegate.removeCasFromOutstandingList(entry.getCasReferenceId()); |
| } |
| } |
| |
| private String getTopParentCasReferenceId(String casReferenceId) throws Exception { |
| if (!getAnalysisEngineController().getLocalCache().containsKey(casReferenceId)) { |
| return null; |
| } |
| CasStateEntry casStateEntry = getAnalysisEngineController().getLocalCache().lookupEntry( |
| casReferenceId); |
| |
| if (casStateEntry.isSubordinate()) { |
| // Recurse until the top CAS reference Id is found |
| return getTopParentCasReferenceId(casStateEntry.getInputCasReferenceId()); |
| } |
| // Return the top ancestor CAS id |
| return casStateEntry.getCasReferenceId(); |
| } |
| |
| private void addIdleTime(Message aMessage) { |
| long t = System.nanoTime(); |
| getAnalysisEngineController().saveReplyTime(t, ""); |
| } |
| |
| private CacheEntry getCacheEntry(String aCasReferenceId) throws Exception { |
| CacheEntry cacheEntry = null; |
| if (getAnalysisEngineController().getInProcessCache().entryExists(aCasReferenceId)) { |
| cacheEntry = getAnalysisEngineController().getInProcessCache().getCacheEntryForCAS( |
| aCasReferenceId); |
| } |
| return cacheEntry; |
| } |
| |
| public void stop() { |
| stop(Channel.CloseAllChannels); |
| } |
| |
| public void stop(int channelsToClose) { |
| aborting = true; |
| try { |
| // Fetch iterator over all Broker Connections. This service may be connected |
| // to many brokers. Each broker connection may handle multiple sessions to |
| // different reply queues |
| Iterator it = connectionMap.keySet().iterator(); |
| JmsEndpointConnection_impl endpointConnection = null; |
| // iterate over connections |
| while (it.hasNext()) { |
| // The key is the broker URL |
| String key = (String) it.next(); |
| // Fetch a connection object for a given URL |
| Object value = connectionMap.get(key); |
| |
| if (value instanceof BrokerConnectionEntry) { |
| BrokerConnectionEntry brokerConnectionEntry = (BrokerConnectionEntry) value; |
| // A connection object may have many endpoint objects. There is a separate |
| // endpoint object per reply queue. |
| Iterator replyEndpointIterator = brokerConnectionEntry.endpointMap.keySet().iterator(); |
| // Iterate over endpoints, each representing a reply queue |
| while (replyEndpointIterator.hasNext()) { |
| // Get endpoint object for a reply queue. The abort() call below |
| // just closes a session and a producer. The JMS Connection is closed |
| // outside of this while-loop when we clean up all the sessions. |
| endpointConnection = brokerConnectionEntry.endpointMap |
| .get(replyEndpointIterator.next()); |
| // Close the session and the producer |
| endpointConnection.abort(); |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb( |
| Level.INFO, |
| CLASS_NAME.getName(), |
| "stop", |
| JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_forced_endpoint_close__INFO", |
| new Object[] { getAnalysisEngineController().getName(), |
| endpointConnection.getEndpoint(), endpointConnection.getServerUri() }); |
| } |
| } |
| // Cancel any pending timers and finally close the JMS Connection to the |
| // broker |
| if (brokerConnectionEntry != null) { |
| if (brokerConnectionEntry.getConnectionTimer() != null) { |
| brokerConnectionEntry.getConnectionTimer().cancelTimer(); |
| } |
| if (brokerConnectionEntry.getConnection() != null) { |
| try { |
| brokerConnectionEntry.getConnection().close(); |
| System.out.println("JMS Connection to Broker: " + key + " Closed"); |
| } catch (Exception ex) { /* ignore, we are stopping */ |
| } |
| } |
| } |
| } |
| } |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(), "stop", |
| JmsConstants.JMS_LOG_RESOURCE_BUNDLE, "UIMAJMS_output_channel_aborted__INFO", |
| new Object[] { getAnalysisEngineController().getName() }); |
| } |
| } catch (Exception e) { |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(), |
| "stop", JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_exception__WARNING", new Object[] { JmsConstants.threadName(), e }); |
| } |
| } |
| } |
| |
| private String extractURLWithProtocol(String aProtocolList, String aProtocol) { |
| StringTokenizer tokenizer = new StringTokenizer(aProtocolList, ","); |
| while (tokenizer.hasMoreTokens()) { |
| String token = tokenizer.nextToken().trim(); |
| if (token.toLowerCase().startsWith(aProtocol.toLowerCase())) { |
| return token; |
| } |
| } |
| return null; |
| } |
| |
| public void cancelTimers() { |
| if (connectionMap.size() > 0) { |
| Iterator<String> it = connectionMap.keySet().iterator(); |
| while (it.hasNext()) { |
| String key = it.next(); |
| BrokerConnectionEntry ce = (BrokerConnectionEntry) connectionMap.get(key); |
| if (ce != null && ce.getConnectionTimer() != null) { |
| ce.getConnectionTimer().cancelTimer(); |
| } |
| } |
| } |
| } |
| |
| public static class BrokerConnectionEntry { |
| private String brokerURL; |
| |
| private Connection connection; |
| |
| private ConnectionTimer connectionTimer; |
| |
| Map<Object, JmsEndpointConnection_impl> endpointMap = new ConcurrentHashMap<Object, JmsEndpointConnection_impl>(); |
| |
| public String getBrokerURL() { |
| return brokerURL; |
| } |
| |
| public void setConnectionTimer(ConnectionTimer aConnectionTimer) { |
| connectionTimer = aConnectionTimer; |
| } |
| |
| public ConnectionTimer getConnectionTimer() { |
| return connectionTimer; |
| } |
| |
| public void setBrokerURL(String brokerURL) { |
| this.brokerURL = brokerURL; |
| } |
| |
| public Connection getConnection() { |
| return connection; |
| } |
| |
| public void setConnection(Connection connection) { |
| this.connection = connection; |
| } |
| |
| public void addEndpointConnection(Object key, JmsEndpointConnection_impl endpointConnection) { |
| endpointMap.put(key, endpointConnection); |
| } |
| |
| public JmsEndpointConnection_impl getEndpointConnection(Object key) { |
| return endpointMap.get(key); |
| } |
| |
| public boolean endpointExists(Object key) { |
| return endpointMap.containsKey(key); |
| } |
| |
| public void removeEndpoint(Object key) { |
| endpointMap.remove(key); |
| } |
| } |
| |
| protected class ConnectionTimer { |
| private final Class CLASS_NAME = ConnectionTimer.class; |
| |
| private Timer timer; |
| |
| private long inactivityTimeout; |
| |
| private AnalysisEngineController controller; |
| |
| private BrokerConnectionEntry brokerDestinations; |
| |
| private long connectionCreationTimestamp; |
| |
| private String componentName = ""; |
| |
| public ConnectionTimer(BrokerConnectionEntry aBrokerDestinations) { |
| brokerDestinations = aBrokerDestinations; |
| } |
| |
| public void setInactivityTimeout(long anInactivityTimeout) { |
| inactivityTimeout = anInactivityTimeout; |
| } |
| |
| public void setAnalysisEngineController(AnalysisEngineController aController) { |
| controller = aController; |
| if (controller != null) { |
| componentName = controller.getComponentName(); |
| } |
| } |
| |
| public void setConnectionCreationTimestamp(long aConnectionCreationTimestamp) { |
| connectionCreationTimestamp = aConnectionCreationTimestamp; |
| } |
| |
| public void startTimer(long aConnectionCreationTimestamp, final Endpoint endpoint) { |
| startTimer(aConnectionCreationTimestamp, endpoint, inactivityTimeout, componentName); |
| } |
| |
| public synchronized void startTimer(long aConnectionCreationTimestamp, final Endpoint endpoint, |
| long currentInactivityTimeout, String aComponentName) { |
| final long cachedConnectionCreationTimestamp = aConnectionCreationTimestamp; |
| Date timeToRun = new Date(System.currentTimeMillis() + currentInactivityTimeout); |
| if (timer != null) { |
| timer.cancel(); |
| } |
| if (controller != null) { |
| timer = new Timer("Controller:" + aComponentName + ":Reply TimerThread-:" + endpoint + ":" |
| + System.nanoTime()); |
| } else { |
| timer = new Timer("Reply TimerThread-:" + endpoint + ":" + System.nanoTime()); |
| } |
| timer.schedule(new TimerTask() { |
| public void run() { |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb( |
| Level.INFO, |
| CLASS_NAME.getName(), |
| "startTimer", |
| JmsConstants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAJMS_inactivity_timer_expired__INFO", |
| new Object[] { Thread.currentThread().getName(), componentName, |
| inactivityTimeout, endpoint }); |
| } |
| if (connectionCreationTimestamp <= cachedConnectionCreationTimestamp) { |
| try { |
| if (brokerDestinations.getConnection() != null |
| && !((ActiveMQConnection) brokerDestinations.getConnection()).isClosed()) { |
| try { |
| brokerDestinations.getConnection().stop(); |
| brokerDestinations.getConnection().close(); |
| brokerDestinations.setConnection(null); |
| } catch (Exception e) { |
| // Ignore this for now. Attempting to close connection that has been closed |
| // Ignore we are shutting down |
| } finally { |
| for (Entry<Object, JmsEndpointConnection_impl> endpoints : brokerDestinations.endpointMap |
| .entrySet()) { |
| endpoints.getValue().close(); // close session and producer |
| } |
| brokerDestinations.endpointMap.clear(); |
| connectionMap.remove(endpoint.getServerURI()); |
| } |
| } |
| brokerDestinations.setConnection(null); |
| } catch (Exception e) { |
| } finally { |
| removeDestinationFromManagedList(brokerDestinations, endpoint); |
| } |
| } |
| cancelTimer(); |
| } |
| }, timeToRun); |
| } |
| |
| private void removeDestinationFromManagedList(BrokerConnectionEntry brokerDestinations, |
| Endpoint endpoint) { |
| String key = endpoint.getEndpoint() + endpoint.getServerURI(); |
| String destination = endpoint.getEndpoint(); |
| if (endpoint.getDestination() != null |
| && endpoint.getDestination() instanceof ActiveMQDestination) { |
| destination = ((ActiveMQDestination) endpoint.getDestination()).getPhysicalName(); |
| key = destination; |
| } |
| if (brokerDestinations.endpointExists(key)) { |
| brokerDestinations.removeEndpoint(key); |
| } |
| |
| } |
| |
| private void cancelTimer() { |
| if (timer != null) { |
| timer.cancel(); |
| timer.purge(); |
| } |
| } |
| |
| public synchronized void stopTimer() { |
| cancelTimer(); |
| timer = null; |
| } |
| } |
| |
| private static class RecoveryThread implements Runnable { |
| Endpoint endpoint; |
| |
| CacheEntry entry; |
| |
| boolean isRequest; |
| |
| AnalysisEngineController controller; |
| |
| JmsOutputChannel outputChannel; |
| |
| public RecoveryThread(JmsOutputChannel channel, Endpoint anEndpoint, CacheEntry anEntry, |
| boolean isRequest, AnalysisEngineController aController) { |
| endpoint = anEndpoint; |
| entry = anEntry; |
| controller = aController; |
| this.isRequest = isRequest; |
| outputChannel = channel; |
| } |
| |
| public void run() { |
| Delegate delegate = outputChannel.lookupDelegate(endpoint.getDelegateKey()); |
| // Removes the failed CAS from the list of CASes pending reply. This also |
| // cancels the timer if this CAS was the oldest pending CAS, and if there |
| // are other CASes pending a fresh timer is started. |
| outputChannel.removeCasFromOutstandingList(entry, isRequest, endpoint.getDelegateKey()); |
| if (delegate != null) { |
| // Mark this delegate as Failed |
| delegate.getEndpoint().setStatus(Endpoint.FAILED); |
| // Destroy listener associated with a reply queue for this delegate |
| InputChannel ic = controller.getInputChannel(delegate.getEndpoint().getDestination() |
| .toString()); |
| if (ic != null && delegate != null && delegate.getEndpoint() != null) { |
| ic.destroyListener(delegate.getEndpoint().getDestination().toString(), endpoint |
| .getDelegateKey()); |
| } |
| } |
| // Setup error context and handle failure in the error handler |
| ErrorContext errorContext = new ErrorContext(); |
| errorContext.add(AsynchAEMessage.Command, AsynchAEMessage.Process); |
| errorContext.add(AsynchAEMessage.CasReference, entry.getCasReferenceId()); |
| errorContext.add(AsynchAEMessage.Endpoint, endpoint); |
| errorContext.handleSilently(true); // dont dump exception to the log |
| // Failure on send treat as timeout |
| delegate.handleError(new MessageTimeoutException(), errorContext); |
| |
| } |
| } |
| } |
