You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by cl...@apache.org on 2014/11/19 00:38:08 UTC
[15/52] [abbrv] [partial] activemq-6 git commit: ACTIVEMQ6-4 Rename
HornetQ* classes to ActiveMQ*
http://git-wip-us.apache.org/repos/asf/activemq-6/blob/034adfbf/activemq-ra/src/main/java/org/apache/activemq/ra/inflow/ActiveMQActivation.java
----------------------------------------------------------------------
diff --git a/activemq-ra/src/main/java/org/apache/activemq/ra/inflow/ActiveMQActivation.java b/activemq-ra/src/main/java/org/apache/activemq/ra/inflow/ActiveMQActivation.java
new file mode 100644
index 0000000..d618e42
--- /dev/null
+++ b/activemq-ra/src/main/java/org/apache/activemq/ra/inflow/ActiveMQActivation.java
@@ -0,0 +1,831 @@
+/*
+ * Copyright 2005-2014 Red Hat, Inc.
+ * Red Hat 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.activemq.ra.inflow;
+
+import javax.jms.Destination;
+import javax.jms.Message;
+import javax.jms.MessageListener;
+import javax.jms.Queue;
+import javax.jms.Topic;
+import javax.naming.Context;
+import javax.naming.InitialContext;
+import javax.resource.ResourceException;
+import javax.resource.spi.endpoint.MessageEndpointFactory;
+import javax.resource.spi.work.Work;
+import javax.resource.spi.work.WorkManager;
+import javax.transaction.xa.XAResource;
+import java.lang.reflect.Method;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.activemq.api.core.ActiveMQException;
+import org.apache.activemq.api.core.ActiveMQExceptionType;
+import org.apache.activemq.api.core.ActiveMQNonExistentQueueException;
+import org.apache.activemq.api.core.ActiveMQNotConnectedException;
+import org.apache.activemq.api.core.SimpleString;
+import org.apache.activemq.api.core.client.ClientSession;
+import org.apache.activemq.api.core.client.ClientSessionFactory;
+import org.apache.activemq.api.jms.ActiveMQJMSClient;
+import org.apache.activemq.core.client.impl.ClientSessionInternal;
+import org.apache.activemq.jms.client.ActiveMQConnectionFactory;
+import org.apache.activemq.jms.client.ActiveMQDestination;
+import org.apache.activemq.jms.server.recovery.XARecoveryConfig;
+import org.apache.activemq.ra.ActiveMQRABundle;
+import org.apache.activemq.ra.ActiveMQRAConnectionFactory;
+import org.apache.activemq.ra.ActiveMQRALogger;
+import org.apache.activemq.ra.ActiveMQRaUtils;
+import org.apache.activemq.ra.ActiveMQResourceAdapter;
+import org.apache.activemq.utils.FutureLatch;
+import org.apache.activemq.utils.SensitiveDataCodec;
+
+/**
+ * The activation.
+ *
+ * @author <a href="adrian@jboss.com">Adrian Brock</a>
+ * @author <a href="jesper.pedersen@jboss.org">Jesper Pedersen</a>
+ * @author <a href="mailto:andy.taylor@jboss.org">Andy Taylor</a>
+ */
+public class ActiveMQActivation
+{
+ /**
+ * Trace enabled
+ */
+ private static boolean trace = ActiveMQRALogger.LOGGER.isTraceEnabled();
+
+ /**
+ * The onMessage method
+ */
+ public static final Method ONMESSAGE;
+
+ /**
+ * The resource adapter
+ */
+ private final ActiveMQResourceAdapter ra;
+
+ /**
+ * The activation spec
+ */
+ private final ActiveMQActivationSpec spec;
+
+ /**
+ * The message endpoint factory
+ */
+ private final MessageEndpointFactory endpointFactory;
+
+ /**
+ * Whether delivery is active
+ */
+ private final AtomicBoolean deliveryActive = new AtomicBoolean(false);
+
+ /**
+ * The destination type
+ */
+ private boolean isTopic = false;
+
+ /**
+ * Is the delivery transacted
+ */
+ private boolean isDeliveryTransacted;
+
+ private ActiveMQDestination destination;
+
+ /**
+ * The name of the temporary subscription name that all the sessions will share
+ */
+ private SimpleString topicTemporaryQueue;
+
+ private final List<ActiveMQMessageHandler> handlers = new ArrayList<ActiveMQMessageHandler>();
+
+ private ActiveMQConnectionFactory factory;
+
+ // Whether we are in the failure recovery loop
+ private final AtomicBoolean inFailure = new AtomicBoolean(false);
+ private XARecoveryConfig resourceRecovery;
+
+ static
+ {
+ try
+ {
+ ONMESSAGE = MessageListener.class.getMethod("onMessage", new Class[]{Message.class});
+ }
+ catch (Exception e)
+ {
+ throw new RuntimeException(e);
+ }
+ }
+
+ /**
+ * Constructor
+ *
+ * @param ra The resource adapter
+ * @param endpointFactory The endpoint factory
+ * @param spec The activation spec
+ * @throws ResourceException Thrown if an error occurs
+ */
+ public ActiveMQActivation(final ActiveMQResourceAdapter ra,
+ final MessageEndpointFactory endpointFactory,
+ final ActiveMQActivationSpec spec) throws ResourceException
+ {
+ spec.validate();
+
+ if (ActiveMQActivation.trace)
+ {
+ ActiveMQRALogger.LOGGER.trace("constructor(" + ra + ", " + endpointFactory + ", " + spec + ")");
+ }
+
+ if (ra.isUseMaskedPassword())
+ {
+ String pass = spec.getOwnPassword();
+ if (pass != null)
+ {
+ SensitiveDataCodec<String> codec = ra.getCodecInstance();
+
+ try
+ {
+ spec.setPassword(codec.decode(pass));
+ }
+ catch (Exception e)
+ {
+ throw new ResourceException(e);
+ }
+ }
+ }
+
+ this.ra = ra;
+ this.endpointFactory = endpointFactory;
+ this.spec = spec;
+ try
+ {
+ isDeliveryTransacted = endpointFactory.isDeliveryTransacted(ActiveMQActivation.ONMESSAGE);
+ }
+ catch (Exception e)
+ {
+ throw new ResourceException(e);
+ }
+ }
+
+ /**
+ * Get the activation spec
+ *
+ * @return The value
+ */
+ public ActiveMQActivationSpec getActivationSpec()
+ {
+ if (ActiveMQActivation.trace)
+ {
+ ActiveMQRALogger.LOGGER.trace("getActivationSpec()");
+ }
+
+ return spec;
+ }
+
+ /**
+ * Get the message endpoint factory
+ *
+ * @return The value
+ */
+ public MessageEndpointFactory getMessageEndpointFactory()
+ {
+ if (ActiveMQActivation.trace)
+ {
+ ActiveMQRALogger.LOGGER.trace("getMessageEndpointFactory()");
+ }
+
+ return endpointFactory;
+ }
+
+ /**
+ * Get whether delivery is transacted
+ *
+ * @return The value
+ */
+ public boolean isDeliveryTransacted()
+ {
+ if (ActiveMQActivation.trace)
+ {
+ ActiveMQRALogger.LOGGER.trace("isDeliveryTransacted()");
+ }
+
+ return isDeliveryTransacted;
+ }
+
+ /**
+ * Get the work manager
+ *
+ * @return The value
+ */
+ public WorkManager getWorkManager()
+ {
+ if (ActiveMQActivation.trace)
+ {
+ ActiveMQRALogger.LOGGER.trace("getWorkManager()");
+ }
+
+ return ra.getWorkManager();
+ }
+
+ /**
+ * Is the destination a topic
+ *
+ * @return The value
+ */
+ public boolean isTopic()
+ {
+ if (ActiveMQActivation.trace)
+ {
+ ActiveMQRALogger.LOGGER.trace("isTopic()");
+ }
+
+ return isTopic;
+ }
+
+ /**
+ * Start the activation
+ *
+ * @throws ResourceException Thrown if an error occurs
+ */
+ public void start() throws ResourceException
+ {
+ if (ActiveMQActivation.trace)
+ {
+ ActiveMQRALogger.LOGGER.trace("start()");
+ }
+ deliveryActive.set(true);
+ ra.getWorkManager().scheduleWork(new SetupActivation());
+ }
+
+ /**
+ * @return the topicTemporaryQueue
+ */
+ public SimpleString getTopicTemporaryQueue()
+ {
+ return topicTemporaryQueue;
+ }
+
+ /**
+ * @param topicTemporaryQueue the topicTemporaryQueue to set
+ */
+ public void setTopicTemporaryQueue(SimpleString topicTemporaryQueue)
+ {
+ this.topicTemporaryQueue = topicTemporaryQueue;
+ }
+
+ /**
+ * @return the list of XAResources for this activation endpoint
+ */
+ public List<XAResource> getXAResources()
+ {
+ List<XAResource> xaresources = new ArrayList<XAResource>();
+ for (ActiveMQMessageHandler handler : handlers)
+ {
+ XAResource xares = handler.getXAResource();
+ if (xares != null)
+ {
+ xaresources.add(xares);
+ }
+ }
+ return xaresources;
+ }
+
+ /**
+ * Stop the activation
+ */
+ public void stop()
+ {
+ if (ActiveMQActivation.trace)
+ {
+ ActiveMQRALogger.LOGGER.trace("stop()");
+ }
+
+ deliveryActive.set(false);
+ teardown();
+ }
+
+ /**
+ * Setup the activation
+ *
+ * @throws Exception Thrown if an error occurs
+ */
+ protected synchronized void setup() throws Exception
+ {
+ ActiveMQRALogger.LOGGER.debug("Setting up " + spec);
+
+ setupCF();
+
+ setupDestination();
+
+ Exception firstException = null;
+
+ for (int i = 0; i < spec.getMaxSession(); i++)
+ {
+ ClientSessionFactory cf = null;
+ ClientSession session = null;
+
+ try
+ {
+ cf = factory.getServerLocator().createSessionFactory();
+ session = setupSession(cf);
+ ActiveMQMessageHandler handler = new ActiveMQMessageHandler(this, ra.getTM(), (ClientSessionInternal) session, cf, i);
+ handler.setup();
+ handlers.add(handler);
+ }
+ catch (Exception e)
+ {
+ if (cf != null)
+ {
+ cf.close();
+ }
+ if (session != null)
+ {
+ session.close();
+ }
+ if (firstException == null)
+ {
+ firstException = e;
+ }
+ }
+ }
+ //if we have any exceptions close all the handlers and throw the first exception.
+ //we don't want partially configured activations, i.e. only 8 out of 15 sessions started so best to stop and log the error.
+ if (firstException != null)
+ {
+ for (ActiveMQMessageHandler handler : handlers)
+ {
+ handler.teardown();
+ }
+ throw firstException;
+ }
+
+ //now start them all together.
+ for (ActiveMQMessageHandler handler : handlers)
+ {
+ handler.start();
+ }
+
+ resourceRecovery = ra.getRecoveryManager().register(factory, spec.getUser(), spec.getPassword());
+
+ ActiveMQRALogger.LOGGER.debug("Setup complete " + this);
+ }
+
+ /**
+ * Teardown the activation
+ */
+ protected synchronized void teardown()
+ {
+ ActiveMQRALogger.LOGGER.debug("Tearing down " + spec);
+
+ if (resourceRecovery != null)
+ {
+ ra.getRecoveryManager().unRegister(resourceRecovery);
+ }
+
+ final ActiveMQMessageHandler[] handlersCopy = new ActiveMQMessageHandler[handlers.size()];
+
+ // We need to do from last to first as any temporary queue will have been created on the first handler
+ // So we invert the handlers here
+ for (int i = 0; i < handlers.size(); i++)
+ {
+ // The index here is the complimentary so it's inverting the array
+ handlersCopy[i] = handlers.get(handlers.size() - i - 1);
+ }
+
+ handlers.clear();
+
+ FutureLatch future = new FutureLatch(handlersCopy.length);
+ List<Thread> interruptThreads = new ArrayList<>();
+ for (ActiveMQMessageHandler handler : handlersCopy)
+ {
+ Thread thread = handler.interruptConsumer(future);
+ if (thread != null)
+ {
+ interruptThreads.add(thread);
+ }
+ }
+
+ //wait for all the consumers to complete any onmessage calls
+ boolean stuckThreads = !future.await(factory.getCallTimeout());
+ //if any are stuck then we need to interrupt them
+ if (stuckThreads)
+ {
+ for (Thread interruptThread : interruptThreads)
+ {
+ try
+ {
+ interruptThread.interrupt();
+ }
+ catch (Exception e)
+ {
+ //ok
+ }
+ }
+ }
+
+ Thread threadTearDown = new Thread("TearDown/ActiveMQActivation")
+ {
+ public void run()
+ {
+ for (ActiveMQMessageHandler handler : handlersCopy)
+ {
+ handler.teardown();
+ }
+ }
+ };
+
+ // We will first start a new thread that will call tearDown on all the instances, trying to graciously shutdown everything.
+ // We will then use the call-timeout to determine a timeout.
+ // if that failed we will then close the connection factory, and interrupt the thread
+ threadTearDown.start();
+
+ try
+ {
+ threadTearDown.join(factory.getCallTimeout());
+ }
+ catch (InterruptedException e)
+ {
+ // nothing to be done on this context.. we will just keep going as we need to send an interrupt to threadTearDown and give up
+ }
+
+ if (threadTearDown.isAlive())
+ {
+ if (factory != null)
+ {
+ // This will interrupt any threads waiting on reconnect
+ factory.close();
+ factory = null;
+ }
+ threadTearDown.interrupt();
+
+ try
+ {
+ threadTearDown.join(5000);
+ }
+ catch (InterruptedException e)
+ {
+ // nothing to be done here.. we are going down anyways
+ }
+
+ if (threadTearDown.isAlive())
+ {
+ ActiveMQRALogger.LOGGER.warn("Thread " + threadTearDown + " couldn't be finished");
+ }
+ }
+
+ if (spec.isHasBeenUpdated() && factory != null)
+ {
+ factory.close();
+ factory = null;
+ }
+
+
+ ActiveMQRALogger.LOGGER.debug("Tearing down complete " + this);
+ }
+
+ protected void setupCF() throws Exception
+ {
+ if (spec.getConnectionFactoryLookup() != null)
+ {
+ Context ctx;
+ if (spec.getParsedJndiParams() == null)
+ {
+ ctx = new InitialContext();
+ }
+ else
+ {
+ ctx = new InitialContext(spec.getParsedJndiParams());
+ }
+ Object fac = ctx.lookup(spec.getConnectionFactoryLookup());
+ if (fac instanceof ActiveMQConnectionFactory)
+ {
+ factory = (ActiveMQConnectionFactory) fac;
+ }
+ else
+ {
+ ActiveMQRAConnectionFactory raFact = (ActiveMQRAConnectionFactory) fac;
+ if (spec.isHasBeenUpdated())
+ {
+ factory = raFact.getResourceAdapter().createActiveMQConnectionFactory(spec);
+ }
+ else
+ {
+ factory = raFact.getDefaultFactory();
+ if (factory != ra.getDefaultActiveMQConnectionFactory())
+ {
+ ActiveMQRALogger.LOGGER.warnDifferentConnectionfactory();
+ }
+ }
+ }
+ }
+ else if (spec.isHasBeenUpdated())
+ {
+ factory = ra.createActiveMQConnectionFactory(spec);
+ }
+ else
+ {
+ factory = ra.getDefaultActiveMQConnectionFactory();
+ }
+ }
+
+ /**
+ * Setup a session
+ *
+ * @param cf
+ * @return The connection
+ * @throws Exception Thrown if an error occurs
+ */
+ protected ClientSession setupSession(ClientSessionFactory cf) throws Exception
+ {
+ ClientSession result = null;
+
+ try
+ {
+ result = ra.createSession(cf,
+ spec.getAcknowledgeModeInt(),
+ spec.getUser(),
+ spec.getPassword(),
+ ra.getPreAcknowledge(),
+ ra.getDupsOKBatchSize(),
+ ra.getTransactionBatchSize(),
+ isDeliveryTransacted,
+ spec.isUseLocalTx(),
+ spec.getTransactionTimeout());
+
+ result.addMetaData("resource-adapter", "inbound");
+ result.addMetaData("jms-session", "");
+ String clientID = ra.getClientID() == null ? spec.getClientID() : ra.getClientID();
+ if (clientID != null)
+ {
+ result.addMetaData("jms-client-id", clientID);
+ }
+
+ ActiveMQRALogger.LOGGER.debug("Using queue connection " + result);
+
+ return result;
+ }
+ catch (Throwable t)
+ {
+ try
+ {
+ if (result != null)
+ {
+ result.close();
+ }
+ }
+ catch (Exception e)
+ {
+ ActiveMQRALogger.LOGGER.trace("Ignored error closing connection", e);
+ }
+ if (t instanceof Exception)
+ {
+ throw (Exception) t;
+ }
+ throw new RuntimeException("Error configuring connection", t);
+ }
+ }
+
+ public SimpleString getAddress()
+ {
+ return destination.getSimpleAddress();
+ }
+
+ protected void setupDestination() throws Exception
+ {
+
+ String destinationName = spec.getDestination();
+
+ if (spec.isUseJNDI())
+ {
+ Context ctx;
+ if (spec.getParsedJndiParams() == null)
+ {
+ ctx = new InitialContext();
+ }
+ else
+ {
+ ctx = new InitialContext(spec.getParsedJndiParams());
+ }
+ ActiveMQRALogger.LOGGER.debug("Using context " + ctx.getEnvironment() + " for " + spec);
+ if (ActiveMQActivation.trace)
+ {
+ ActiveMQRALogger.LOGGER.trace("setupDestination(" + ctx + ")");
+ }
+
+ String destinationTypeString = spec.getDestinationType();
+ if (destinationTypeString != null && !destinationTypeString.trim().equals(""))
+ {
+ ActiveMQRALogger.LOGGER.debug("Destination type defined as " + destinationTypeString);
+
+ Class<?> destinationType;
+ if (Topic.class.getName().equals(destinationTypeString))
+ {
+ destinationType = Topic.class;
+ isTopic = true;
+ }
+ else
+ {
+ destinationType = Queue.class;
+ }
+
+ ActiveMQRALogger.LOGGER.debug("Retrieving " + destinationType.getName() + " \"" + destinationName + "\" from JNDI");
+
+ try
+ {
+ destination = (ActiveMQDestination) ActiveMQRaUtils.lookup(ctx, destinationName, destinationType);
+ }
+ catch (Exception e)
+ {
+ if (destinationName == null)
+ {
+ throw ActiveMQRABundle.BUNDLE.noDestinationName();
+ }
+
+ String calculatedDestinationName = destinationName.substring(destinationName.lastIndexOf('/') + 1);
+
+ ActiveMQRALogger.LOGGER.debug("Unable to retrieve " + destinationName +
+ " from JNDI. Creating a new " + destinationType.getName() +
+ " named " + calculatedDestinationName + " to be used by the MDB.");
+
+ // If there is no binding on naming, we will just create a new instance
+ if (isTopic)
+ {
+ destination = (ActiveMQDestination) ActiveMQJMSClient.createTopic(calculatedDestinationName);
+ }
+ else
+ {
+ destination = (ActiveMQDestination) ActiveMQJMSClient.createQueue(calculatedDestinationName);
+ }
+ }
+ }
+ else
+ {
+ ActiveMQRALogger.LOGGER.debug("Destination type not defined in MDB activation configuration.");
+ ActiveMQRALogger.LOGGER.debug("Retrieving " + Destination.class.getName() + " \"" + destinationName + "\" from JNDI");
+
+ destination = (ActiveMQDestination) ActiveMQRaUtils.lookup(ctx, destinationName, Destination.class);
+ if (destination instanceof Topic)
+ {
+ isTopic = true;
+ }
+ }
+ }
+ else
+ {
+ ActiveMQRALogger.LOGGER.instantiatingDestination(spec.getDestinationType(), spec.getDestination());
+
+ if (Topic.class.getName().equals(spec.getDestinationType()))
+ {
+ destination = (ActiveMQDestination) ActiveMQJMSClient.createTopic(spec.getDestination());
+ isTopic = true;
+ }
+ else
+ {
+ destination = (ActiveMQDestination) ActiveMQJMSClient.createQueue(spec.getDestination());
+ }
+ }
+ }
+
+ /**
+ * Get a string representation
+ *
+ * @return The value
+ */
+ @Override
+ public String toString()
+ {
+ StringBuffer buffer = new StringBuffer();
+ buffer.append(ActiveMQActivation.class.getName()).append('(');
+ buffer.append("spec=").append(spec.getClass().getName());
+ buffer.append(" mepf=").append(endpointFactory.getClass().getName());
+ buffer.append(" active=").append(deliveryActive.get());
+ if (spec.getDestination() != null)
+ {
+ buffer.append(" destination=").append(spec.getDestination());
+ }
+ buffer.append(" transacted=").append(isDeliveryTransacted);
+ buffer.append(')');
+ return buffer.toString();
+ }
+
+ /**
+ * Handles any failure by trying to reconnect
+ *
+ * @param failure the reason for the failure
+ */
+ public void handleFailure(Throwable failure)
+ {
+ if (failure instanceof ActiveMQException && ((ActiveMQException) failure).getType() == ActiveMQExceptionType.QUEUE_DOES_NOT_EXIST)
+ {
+ ActiveMQRALogger.LOGGER.awaitingTopicQueueCreation(getActivationSpec().getDestination());
+ }
+ else if (failure instanceof ActiveMQException && ((ActiveMQException) failure).getType() == ActiveMQExceptionType.NOT_CONNECTED)
+ {
+ ActiveMQRALogger.LOGGER.awaitingJMSServerCreation();
+ }
+ else
+ {
+ ActiveMQRALogger.LOGGER.failureInActivation(failure, spec);
+ }
+ int reconnectCount = 0;
+ int setupAttempts = spec.getSetupAttempts();
+ long setupInterval = spec.getSetupInterval();
+
+ // Only enter the failure loop once
+ if (inFailure.getAndSet(true))
+ return;
+ try
+ {
+ Throwable lastException = failure;
+ while (deliveryActive.get() && (setupAttempts == -1 || reconnectCount < setupAttempts))
+ {
+ teardown();
+
+ try
+ {
+ Thread.sleep(setupInterval);
+ }
+ catch (InterruptedException e)
+ {
+ ActiveMQRALogger.LOGGER.debug("Interrupted trying to reconnect " + spec, e);
+ break;
+ }
+
+ if (reconnectCount < 1)
+ {
+ ActiveMQRALogger.LOGGER.attemptingReconnect(spec);
+ }
+ try
+ {
+ setup();
+ ActiveMQRALogger.LOGGER.reconnected();
+ break;
+ }
+ catch (Throwable t)
+ {
+ if (failure instanceof ActiveMQException && ((ActiveMQException) failure).getType() == ActiveMQExceptionType.QUEUE_DOES_NOT_EXIST)
+ {
+ if (lastException == null || !(t instanceof ActiveMQNonExistentQueueException))
+ {
+ lastException = t;
+ ActiveMQRALogger.LOGGER.awaitingTopicQueueCreation(getActivationSpec().getDestination());
+ }
+ }
+ else if (failure instanceof ActiveMQException && ((ActiveMQException) failure).getType() == ActiveMQExceptionType.NOT_CONNECTED)
+ {
+ if (lastException == null || !(t instanceof ActiveMQNotConnectedException))
+ {
+ lastException = t;
+ ActiveMQRALogger.LOGGER.awaitingJMSServerCreation();
+ }
+ }
+ else
+ {
+ ActiveMQRALogger.LOGGER.errorReconnecting(t, spec);
+ }
+ }
+ ++reconnectCount;
+ }
+ }
+ finally
+ {
+ // Leaving failure recovery loop
+ inFailure.set(false);
+ }
+ }
+
+ public ActiveMQConnectionFactory getConnectionFactory()
+ {
+ return this.factory;
+ }
+
+ /**
+ * Handles the setup
+ */
+ private class SetupActivation implements Work
+ {
+ public void run()
+ {
+ try
+ {
+ setup();
+ }
+ catch (Throwable t)
+ {
+ handleFailure(t);
+ }
+ }
+
+ public void release()
+ {
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/activemq-6/blob/034adfbf/activemq-ra/src/main/java/org/apache/activemq/ra/inflow/ActiveMQActivationSpec.java
----------------------------------------------------------------------
diff --git a/activemq-ra/src/main/java/org/apache/activemq/ra/inflow/ActiveMQActivationSpec.java b/activemq-ra/src/main/java/org/apache/activemq/ra/inflow/ActiveMQActivationSpec.java
new file mode 100644
index 0000000..06f9c7a
--- /dev/null
+++ b/activemq-ra/src/main/java/org/apache/activemq/ra/inflow/ActiveMQActivationSpec.java
@@ -0,0 +1,945 @@
+/*
+ * Copyright 2005-2014 Red Hat, Inc.
+ * Red Hat 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.activemq.ra.inflow;
+
+import javax.jms.Queue;
+import javax.jms.Session;
+import javax.jms.Topic;
+import javax.resource.ResourceException;
+import javax.resource.spi.ActivationSpec;
+import javax.resource.spi.InvalidPropertyException;
+import javax.resource.spi.ResourceAdapter;
+import java.beans.IntrospectionException;
+import java.beans.PropertyDescriptor;
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Hashtable;
+import java.util.Iterator;
+import java.util.List;
+
+import org.apache.activemq.ra.ConnectionFactoryProperties;
+import org.apache.activemq.ra.ActiveMQRALogger;
+import org.apache.activemq.ra.ActiveMQRaUtils;
+import org.apache.activemq.ra.ActiveMQResourceAdapter;
+
+/**
+ * The activation spec
+ * These properties are set on the MDB ActivactionProperties
+ *
+ * @author <a href="adrian@jboss.com">Adrian Brock</a>
+ * @author <a href="jesper.pedersen@jboss.org">Jesper Pedersen</a>
+ * @author <a href="mailto:andy.taylor@jboss.org">Andy Taylor</a>
+ * @author <a href="mailto:clebert.suconic@jboss.org">Clebert Suconic</a>
+ */
+public class ActiveMQActivationSpec extends ConnectionFactoryProperties implements ActivationSpec, Serializable
+{
+ private static final long serialVersionUID = -7997041053897964654L;
+
+ private static final int DEFAULT_MAX_SESSION = 15;
+
+ /**
+ * Whether trace is enabled
+ */
+ private static boolean trace = ActiveMQRALogger.LOGGER.isTraceEnabled();
+
+ public String strConnectorClassName;
+
+ public String strConnectionParameters;
+
+ /**
+ * The resource adapter
+ */
+ private ActiveMQResourceAdapter ra;
+
+ /**
+ * The connection factory lookup
+ */
+ private String connectionFactoryLookup;
+
+ /**
+ * The destination
+ */
+ private String destination;
+
+ /**
+ * The destination type
+ */
+ private String destinationType;
+
+ /**
+ * The message selector
+ */
+ private String messageSelector;
+
+ /**
+ * The acknowledgement mode
+ */
+ private int acknowledgeMode;
+
+ /**
+ * The subscription durability
+ */
+ private boolean subscriptionDurability;
+
+ /**
+ * The subscription name
+ */
+ private String subscriptionName;
+
+ /**
+ * If this is true, a durable subscription could be shared by multiple MDB instances
+ */
+ private boolean shareSubscriptions;
+
+ /**
+ * The user
+ */
+ private String user;
+
+ /**
+ * The password
+ */
+ private String password;
+
+ /**
+ * The maximum number of sessions
+ */
+ private Integer maxSession;
+
+ /**
+ * Transaction timeout
+ */
+ private Integer transactionTimeout;
+
+ private Boolean useJNDI = true;
+
+ private String jndiParams = null;
+
+ private Hashtable parsedJndiParams;
+
+ /* use local tx instead of XA*/
+ private Boolean localTx;
+
+ // undefined by default, default is specified at the RA level in ActiveMQRAProperties
+ private Integer setupAttempts;
+
+ // undefined by default, default is specified at the RA level in ActiveMQRAProperties
+ private Long setupInterval;
+
+ /**
+ * Constructor
+ */
+ public ActiveMQActivationSpec()
+ {
+ if (ActiveMQActivationSpec.trace)
+ {
+ ActiveMQRALogger.LOGGER.trace("constructor()");
+ }
+
+ ra = null;
+ destination = null;
+ destinationType = null;
+ messageSelector = null;
+ acknowledgeMode = Session.AUTO_ACKNOWLEDGE;
+ subscriptionDurability = false;
+ subscriptionName = null;
+ user = null;
+ password = null;
+ maxSession = DEFAULT_MAX_SESSION;
+ transactionTimeout = 0;
+ }
+
+ /**
+ * Get the resource adapter
+ *
+ * @return The resource adapter
+ */
+ public ResourceAdapter getResourceAdapter()
+ {
+ if (ActiveMQActivationSpec.trace)
+ {
+ ActiveMQRALogger.LOGGER.trace("getResourceAdapter()");
+ }
+
+ return ra;
+ }
+
+ /**
+ * @return the useJNDI
+ */
+ public boolean isUseJNDI()
+ {
+ if (useJNDI == null)
+ {
+ return ra.isUseJNDI();
+ }
+ return useJNDI;
+ }
+
+ /**
+ * @param value the useJNDI to set
+ */
+ public void setUseJNDI(final boolean value)
+ {
+ useJNDI = value;
+ }
+
+ /**
+ * @return return the jndi params to use
+ */
+ public String getJndiParams()
+ {
+ if (jndiParams == null)
+ {
+ return ra.getJndiParams();
+ }
+ return jndiParams;
+ }
+
+ public void setJndiParams(String jndiParams)
+ {
+ this.jndiParams = jndiParams;
+ parsedJndiParams = ActiveMQRaUtils.parseHashtableConfig(jndiParams);
+ }
+
+ public Hashtable<?, ?> getParsedJndiParams()
+ {
+ if (parsedJndiParams == null)
+ {
+ return ra.getParsedJndiParams();
+ }
+ return parsedJndiParams;
+ }
+
+ /**
+ * Set the resource adapter
+ *
+ * @param ra The resource adapter
+ * @throws ResourceException Thrown if incorrect resource adapter
+ */
+ public void setResourceAdapter(final ResourceAdapter ra) throws ResourceException
+ {
+ if (ActiveMQActivationSpec.trace)
+ {
+ ActiveMQRALogger.LOGGER.trace("setResourceAdapter(" + ra + ")");
+ }
+
+ if (ra == null || !(ra instanceof ActiveMQResourceAdapter))
+ {
+ throw new ResourceException("Resource adapter is " + ra);
+ }
+
+ this.ra = (ActiveMQResourceAdapter) ra;
+ }
+
+ /**
+ * Get the connection factory lookup
+ *
+ * @return The value
+ */
+ public String getConnectionFactoryLookup()
+ {
+ if (ActiveMQActivationSpec.trace)
+ {
+ ActiveMQRALogger.LOGGER.trace("getConnectionFactoryLookup() ->" + connectionFactoryLookup);
+ }
+
+ return connectionFactoryLookup;
+ }
+
+ /**
+ * Set the connection factory lookup
+ *
+ * @param value The value
+ */
+ public void setConnectionFactoryLookup(final String value)
+ {
+ if (ActiveMQActivationSpec.trace)
+ {
+ ActiveMQRALogger.LOGGER.trace("setConnectionFactoryLookup(" + value + ")");
+ }
+
+ connectionFactoryLookup = value;
+ }
+
+ /**
+ * Get the destination
+ *
+ * @return The value
+ */
+ public String getDestination()
+ {
+ if (ActiveMQActivationSpec.trace)
+ {
+ ActiveMQRALogger.LOGGER.trace("getDestination()");
+ }
+
+ return destination;
+ }
+
+ /**
+ * Set the destination
+ *
+ * @param value The value
+ */
+ public void setDestination(final String value)
+ {
+ if (ActiveMQActivationSpec.trace)
+ {
+ ActiveMQRALogger.LOGGER.trace("setDestination(" + value + ")");
+ }
+
+ destination = value;
+ }
+
+ /**
+ * Get the destination lookup
+ *
+ * @return The value
+ */
+ public String getDestinationLookup()
+ {
+ return getDestination();
+ }
+
+ /**
+ * Set the destination
+ *
+ * @param value The value
+ */
+ public void setDestinationLookup(final String value)
+ {
+ setDestination(value);
+ setUseJNDI(true);
+ }
+
+ /**
+ * Get the destination type
+ *
+ * @return The value
+ */
+ public String getDestinationType()
+ {
+ if (ActiveMQActivationSpec.trace)
+ {
+ ActiveMQRALogger.LOGGER.trace("getDestinationType()");
+ }
+
+ return destinationType;
+ }
+
+ /**
+ * Set the destination type
+ *
+ * @param value The value
+ */
+ public void setDestinationType(final String value)
+ {
+ if (ActiveMQActivationSpec.trace)
+ {
+ ActiveMQRALogger.LOGGER.trace("setDestinationType(" + value + ")");
+ }
+
+ destinationType = value;
+ }
+
+ /**
+ * Get the message selector
+ *
+ * @return The value
+ */
+ public String getMessageSelector()
+ {
+ if (ActiveMQActivationSpec.trace)
+ {
+ ActiveMQRALogger.LOGGER.trace("getMessageSelector()");
+ }
+
+ return messageSelector;
+ }
+
+ /**
+ * Set the message selector
+ *
+ * @param value The value
+ */
+ public void setMessageSelector(final String value)
+ {
+ if (ActiveMQActivationSpec.trace)
+ {
+ ActiveMQRALogger.LOGGER.trace("setMessageSelector(" + value + ")");
+ }
+
+ messageSelector = value;
+ }
+
+ /**
+ * Get the acknowledge mode
+ *
+ * @return The value
+ */
+ public String getAcknowledgeMode()
+ {
+ if (ActiveMQActivationSpec.trace)
+ {
+ ActiveMQRALogger.LOGGER.trace("getAcknowledgeMode()");
+ }
+
+ if (Session.DUPS_OK_ACKNOWLEDGE == acknowledgeMode)
+ {
+ return "Dups-ok-acknowledge";
+ }
+ else
+ {
+ return "Auto-acknowledge";
+ }
+ }
+
+ /**
+ * Set the acknowledge mode
+ *
+ * @param value The value
+ */
+ public void setAcknowledgeMode(final String value)
+ {
+ if (ActiveMQActivationSpec.trace)
+ {
+ ActiveMQRALogger.LOGGER.trace("setAcknowledgeMode(" + value + ")");
+ }
+
+ if ("DUPS_OK_ACKNOWLEDGE".equalsIgnoreCase(value) || "Dups-ok-acknowledge".equalsIgnoreCase(value))
+ {
+ acknowledgeMode = Session.DUPS_OK_ACKNOWLEDGE;
+ }
+ else if ("AUTO_ACKNOWLEDGE".equalsIgnoreCase(value) || "Auto-acknowledge".equalsIgnoreCase(value))
+ {
+ acknowledgeMode = Session.AUTO_ACKNOWLEDGE;
+ }
+ else
+ {
+ throw new IllegalArgumentException("Unsupported acknowledgement mode " + value);
+ }
+ }
+
+ /**
+ * @return the acknowledgement mode
+ */
+ public int getAcknowledgeModeInt()
+ {
+ if (ActiveMQActivationSpec.trace)
+ {
+ ActiveMQRALogger.LOGGER.trace("getAcknowledgeMode()");
+ }
+
+ return acknowledgeMode;
+ }
+
+ /**
+ * Get the subscription durability
+ *
+ * @return The value
+ */
+ public String getSubscriptionDurability()
+ {
+ if (ActiveMQActivationSpec.trace)
+ {
+ ActiveMQRALogger.LOGGER.trace("getSubscriptionDurability()");
+ }
+
+ if (subscriptionDurability)
+ {
+ return "Durable";
+ }
+ else
+ {
+ return "NonDurable";
+ }
+ }
+
+ /**
+ * Set the subscription durability
+ *
+ * @param value The value
+ */
+ public void setSubscriptionDurability(final String value)
+ {
+ if (ActiveMQActivationSpec.trace)
+ {
+ ActiveMQRALogger.LOGGER.trace("setSubscriptionDurability(" + value + ")");
+ }
+
+ subscriptionDurability = "Durable".equals(value);
+ }
+
+ /**
+ * Get the status of subscription durability
+ *
+ * @return The value
+ */
+ public boolean isSubscriptionDurable()
+ {
+ if (ActiveMQActivationSpec.trace)
+ {
+ ActiveMQRALogger.LOGGER.trace("isSubscriptionDurable()");
+ }
+
+ return subscriptionDurability;
+ }
+
+ /**
+ * Get the subscription name
+ *
+ * @return The value
+ */
+ public String getSubscriptionName()
+ {
+ if (ActiveMQActivationSpec.trace)
+ {
+ ActiveMQRALogger.LOGGER.trace("getSubscriptionName()");
+ }
+
+ return subscriptionName;
+ }
+
+ /**
+ * Set the subscription name
+ *
+ * @param value The value
+ */
+ public void setSubscriptionName(final String value)
+ {
+ if (ActiveMQActivationSpec.trace)
+ {
+ ActiveMQRALogger.LOGGER.trace("setSubscriptionName(" + value + ")");
+ }
+
+ subscriptionName = value;
+ }
+
+
+ /**
+ * @return the shareDurableSubscriptions
+ */
+ public boolean isShareSubscriptions()
+ {
+ if (ActiveMQActivationSpec.trace)
+ {
+ ActiveMQRALogger.LOGGER.trace("isShareSubscriptions() = " + shareSubscriptions);
+ }
+
+ return shareSubscriptions;
+ }
+
+ /**
+ * @param shareSubscriptions the shareDurableSubscriptions to set
+ */
+ public void setShareSubscriptions(boolean shareSubscriptions)
+ {
+ if (ActiveMQActivationSpec.trace)
+ {
+ ActiveMQRALogger.LOGGER.trace("setShareSubscriptions(" + shareSubscriptions + ")");
+ }
+
+ this.shareSubscriptions = shareSubscriptions;
+ }
+
+ /**
+ * Get the user
+ *
+ * @return The value
+ */
+ public String getUser()
+ {
+ if (ActiveMQActivationSpec.trace)
+ {
+ ActiveMQRALogger.LOGGER.trace("getUser()");
+ }
+
+ if (user == null)
+ {
+ return ra.getUserName();
+ }
+ else
+ {
+ return user;
+ }
+ }
+
+ /**
+ * Set the user
+ *
+ * @param value The value
+ */
+ public void setUser(final String value)
+ {
+ if (ActiveMQActivationSpec.trace)
+ {
+ ActiveMQRALogger.LOGGER.trace("setUser(" + value + ")");
+ }
+
+ user = value;
+ }
+
+ /**
+ * Get the password
+ *
+ * @return The value
+ */
+ public String getPassword()
+ {
+ if (ActiveMQActivationSpec.trace)
+ {
+ ActiveMQRALogger.LOGGER.trace("getPassword()");
+ }
+
+ if (password == null)
+ {
+ return ra.getPassword();
+ }
+ else
+ {
+ return password;
+ }
+ }
+
+ public String getOwnPassword()
+ {
+ return password;
+ }
+
+ /**
+ * Set the password
+ *
+ * @param value The value
+ */
+ public void setPassword(final String value) throws Exception
+ {
+ if (ActiveMQActivationSpec.trace)
+ {
+ ActiveMQRALogger.LOGGER.trace("setPassword(****)");
+ }
+
+ password = value;
+ }
+
+ /**
+ * Get the number of max session
+ *
+ * @return The value
+ */
+ public Integer getMaxSession()
+ {
+ if (ActiveMQActivationSpec.trace)
+ {
+ ActiveMQRALogger.LOGGER.trace("getMaxSession()");
+ }
+
+ if (maxSession == null)
+ {
+ return DEFAULT_MAX_SESSION;
+ }
+
+ return maxSession;
+ }
+
+ /**
+ * Set the number of max session
+ *
+ * @param value The value
+ */
+ public void setMaxSession(final Integer value)
+ {
+ if (ActiveMQActivationSpec.trace)
+ {
+ ActiveMQRALogger.LOGGER.trace("setMaxSession(" + value + ")");
+ }
+
+ maxSession = value;
+ }
+
+ /**
+ * Get the transaction timeout
+ *
+ * @return The value
+ */
+ public Integer getTransactionTimeout()
+ {
+ if (ActiveMQActivationSpec.trace)
+ {
+ ActiveMQRALogger.LOGGER.trace("getTransactionTimeout()");
+ }
+
+ return transactionTimeout;
+ }
+
+ /**
+ * Set the transaction timeout
+ *
+ * @param value The value
+ */
+ public void setTransactionTimeout(final Integer value)
+ {
+ if (ActiveMQActivationSpec.trace)
+ {
+ ActiveMQRALogger.LOGGER.trace("setTransactionTimeout(" + value + ")");
+ }
+
+ transactionTimeout = value;
+ }
+
+ public Boolean isUseLocalTx()
+ {
+ if (localTx == null)
+ {
+ return ra.getUseLocalTx();
+ }
+ else
+ {
+ return localTx;
+ }
+ }
+
+ public void setUseLocalTx(final Boolean localTx)
+ {
+ this.localTx = localTx;
+ }
+
+ public int getSetupAttempts()
+ {
+ if (ActiveMQActivationSpec.trace)
+ {
+ ActiveMQRALogger.LOGGER.trace("getSetupAttempts()");
+ }
+
+ if (setupAttempts == null)
+ {
+ return ra.getSetupAttempts();
+ }
+ else
+ {
+ return setupAttempts;
+ }
+ }
+
+ public void setSetupAttempts(int setupAttempts)
+ {
+ if (ActiveMQActivationSpec.trace)
+ {
+ ActiveMQRALogger.LOGGER.trace("setSetupAttempts(" + setupAttempts + ")");
+ }
+
+ this.setupAttempts = setupAttempts;
+ }
+
+ public long getSetupInterval()
+ {
+ if (ActiveMQActivationSpec.trace)
+ {
+ ActiveMQRALogger.LOGGER.trace("getSetupInterval()");
+ }
+
+ if (setupInterval == null)
+ {
+ return ra.getSetupInterval();
+ }
+ else
+ {
+ return setupInterval;
+ }
+ }
+
+ public void setSetupInterval(long setupInterval)
+ {
+ if (ActiveMQActivationSpec.trace)
+ {
+ ActiveMQRALogger.LOGGER.trace("setSetupInterval(" + setupInterval + ")");
+ }
+
+ this.setupInterval = setupInterval;
+ }
+
+ /**
+ * Validate
+ *
+ * @throws InvalidPropertyException Thrown if a validation exception occurs
+ */
+ public void validate() throws InvalidPropertyException
+ {
+ if (ActiveMQActivationSpec.trace)
+ {
+ ActiveMQRALogger.LOGGER.trace("validate()");
+ }
+
+ List<String> errorMessages = new ArrayList<String>();
+ List<PropertyDescriptor> propsNotSet = new ArrayList<PropertyDescriptor>();
+
+ try
+ {
+ if (destination == null || destination.trim().equals(""))
+ {
+ propsNotSet.add(new PropertyDescriptor("destination", ActiveMQActivationSpec.class));
+ errorMessages.add("Destination is mandatory.");
+ }
+
+ if (destinationType != null && !Topic.class.getName().equals(destinationType) && !Queue.class.getName().equals(destinationType))
+ {
+ propsNotSet.add(new PropertyDescriptor("destinationType", ActiveMQActivationSpec.class));
+ errorMessages.add("If set, the destinationType must be either 'javax.jms.Topic' or 'javax.jms.Queue'.");
+ }
+
+ if ((destinationType == null || destinationType.length() == 0 || Topic.class.getName().equals(destinationType)) && isSubscriptionDurable() && (subscriptionName == null || subscriptionName.length() == 0))
+ {
+ propsNotSet.add(new PropertyDescriptor("subscriptionName", ActiveMQActivationSpec.class));
+ errorMessages.add("If subscription is durable then subscription name must be specified.");
+ }
+ }
+ catch (IntrospectionException e)
+ {
+ e.printStackTrace();
+ }
+
+ if (propsNotSet.size() > 0)
+ {
+ StringBuffer b = new StringBuffer();
+ b.append("Invalid settings:");
+ for (Iterator<String> iter = errorMessages.iterator(); iter.hasNext();)
+ {
+ b.append(" ");
+ b.append(iter.next());
+ }
+ InvalidPropertyException e = new InvalidPropertyException(b.toString());
+ final PropertyDescriptor[] descriptors = propsNotSet.toArray(new PropertyDescriptor[propsNotSet.size()]);
+ e.setInvalidPropertyDescriptors(descriptors);
+ throw e;
+ }
+ }
+
+ public String getConnectorClassName()
+ {
+ return strConnectorClassName;
+ }
+
+ public void setConnectorClassName(final String connectorClassName)
+ {
+ if (ActiveMQActivationSpec.trace)
+ {
+ ActiveMQRALogger.LOGGER.trace("setConnectorClassName(" + connectorClassName + ")");
+ }
+
+ strConnectorClassName = connectorClassName;
+
+ setParsedConnectorClassNames(ActiveMQRaUtils.parseConnectorConnectorConfig(connectorClassName));
+ }
+
+ /**
+ * @return the connectionParameters
+ */
+ public String getConnectionParameters()
+ {
+ return strConnectionParameters;
+ }
+
+ public void setConnectionParameters(final String configuration)
+ {
+ strConnectionParameters = configuration;
+ setParsedConnectionParameters(ActiveMQRaUtils.parseConfig(configuration));
+ }
+
+ /**
+ * Get a string representation
+ *
+ * @return The value
+ */
+ @Override
+ public String toString()
+ {
+ StringBuffer buffer = new StringBuffer();
+ buffer.append(ActiveMQActivationSpec.class.getName()).append('(');
+ buffer.append("ra=").append(ra);
+ if (messageSelector != null)
+ {
+ buffer.append(" connectionFactoryLookup=").append(connectionFactoryLookup);
+ }
+ buffer.append(" destination=").append(destination);
+ buffer.append(" destinationType=").append(destinationType);
+ if (messageSelector != null)
+ {
+ buffer.append(" selector=").append(messageSelector);
+ }
+ buffer.append(" ack=").append(getAcknowledgeMode());
+ buffer.append(" durable=").append(subscriptionDurability);
+ buffer.append(" clientID=").append(getClientID());
+ if (subscriptionName != null)
+ {
+ buffer.append(" subscription=").append(subscriptionName);
+ }
+ buffer.append(" user=").append(user);
+ if (password != null)
+ {
+ buffer.append(" password=").append("****");
+ }
+ buffer.append(" maxSession=").append(maxSession);
+ buffer.append(')');
+ return buffer.toString();
+ }
+
+ // here for backwards compatibilty
+ public void setUseDLQ(final boolean b)
+ {
+ }
+
+ public void setDLQJNDIName(final String name)
+ {
+ }
+
+ public void setDLQHandler(final String handler)
+ {
+ }
+
+ public void setDLQMaxResent(final int maxResent)
+ {
+ }
+
+ public void setProviderAdapterJNDI(final String jndi)
+ {
+ }
+
+ /**
+ * @param keepAlive the keepAlive to set
+ */
+ public void setKeepAlive(boolean keepAlive)
+ {
+ }
+
+ /**
+ * @param keepAliveMillis the keepAliveMillis to set
+ */
+ public void setKeepAliveMillis(long keepAliveMillis)
+ {
+ }
+
+
+ public void setReconnectInterval(long interval)
+ {
+ }
+
+ public void setMinSession(final Integer value)
+ {
+ }
+
+ public void setMaxMessages(final Integer value)
+ {
+ }
+
+
+}
http://git-wip-us.apache.org/repos/asf/activemq-6/blob/034adfbf/activemq-ra/src/main/java/org/apache/activemq/ra/inflow/ActiveMQMessageHandler.java
----------------------------------------------------------------------
diff --git a/activemq-ra/src/main/java/org/apache/activemq/ra/inflow/ActiveMQMessageHandler.java b/activemq-ra/src/main/java/org/apache/activemq/ra/inflow/ActiveMQMessageHandler.java
new file mode 100644
index 0000000..abd68d0
--- /dev/null
+++ b/activemq-ra/src/main/java/org/apache/activemq/ra/inflow/ActiveMQMessageHandler.java
@@ -0,0 +1,430 @@
+/*
+ * Copyright 2005-2014 Red Hat, Inc.
+ * Red Hat 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.activemq.ra.inflow;
+
+import javax.jms.MessageListener;
+import javax.resource.ResourceException;
+import javax.resource.spi.endpoint.MessageEndpoint;
+import javax.resource.spi.endpoint.MessageEndpointFactory;
+import javax.transaction.Transaction;
+import javax.transaction.TransactionManager;
+import javax.transaction.xa.XAResource;
+import java.util.UUID;
+
+import org.apache.activemq.api.core.ActiveMQException;
+import org.apache.activemq.api.core.SimpleString;
+import org.apache.activemq.api.core.client.ClientMessage;
+import org.apache.activemq.api.core.client.ClientSession.QueueQuery;
+import org.apache.activemq.api.core.client.ClientSessionFactory;
+import org.apache.activemq.api.core.client.MessageHandler;
+import org.apache.activemq.core.client.impl.ClientConsumerInternal;
+import org.apache.activemq.core.client.impl.ClientSessionFactoryInternal;
+import org.apache.activemq.core.client.impl.ClientSessionInternal;
+import org.apache.activemq.jms.client.ActiveMQDestination;
+import org.apache.activemq.jms.client.ActiveMQMessage;
+import org.apache.activemq.ra.ActiveMQRALogger;
+import org.apache.activemq.ra.ActiveMQResourceAdapter;
+import org.apache.activemq.ra.ActiveMQXAResourceWrapper;
+import org.apache.activemq.utils.FutureLatch;
+
+/**
+ * The message handler
+ *
+ * @author <a href="adrian@jboss.com">Adrian Brock</a>
+ * @author <a href="mailto:jesper.pedersen@jboss.org">Jesper Pedersen</a>
+ * @author <a href="mailto:andy.taylor@jboss.org">Andy Taylor</a>
+ * @author <a href="mailto:mtaylor@redhat.com">Martyn Taylor</a>
+ */
+public class ActiveMQMessageHandler implements MessageHandler
+{
+ /**
+ * Trace enabled
+ */
+ private static boolean trace = ActiveMQRALogger.LOGGER.isTraceEnabled();
+ /**
+ * The session
+ */
+ private final ClientSessionInternal session;
+
+ private ClientConsumerInternal consumer;
+
+ /**
+ * The endpoint
+ */
+ private MessageEndpoint endpoint;
+
+ private final ActiveMQActivation activation;
+
+ private boolean useLocalTx;
+
+ private boolean transacted;
+
+ private boolean useXA = false;
+
+ private final int sessionNr;
+
+ private final TransactionManager tm;
+
+ private ClientSessionFactory cf;
+
+ public ActiveMQMessageHandler(final ActiveMQActivation activation,
+ final TransactionManager tm,
+ final ClientSessionInternal session,
+ final ClientSessionFactory cf,
+ final int sessionNr)
+ {
+ this.activation = activation;
+ this.session = session;
+ this.cf = cf;
+ this.sessionNr = sessionNr;
+ this.tm = tm;
+ }
+
+ public void setup() throws Exception
+ {
+ if (ActiveMQMessageHandler.trace)
+ {
+ ActiveMQRALogger.LOGGER.trace("setup()");
+ }
+
+ ActiveMQActivationSpec spec = activation.getActivationSpec();
+ String selector = spec.getMessageSelector();
+
+ // Create the message consumer
+ SimpleString selectorString = selector == null || selector.trim().equals("") ? null : new SimpleString(selector);
+ if (activation.isTopic() && spec.isSubscriptionDurable())
+ {
+ SimpleString queueName = new SimpleString(ActiveMQDestination.createQueueNameForDurableSubscription(true,
+ spec.getClientID(),
+ spec.getSubscriptionName()));
+
+ QueueQuery subResponse = session.queueQuery(queueName);
+
+ if (!subResponse.isExists())
+ {
+ session.createQueue(activation.getAddress(), queueName, selectorString, true);
+ }
+ else
+ {
+ // The check for already exists should be done only at the first session
+ // As a deployed MDB could set up multiple instances in order to process messages in parallel.
+ if (sessionNr == 0 && subResponse.getConsumerCount() > 0)
+ {
+ if (!spec.isShareSubscriptions())
+ {
+ throw new javax.jms.IllegalStateException("Cannot create a subscriber on the durable subscription since it already has subscriber(s)");
+ }
+ else if (ActiveMQRALogger.LOGGER.isDebugEnabled())
+ {
+ ActiveMQRALogger.LOGGER.debug("the mdb on destination " + queueName + " already had " +
+ subResponse.getConsumerCount() +
+ " consumers but the MDB is configured to share subscriptions, so no exceptions are thrown");
+ }
+ }
+
+ SimpleString oldFilterString = subResponse.getFilterString();
+
+ boolean selectorChanged = selector == null && oldFilterString != null ||
+ oldFilterString == null &&
+ selector != null ||
+ (oldFilterString != null && selector != null && !oldFilterString.toString()
+ .equals(selector));
+
+ SimpleString oldTopicName = subResponse.getAddress();
+
+ boolean topicChanged = !oldTopicName.equals(activation.getAddress());
+
+ if (selectorChanged || topicChanged)
+ {
+ // Delete the old durable sub
+ session.deleteQueue(queueName);
+
+ // Create the new one
+ session.createQueue(activation.getAddress(), queueName, selectorString, true);
+ }
+ }
+ consumer = (ClientConsumerInternal) session.createConsumer(queueName, null, false);
+ }
+ else
+ {
+ SimpleString tempQueueName;
+ if (activation.isTopic())
+ {
+ if (activation.getTopicTemporaryQueue() == null)
+ {
+ tempQueueName = new SimpleString(UUID.randomUUID().toString());
+ session.createTemporaryQueue(activation.getAddress(), tempQueueName, selectorString);
+ activation.setTopicTemporaryQueue(tempQueueName);
+ }
+ else
+ {
+ tempQueueName = activation.getTopicTemporaryQueue();
+ QueueQuery queueQuery = session.queueQuery(tempQueueName);
+ if (!queueQuery.isExists())
+ {
+ // this is because we could be using remote servers (in cluster maybe)
+ // and the queue wasn't created on that node yet.
+ session.createTemporaryQueue(activation.getAddress(), tempQueueName, selectorString);
+ }
+ }
+ }
+ else
+ {
+ tempQueueName = activation.getAddress();
+ }
+ consumer = (ClientConsumerInternal) session.createConsumer(tempQueueName, selectorString);
+ }
+
+ // Create the endpoint, if we are transacted pass the session so it is enlisted, unless using Local TX
+ MessageEndpointFactory endpointFactory = activation.getMessageEndpointFactory();
+ useLocalTx = !activation.isDeliveryTransacted() && activation.getActivationSpec().isUseLocalTx();
+ transacted = activation.isDeliveryTransacted();
+ if (activation.isDeliveryTransacted() && !activation.getActivationSpec().isUseLocalTx())
+ {
+ XAResource xaResource = new ActiveMQXAResourceWrapper(session,
+ ((ActiveMQResourceAdapter) spec.getResourceAdapter()).getJndiName(),
+ ((ClientSessionFactoryInternal) cf).getLiveNodeId());
+ endpoint = endpointFactory.createEndpoint(xaResource);
+ useXA = true;
+ }
+ else
+ {
+ endpoint = endpointFactory.createEndpoint(null);
+ useXA = false;
+ }
+ consumer.setMessageHandler(this);
+ }
+
+ XAResource getXAResource()
+ {
+ return useXA ? session : null;
+ }
+
+ public Thread interruptConsumer(FutureLatch future)
+ {
+ try
+ {
+ if (consumer != null)
+ {
+ return consumer.prepareForClose(future);
+ }
+ }
+ catch (Throwable e)
+ {
+ ActiveMQRALogger.LOGGER.warn("Error interrupting handler on endpoint " + endpoint + " handler=" + consumer);
+ }
+ return null;
+ }
+
+ /**
+ * Stop the handler
+ */
+ public void teardown()
+ {
+ if (ActiveMQMessageHandler.trace)
+ {
+ ActiveMQRALogger.LOGGER.trace("teardown()");
+ }
+
+ try
+ {
+ if (endpoint != null)
+ {
+ endpoint.release();
+ endpoint = null;
+ }
+ }
+ catch (Throwable t)
+ {
+ ActiveMQRALogger.LOGGER.debug("Error releasing endpoint " + endpoint, t);
+ }
+
+ try
+ {
+ consumer.close();
+ if (activation.getTopicTemporaryQueue() != null)
+ {
+ // We need to delete temporary topics when the activation is stopped or messages will build up on the server
+ SimpleString tmpQueue = activation.getTopicTemporaryQueue();
+ QueueQuery subResponse = session.queueQuery(tmpQueue);
+ if (subResponse.getConsumerCount() == 0)
+ {
+ // This is optional really, since we now use temporaryQueues, we could simply ignore this
+ // and the server temporary queue would remove this as soon as the queue was removed
+ session.deleteQueue(tmpQueue);
+ }
+ }
+ }
+ catch (Throwable t)
+ {
+ ActiveMQRALogger.LOGGER.debug("Error closing core-queue consumer", t);
+ }
+
+ try
+ {
+ if (session != null)
+ {
+ session.close();
+ }
+ }
+ catch (Throwable t)
+ {
+ ActiveMQRALogger.LOGGER.debug("Error releasing session " + session, t);
+ }
+
+ try
+ {
+ if (cf != null)
+ {
+ cf.close();
+ }
+ }
+ catch (Throwable t)
+ {
+ ActiveMQRALogger.LOGGER.debug("Error releasing session factory " + session, t);
+ }
+ }
+
+ public void onMessage(final ClientMessage message)
+ {
+ if (ActiveMQMessageHandler.trace)
+ {
+ ActiveMQRALogger.LOGGER.trace("onMessage(" + message + ")");
+ }
+
+ ActiveMQMessage msg = ActiveMQMessage.createMessage(message, session);
+ boolean beforeDelivery = false;
+
+ try
+ {
+ if (activation.getActivationSpec().getTransactionTimeout() > 0 && tm != null)
+ {
+ tm.setTransactionTimeout(activation.getActivationSpec().getTransactionTimeout());
+ }
+ endpoint.beforeDelivery(ActiveMQActivation.ONMESSAGE);
+ beforeDelivery = true;
+ msg.doBeforeReceive();
+
+ //In the transacted case the message must be acked *before* onMessage is called
+
+ if (transacted)
+ {
+ message.acknowledge();
+ }
+
+ ((MessageListener) endpoint).onMessage(msg);
+
+ if (!transacted)
+ {
+ message.acknowledge();
+ }
+
+ try
+ {
+ endpoint.afterDelivery();
+ }
+ catch (ResourceException e)
+ {
+ ActiveMQRALogger.LOGGER.unableToCallAfterDelivery(e);
+ return;
+ }
+ if (useLocalTx)
+ {
+ session.commit();
+ }
+
+ if (trace)
+ {
+ ActiveMQRALogger.LOGGER.trace("finished onMessage on " + message);
+ }
+ }
+ catch (Throwable e)
+ {
+ ActiveMQRALogger.LOGGER.errorDeliveringMessage(e);
+ // we need to call before/afterDelivery as a pair
+ if (beforeDelivery)
+ {
+ if (useXA && tm != null)
+ {
+ // This is the job for the container,
+ // however if the container throws an exception because of some other errors,
+ // there are situations where the container is not setting the rollback only
+ // this is to avoid a scenario where afterDelivery would kick in
+ try
+ {
+ Transaction tx = tm.getTransaction();
+ if (tx != null)
+ {
+ tx.setRollbackOnly();
+ }
+ }
+ catch (Exception e1)
+ {
+ ActiveMQRALogger.LOGGER.warn("unnable to clear the transaction", e1);
+ try
+ {
+ session.rollback();
+ }
+ catch (ActiveMQException e2)
+ {
+ ActiveMQRALogger.LOGGER.warn("Unable to rollback", e2);
+ return;
+ }
+ }
+ }
+
+ MessageEndpoint endToUse = endpoint;
+ try
+ {
+ // to avoid a NPE that would happen while the RA is in tearDown
+ if (endToUse != null)
+ {
+ endToUse.afterDelivery();
+ }
+ }
+ catch (ResourceException e1)
+ {
+ ActiveMQRALogger.LOGGER.unableToCallAfterDelivery(e1);
+ }
+ }
+ if (useLocalTx || !activation.isDeliveryTransacted())
+ {
+ try
+ {
+ session.rollback(true);
+ }
+ catch (ActiveMQException e1)
+ {
+ ActiveMQRALogger.LOGGER.unableToRollbackTX();
+ }
+ }
+ }
+ finally
+ {
+ try
+ {
+ session.resetIfNeeded();
+ }
+ catch (ActiveMQException e)
+ {
+ ActiveMQRALogger.LOGGER.unableToResetSession();
+ }
+ }
+
+ }
+
+ public void start() throws ActiveMQException
+ {
+ session.start();
+ }
+}
http://git-wip-us.apache.org/repos/asf/activemq-6/blob/034adfbf/activemq-ra/src/main/java/org/apache/activemq/ra/inflow/HornetQActivation.java
----------------------------------------------------------------------
diff --git a/activemq-ra/src/main/java/org/apache/activemq/ra/inflow/HornetQActivation.java b/activemq-ra/src/main/java/org/apache/activemq/ra/inflow/HornetQActivation.java
deleted file mode 100644
index 8bf12b6..0000000
--- a/activemq-ra/src/main/java/org/apache/activemq/ra/inflow/HornetQActivation.java
+++ /dev/null
@@ -1,831 +0,0 @@
-/*
- * Copyright 2005-2014 Red Hat, Inc.
- * Red Hat 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.activemq.ra.inflow;
-
-import javax.jms.Destination;
-import javax.jms.Message;
-import javax.jms.MessageListener;
-import javax.jms.Queue;
-import javax.jms.Topic;
-import javax.naming.Context;
-import javax.naming.InitialContext;
-import javax.resource.ResourceException;
-import javax.resource.spi.endpoint.MessageEndpointFactory;
-import javax.resource.spi.work.Work;
-import javax.resource.spi.work.WorkManager;
-import javax.transaction.xa.XAResource;
-import java.lang.reflect.Method;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.concurrent.atomic.AtomicBoolean;
-
-import org.apache.activemq.api.core.ActiveMQException;
-import org.apache.activemq.api.core.ActiveMQExceptionType;
-import org.apache.activemq.api.core.ActiveMQNonExistentQueueException;
-import org.apache.activemq.api.core.ActiveMQNotConnectedException;
-import org.apache.activemq.api.core.SimpleString;
-import org.apache.activemq.api.core.client.ClientSession;
-import org.apache.activemq.api.core.client.ClientSessionFactory;
-import org.apache.activemq.api.jms.HornetQJMSClient;
-import org.apache.activemq.core.client.impl.ClientSessionInternal;
-import org.apache.activemq.jms.client.HornetQConnectionFactory;
-import org.apache.activemq.jms.client.HornetQDestination;
-import org.apache.activemq.jms.server.recovery.XARecoveryConfig;
-import org.apache.activemq.ra.HornetQRABundle;
-import org.apache.activemq.ra.HornetQRAConnectionFactory;
-import org.apache.activemq.ra.HornetQRALogger;
-import org.apache.activemq.ra.HornetQRaUtils;
-import org.apache.activemq.ra.HornetQResourceAdapter;
-import org.apache.activemq.utils.FutureLatch;
-import org.apache.activemq.utils.SensitiveDataCodec;
-
-/**
- * The activation.
- *
- * @author <a href="adrian@jboss.com">Adrian Brock</a>
- * @author <a href="jesper.pedersen@jboss.org">Jesper Pedersen</a>
- * @author <a href="mailto:andy.taylor@jboss.org">Andy Taylor</a>
- */
-public class HornetQActivation
-{
- /**
- * Trace enabled
- */
- private static boolean trace = HornetQRALogger.LOGGER.isTraceEnabled();
-
- /**
- * The onMessage method
- */
- public static final Method ONMESSAGE;
-
- /**
- * The resource adapter
- */
- private final HornetQResourceAdapter ra;
-
- /**
- * The activation spec
- */
- private final HornetQActivationSpec spec;
-
- /**
- * The message endpoint factory
- */
- private final MessageEndpointFactory endpointFactory;
-
- /**
- * Whether delivery is active
- */
- private final AtomicBoolean deliveryActive = new AtomicBoolean(false);
-
- /**
- * The destination type
- */
- private boolean isTopic = false;
-
- /**
- * Is the delivery transacted
- */
- private boolean isDeliveryTransacted;
-
- private HornetQDestination destination;
-
- /**
- * The name of the temporary subscription name that all the sessions will share
- */
- private SimpleString topicTemporaryQueue;
-
- private final List<HornetQMessageHandler> handlers = new ArrayList<HornetQMessageHandler>();
-
- private HornetQConnectionFactory factory;
-
- // Whether we are in the failure recovery loop
- private final AtomicBoolean inFailure = new AtomicBoolean(false);
- private XARecoveryConfig resourceRecovery;
-
- static
- {
- try
- {
- ONMESSAGE = MessageListener.class.getMethod("onMessage", new Class[]{Message.class});
- }
- catch (Exception e)
- {
- throw new RuntimeException(e);
- }
- }
-
- /**
- * Constructor
- *
- * @param ra The resource adapter
- * @param endpointFactory The endpoint factory
- * @param spec The activation spec
- * @throws ResourceException Thrown if an error occurs
- */
- public HornetQActivation(final HornetQResourceAdapter ra,
- final MessageEndpointFactory endpointFactory,
- final HornetQActivationSpec spec) throws ResourceException
- {
- spec.validate();
-
- if (HornetQActivation.trace)
- {
- HornetQRALogger.LOGGER.trace("constructor(" + ra + ", " + endpointFactory + ", " + spec + ")");
- }
-
- if (ra.isUseMaskedPassword())
- {
- String pass = spec.getOwnPassword();
- if (pass != null)
- {
- SensitiveDataCodec<String> codec = ra.getCodecInstance();
-
- try
- {
- spec.setPassword(codec.decode(pass));
- }
- catch (Exception e)
- {
- throw new ResourceException(e);
- }
- }
- }
-
- this.ra = ra;
- this.endpointFactory = endpointFactory;
- this.spec = spec;
- try
- {
- isDeliveryTransacted = endpointFactory.isDeliveryTransacted(HornetQActivation.ONMESSAGE);
- }
- catch (Exception e)
- {
- throw new ResourceException(e);
- }
- }
-
- /**
- * Get the activation spec
- *
- * @return The value
- */
- public HornetQActivationSpec getActivationSpec()
- {
- if (HornetQActivation.trace)
- {
- HornetQRALogger.LOGGER.trace("getActivationSpec()");
- }
-
- return spec;
- }
-
- /**
- * Get the message endpoint factory
- *
- * @return The value
- */
- public MessageEndpointFactory getMessageEndpointFactory()
- {
- if (HornetQActivation.trace)
- {
- HornetQRALogger.LOGGER.trace("getMessageEndpointFactory()");
- }
-
- return endpointFactory;
- }
-
- /**
- * Get whether delivery is transacted
- *
- * @return The value
- */
- public boolean isDeliveryTransacted()
- {
- if (HornetQActivation.trace)
- {
- HornetQRALogger.LOGGER.trace("isDeliveryTransacted()");
- }
-
- return isDeliveryTransacted;
- }
-
- /**
- * Get the work manager
- *
- * @return The value
- */
- public WorkManager getWorkManager()
- {
- if (HornetQActivation.trace)
- {
- HornetQRALogger.LOGGER.trace("getWorkManager()");
- }
-
- return ra.getWorkManager();
- }
-
- /**
- * Is the destination a topic
- *
- * @return The value
- */
- public boolean isTopic()
- {
- if (HornetQActivation.trace)
- {
- HornetQRALogger.LOGGER.trace("isTopic()");
- }
-
- return isTopic;
- }
-
- /**
- * Start the activation
- *
- * @throws ResourceException Thrown if an error occurs
- */
- public void start() throws ResourceException
- {
- if (HornetQActivation.trace)
- {
- HornetQRALogger.LOGGER.trace("start()");
- }
- deliveryActive.set(true);
- ra.getWorkManager().scheduleWork(new SetupActivation());
- }
-
- /**
- * @return the topicTemporaryQueue
- */
- public SimpleString getTopicTemporaryQueue()
- {
- return topicTemporaryQueue;
- }
-
- /**
- * @param topicTemporaryQueue the topicTemporaryQueue to set
- */
- public void setTopicTemporaryQueue(SimpleString topicTemporaryQueue)
- {
- this.topicTemporaryQueue = topicTemporaryQueue;
- }
-
- /**
- * @return the list of XAResources for this activation endpoint
- */
- public List<XAResource> getXAResources()
- {
- List<XAResource> xaresources = new ArrayList<XAResource>();
- for (HornetQMessageHandler handler : handlers)
- {
- XAResource xares = handler.getXAResource();
- if (xares != null)
- {
- xaresources.add(xares);
- }
- }
- return xaresources;
- }
-
- /**
- * Stop the activation
- */
- public void stop()
- {
- if (HornetQActivation.trace)
- {
- HornetQRALogger.LOGGER.trace("stop()");
- }
-
- deliveryActive.set(false);
- teardown();
- }
-
- /**
- * Setup the activation
- *
- * @throws Exception Thrown if an error occurs
- */
- protected synchronized void setup() throws Exception
- {
- HornetQRALogger.LOGGER.debug("Setting up " + spec);
-
- setupCF();
-
- setupDestination();
-
- Exception firstException = null;
-
- for (int i = 0; i < spec.getMaxSession(); i++)
- {
- ClientSessionFactory cf = null;
- ClientSession session = null;
-
- try
- {
- cf = factory.getServerLocator().createSessionFactory();
- session = setupSession(cf);
- HornetQMessageHandler handler = new HornetQMessageHandler(this, ra.getTM(), (ClientSessionInternal) session, cf, i);
- handler.setup();
- handlers.add(handler);
- }
- catch (Exception e)
- {
- if (cf != null)
- {
- cf.close();
- }
- if (session != null)
- {
- session.close();
- }
- if (firstException == null)
- {
- firstException = e;
- }
- }
- }
- //if we have any exceptions close all the handlers and throw the first exception.
- //we don't want partially configured activations, i.e. only 8 out of 15 sessions started so best to stop and log the error.
- if (firstException != null)
- {
- for (HornetQMessageHandler handler : handlers)
- {
- handler.teardown();
- }
- throw firstException;
- }
-
- //now start them all together.
- for (HornetQMessageHandler handler : handlers)
- {
- handler.start();
- }
-
- resourceRecovery = ra.getRecoveryManager().register(factory, spec.getUser(), spec.getPassword());
-
- HornetQRALogger.LOGGER.debug("Setup complete " + this);
- }
-
- /**
- * Teardown the activation
- */
- protected synchronized void teardown()
- {
- HornetQRALogger.LOGGER.debug("Tearing down " + spec);
-
- if (resourceRecovery != null)
- {
- ra.getRecoveryManager().unRegister(resourceRecovery);
- }
-
- final HornetQMessageHandler[] handlersCopy = new HornetQMessageHandler[handlers.size()];
-
- // We need to do from last to first as any temporary queue will have been created on the first handler
- // So we invert the handlers here
- for (int i = 0; i < handlers.size(); i++)
- {
- // The index here is the complimentary so it's inverting the array
- handlersCopy[i] = handlers.get(handlers.size() - i - 1);
- }
-
- handlers.clear();
-
- FutureLatch future = new FutureLatch(handlersCopy.length);
- List<Thread> interruptThreads = new ArrayList<>();
- for (HornetQMessageHandler handler : handlersCopy)
- {
- Thread thread = handler.interruptConsumer(future);
- if (thread != null)
- {
- interruptThreads.add(thread);
- }
- }
-
- //wait for all the consumers to complete any onmessage calls
- boolean stuckThreads = !future.await(factory.getCallTimeout());
- //if any are stuck then we need to interrupt them
- if (stuckThreads)
- {
- for (Thread interruptThread : interruptThreads)
- {
- try
- {
- interruptThread.interrupt();
- }
- catch (Exception e)
- {
- //ok
- }
- }
- }
-
- Thread threadTearDown = new Thread("TearDown/HornetQActivation")
- {
- public void run()
- {
- for (HornetQMessageHandler handler : handlersCopy)
- {
- handler.teardown();
- }
- }
- };
-
- // We will first start a new thread that will call tearDown on all the instances, trying to graciously shutdown everything.
- // We will then use the call-timeout to determine a timeout.
- // if that failed we will then close the connection factory, and interrupt the thread
- threadTearDown.start();
-
- try
- {
- threadTearDown.join(factory.getCallTimeout());
- }
- catch (InterruptedException e)
- {
- // nothing to be done on this context.. we will just keep going as we need to send an interrupt to threadTearDown and give up
- }
-
- if (threadTearDown.isAlive())
- {
- if (factory != null)
- {
- // This will interrupt any threads waiting on reconnect
- factory.close();
- factory = null;
- }
- threadTearDown.interrupt();
-
- try
- {
- threadTearDown.join(5000);
- }
- catch (InterruptedException e)
- {
- // nothing to be done here.. we are going down anyways
- }
-
- if (threadTearDown.isAlive())
- {
- HornetQRALogger.LOGGER.warn("Thread " + threadTearDown + " couldn't be finished");
- }
- }
-
- if (spec.isHasBeenUpdated() && factory != null)
- {
- factory.close();
- factory = null;
- }
-
-
- HornetQRALogger.LOGGER.debug("Tearing down complete " + this);
- }
-
- protected void setupCF() throws Exception
- {
- if (spec.getConnectionFactoryLookup() != null)
- {
- Context ctx;
- if (spec.getParsedJndiParams() == null)
- {
- ctx = new InitialContext();
- }
- else
- {
- ctx = new InitialContext(spec.getParsedJndiParams());
- }
- Object fac = ctx.lookup(spec.getConnectionFactoryLookup());
- if (fac instanceof HornetQConnectionFactory)
- {
- factory = (HornetQConnectionFactory) fac;
- }
- else
- {
- HornetQRAConnectionFactory raFact = (HornetQRAConnectionFactory) fac;
- if (spec.isHasBeenUpdated())
- {
- factory = raFact.getResourceAdapter().createHornetQConnectionFactory(spec);
- }
- else
- {
- factory = raFact.getDefaultFactory();
- if (factory != ra.getDefaultHornetQConnectionFactory())
- {
- HornetQRALogger.LOGGER.warnDifferentConnectionfactory();
- }
- }
- }
- }
- else if (spec.isHasBeenUpdated())
- {
- factory = ra.createHornetQConnectionFactory(spec);
- }
- else
- {
- factory = ra.getDefaultHornetQConnectionFactory();
- }
- }
-
- /**
- * Setup a session
- *
- * @param cf
- * @return The connection
- * @throws Exception Thrown if an error occurs
- */
- protected ClientSession setupSession(ClientSessionFactory cf) throws Exception
- {
- ClientSession result = null;
-
- try
- {
- result = ra.createSession(cf,
- spec.getAcknowledgeModeInt(),
- spec.getUser(),
- spec.getPassword(),
- ra.getPreAcknowledge(),
- ra.getDupsOKBatchSize(),
- ra.getTransactionBatchSize(),
- isDeliveryTransacted,
- spec.isUseLocalTx(),
- spec.getTransactionTimeout());
-
- result.addMetaData("resource-adapter", "inbound");
- result.addMetaData("jms-session", "");
- String clientID = ra.getClientID() == null ? spec.getClientID() : ra.getClientID();
- if (clientID != null)
- {
- result.addMetaData("jms-client-id", clientID);
- }
-
- HornetQRALogger.LOGGER.debug("Using queue connection " + result);
-
- return result;
- }
- catch (Throwable t)
- {
- try
- {
- if (result != null)
- {
- result.close();
- }
- }
- catch (Exception e)
- {
- HornetQRALogger.LOGGER.trace("Ignored error closing connection", e);
- }
- if (t instanceof Exception)
- {
- throw (Exception) t;
- }
- throw new RuntimeException("Error configuring connection", t);
- }
- }
-
- public SimpleString getAddress()
- {
- return destination.getSimpleAddress();
- }
-
- protected void setupDestination() throws Exception
- {
-
- String destinationName = spec.getDestination();
-
- if (spec.isUseJNDI())
- {
- Context ctx;
- if (spec.getParsedJndiParams() == null)
- {
- ctx = new InitialContext();
- }
- else
- {
- ctx = new InitialContext(spec.getParsedJndiParams());
- }
- HornetQRALogger.LOGGER.debug("Using context " + ctx.getEnvironment() + " for " + spec);
- if (HornetQActivation.trace)
- {
- HornetQRALogger.LOGGER.trace("setupDestination(" + ctx + ")");
- }
-
- String destinationTypeString = spec.getDestinationType();
- if (destinationTypeString != null && !destinationTypeString.trim().equals(""))
- {
- HornetQRALogger.LOGGER.debug("Destination type defined as " + destinationTypeString);
-
- Class<?> destinationType;
- if (Topic.class.getName().equals(destinationTypeString))
- {
- destinationType = Topic.class;
- isTopic = true;
- }
- else
- {
- destinationType = Queue.class;
- }
-
- HornetQRALogger.LOGGER.debug("Retrieving " + destinationType.getName() + " \"" + destinationName + "\" from JNDI");
-
- try
- {
- destination = (HornetQDestination) HornetQRaUtils.lookup(ctx, destinationName, destinationType);
- }
- catch (Exception e)
- {
- if (destinationName == null)
- {
- throw HornetQRABundle.BUNDLE.noDestinationName();
- }
-
- String calculatedDestinationName = destinationName.substring(destinationName.lastIndexOf('/') + 1);
-
- HornetQRALogger.LOGGER.debug("Unable to retrieve " + destinationName +
- " from JNDI. Creating a new " + destinationType.getName() +
- " named " + calculatedDestinationName + " to be used by the MDB.");
-
- // If there is no binding on naming, we will just create a new instance
- if (isTopic)
- {
- destination = (HornetQDestination) HornetQJMSClient.createTopic(calculatedDestinationName);
- }
- else
- {
- destination = (HornetQDestination) HornetQJMSClient.createQueue(calculatedDestinationName);
- }
- }
- }
- else
- {
- HornetQRALogger.LOGGER.debug("Destination type not defined in MDB activation configuration.");
- HornetQRALogger.LOGGER.debug("Retrieving " + Destination.class.getName() + " \"" + destinationName + "\" from JNDI");
-
- destination = (HornetQDestination) HornetQRaUtils.lookup(ctx, destinationName, Destination.class);
- if (destination instanceof Topic)
- {
- isTopic = true;
- }
- }
- }
- else
- {
- HornetQRALogger.LOGGER.instantiatingDestination(spec.getDestinationType(), spec.getDestination());
-
- if (Topic.class.getName().equals(spec.getDestinationType()))
- {
- destination = (HornetQDestination) HornetQJMSClient.createTopic(spec.getDestination());
- isTopic = true;
- }
- else
- {
- destination = (HornetQDestination) HornetQJMSClient.createQueue(spec.getDestination());
- }
- }
- }
-
- /**
- * Get a string representation
- *
- * @return The value
- */
- @Override
- public String toString()
- {
- StringBuffer buffer = new StringBuffer();
- buffer.append(HornetQActivation.class.getName()).append('(');
- buffer.append("spec=").append(spec.getClass().getName());
- buffer.append(" mepf=").append(endpointFactory.getClass().getName());
- buffer.append(" active=").append(deliveryActive.get());
- if (spec.getDestination() != null)
- {
- buffer.append(" destination=").append(spec.getDestination());
- }
- buffer.append(" transacted=").append(isDeliveryTransacted);
- buffer.append(')');
- return buffer.toString();
- }
-
- /**
- * Handles any failure by trying to reconnect
- *
- * @param failure the reason for the failure
- */
- public void handleFailure(Throwable failure)
- {
- if (failure instanceof ActiveMQException && ((ActiveMQException) failure).getType() == ActiveMQExceptionType.QUEUE_DOES_NOT_EXIST)
- {
- HornetQRALogger.LOGGER.awaitingTopicQueueCreation(getActivationSpec().getDestination());
- }
- else if (failure instanceof ActiveMQException && ((ActiveMQException) failure).getType() == ActiveMQExceptionType.NOT_CONNECTED)
- {
- HornetQRALogger.LOGGER.awaitingJMSServerCreation();
- }
- else
- {
- HornetQRALogger.LOGGER.failureInActivation(failure, spec);
- }
- int reconnectCount = 0;
- int setupAttempts = spec.getSetupAttempts();
- long setupInterval = spec.getSetupInterval();
-
- // Only enter the failure loop once
- if (inFailure.getAndSet(true))
- return;
- try
- {
- Throwable lastException = failure;
- while (deliveryActive.get() && (setupAttempts == -1 || reconnectCount < setupAttempts))
- {
- teardown();
-
- try
- {
- Thread.sleep(setupInterval);
- }
- catch (InterruptedException e)
- {
- HornetQRALogger.LOGGER.debug("Interrupted trying to reconnect " + spec, e);
- break;
- }
-
- if (reconnectCount < 1)
- {
- HornetQRALogger.LOGGER.attemptingReconnect(spec);
- }
- try
- {
- setup();
- HornetQRALogger.LOGGER.reconnected();
- break;
- }
- catch (Throwable t)
- {
- if (failure instanceof ActiveMQException && ((ActiveMQException) failure).getType() == ActiveMQExceptionType.QUEUE_DOES_NOT_EXIST)
- {
- if (lastException == null || !(t instanceof ActiveMQNonExistentQueueException))
- {
- lastException = t;
- HornetQRALogger.LOGGER.awaitingTopicQueueCreation(getActivationSpec().getDestination());
- }
- }
- else if (failure instanceof ActiveMQException && ((ActiveMQException) failure).getType() == ActiveMQExceptionType.NOT_CONNECTED)
- {
- if (lastException == null || !(t instanceof ActiveMQNotConnectedException))
- {
- lastException = t;
- HornetQRALogger.LOGGER.awaitingJMSServerCreation();
- }
- }
- else
- {
- HornetQRALogger.LOGGER.errorReconnecting(t, spec);
- }
- }
- ++reconnectCount;
- }
- }
- finally
- {
- // Leaving failure recovery loop
- inFailure.set(false);
- }
- }
-
- public HornetQConnectionFactory getConnectionFactory()
- {
- return this.factory;
- }
-
- /**
- * Handles the setup
- */
- private class SetupActivation implements Work
- {
- public void run()
- {
- try
- {
- setup();
- }
- catch (Throwable t)
- {
- handleFailure(t);
- }
- }
-
- public void release()
- {
- }
- }
-}