Index: modules/jms/pom.xml
===================================================================
--- modules/jms/pom.xml (revision 712645)
+++ modules/jms/pom.xml (working copy)
@@ -71,11 +71,23 @@
${axis2-transport-base.version}
+
+ org.apache.geronimo.specs
+ geronimo-jta_1.1_spec
+ 1.1.1
+
+
org.apache.geronimo.specs
geronimo-jms_1.1_spec
${jms-1.1-spec.version}
+
+
+ org.apache.geronimo.specs
+ geronimo-jms_1.1_spec
+ ${jms-1.1-spec.version}
+
junit
Index: modules/jms/src/main/java/org/apache/axis2/transport/jms/JMSConnectionFactory.java
===================================================================
--- modules/jms/src/main/java/org/apache/axis2/transport/jms/JMSConnectionFactory.java (revision 712645)
+++ modules/jms/src/main/java/org/apache/axis2/transport/jms/JMSConnectionFactory.java (working copy)
@@ -30,6 +30,7 @@
import java.util.HashMap;
import java.util.Hashtable;
import java.util.Map;
+import javax.transaction.UserTransaction;
/**
* Encapsulate a JMS Connection factory definition within an Axis2.xml
@@ -279,6 +280,7 @@
}
try {
+ String userTransactionJNDIName = jndiProperties.get(JMSConstants.USER_TANSACTION_JNDI_NAME_PARAM);
session = JMSUtils.createSession(connection, false, Session.AUTO_ACKNOWLEDGE, destinationType);
Destination destination = null;
@@ -291,8 +293,22 @@
}
MessageConsumer consumer = JMSUtils.createConsumer(session, destination);
- consumer.setMessageListener(new JMSMessageReceiver(jmsListener, this, workerPool,
- cfgCtx, endpoint));
+ if (userTransactionJNDIName == null) {
+ consumer.setMessageListener(new JMSMessageReceiver(jmsListener, this, workerPool,
+ cfgCtx, endpoint, null, consumer, false));
+ } else {
+ String autoCommit = jndiProperties.get(JMSConstants.AUTO_COMMIT_JNDI_NAME_PARAM);
+ UserTransaction userTransaction = null;
+ try {
+ userTransaction = JMSUtils.lookup(getContext(), UserTransaction.class, userTransactionJNDIName);
+ } catch (NamingException e) {
+ handleException("Can not lookup user transaction from jndi ", e);
+ }
+ JMSMessageReceiver jMSMessageReceiver = new JMSMessageReceiver(jmsListener, this, workerPool, cfgCtx,
+ endpoint, userTransaction, consumer, Boolean.parseBoolean(autoCommit));
+ jMSMessageReceiver.startReceiveTransact();
+
+ }
jmsSessions.put(destinationJNDIname, session);
// catches NameNotFound and JMSExceptions and marks service as faulty
Index: modules/jms/src/main/java/org/apache/axis2/transport/jms/JMSConstants.java
===================================================================
--- modules/jms/src/main/java/org/apache/axis2/transport/jms/JMSConstants.java (revision 712645)
+++ modules/jms/src/main/java/org/apache/axis2/transport/jms/JMSConstants.java (working copy)
@@ -84,6 +84,14 @@
/**
* The Parameter name indicating the JMS connection factory JNDI name
*/
+ public static final String USER_TANSACTION_JNDI_NAME_PARAM = "transport.jms.UserTransactionJNDIName";
+ /**
+ * The Parameter name indicating to auto commit active transactions at the end of mediation
+ */
+ public static final String AUTO_COMMIT_JNDI_NAME_PARAM = "transport.jms.AutoCommit";
+ /**
+ * The Parameter name indicating the JMS connection factory JNDI name
+ */
public static final String CONFAC_JNDI_NAME_PARAM = "transport.jms.ConnectionFactoryJNDIName";
/**
* The parameter indicating the expected content type for messages received by the service.
@@ -147,4 +155,13 @@
* A MessageContext property or client Option stating the JMS type
*/
public static final String JMS_TYPE = "JMS_TYPE";
+ /**
+ * Default transaction timeout in seconds.
+ */
+ public static final int DEFAULT_TRANSACTION_TIMEOUT = 30;
+ /**
+ * A MessageContext property stating the UserTransaction.
+ */
+ public static final String USER_TRANSACION = "USER_TRANSACION";
+
}
Index: modules/jms/src/main/java/org/apache/axis2/transport/jms/JMSMessageReceiver.java
===================================================================
--- modules/jms/src/main/java/org/apache/axis2/transport/jms/JMSMessageReceiver.java (revision 712645)
+++ modules/jms/src/main/java/org/apache/axis2/transport/jms/JMSMessageReceiver.java (working copy)
@@ -32,6 +32,10 @@
import javax.jms.*;
import javax.xml.namespace.QName;
+import javax.transaction.UserTransaction;
+import javax.transaction.NotSupportedException;
+import javax.transaction.Status;
+import javax.naming.NamingException;
/**
* This is the actual receiver which listens for and accepts JMS messages, and
@@ -55,6 +59,12 @@
final JMSEndpoint endpoint;
/** Metrics collector */
private MetricsCollector metrics = null;
+ /** User Transaction */
+ private UserTransaction userTransaction = null;
+ /** The JMS message consumer */
+ private MessageConsumer consumer =null;
+ /** Auto commit active transactions at the end of mediation */
+ private boolean autoCommit = false;
/**
* Create a new JMSMessage receiver
@@ -63,17 +73,26 @@
* @param jmsConFac the JMS connection factory we are associated with
* @param workerPool the worker thread pool to be used
* @param cfgCtx the axis ConfigurationContext
- * @param serviceName the name of the Axis service
+ * @param endpoint the axis Endpoint
*/
JMSMessageReceiver(JMSListener jmsListener, JMSConnectionFactory jmsConFac,
- WorkerPool workerPool, ConfigurationContext cfgCtx, JMSEndpoint endpoint) {
+ WorkerPool workerPool, ConfigurationContext cfgCtx, JMSEndpoint endpoint,
+ UserTransaction userTransaction, MessageConsumer consumer, boolean autoCommit) {
this.jmsListener = jmsListener;
this.jmsConnectionFactory = jmsConFac;
this.workerPool = workerPool;
this.cfgCtx = cfgCtx;
this.endpoint = endpoint;
this.metrics = jmsListener.getMetricsCollector();
+ this.userTransaction = userTransaction;
+ this.consumer = consumer;
+ this.autoCommit = autoCommit;
}
+
+
+ public void startReceiveTransact( ){
+ workerPool.execute(new Worker(null,this));
+ }
/**
* The entry point on the reception of each JMS message
@@ -118,7 +137,7 @@
}
} catch (JMSException ignore) {}
- workerPool.execute(new Worker(message));
+ workerPool.execute(new Worker(message,this));
}
private void handleException(String msg, Exception e) {
@@ -139,13 +158,43 @@
class Worker implements Runnable {
private Message message = null;
+ private JMSMessageReceiver messageReceiver = null;
- Worker(Message message) {
+ Worker(Message message, JMSMessageReceiver messageReceiver) {
this.message = message;
+ this.messageReceiver = messageReceiver;
}
public void run() {
+ //recieve message in a jta trasnaction
+ if (message == null && messageReceiver.userTransaction != null) {
+ try {
+ if (userTransaction.getStatus() != Status.STATUS_NO_TRANSACTION) {
+ log.debug("Incomplete previous user transaction on a thread");
+ try {
+ userTransaction.rollback();
+ } catch (Exception ignore) {
+ }
+ //Thread.sleep(JMSConstants.DEFAULT_TRANSACTION_TIMEOUT * 1000);
+ }
+ messageReceiver.userTransaction.setTransactionTimeout(JMSConstants.DEFAULT_TRANSACTION_TIMEOUT);
+ messageReceiver.userTransaction.begin();
+ message = messageReceiver.consumer.receive(JMSConstants.DEFAULT_TRANSACTION_TIMEOUT * 1000 / 2);
+ } catch (Exception e) {
+ log.debug("Exception while receiving a message within a user transaction", e);
+ }
+
+ if (message == null) {
+ try {
+ userTransaction.rollback();
+ } catch (Exception ignore) {
+ }
+ workerPool.execute(new Worker(null, messageReceiver));
+ return;
+ }
+ }
+
MessageContext msgContext = jmsListener.createMessageContext();
// set the JMS Message ID as the Message ID of the MessageContext
@@ -153,6 +202,8 @@
msgContext.setMessageID(message.getJMSMessageID());
msgContext.setProperty(JMSConstants.JMS_COORELATION_ID, message.getJMSMessageID());
} catch (JMSException ignore) {}
+
+ msgContext.setProperty(JMSConstants.USER_TRANSACION, messageReceiver.userTransaction);
AxisService service = null;
try {
@@ -215,6 +266,16 @@
jmsListener.error(service, e);
log.error("Exception while processing incoming message", e);
}
+
+ if (messageReceiver.userTransaction != null) {
+ try {
+ if (userTransaction.getStatus() == Status.STATUS_ACTIVE && messageReceiver.autoCommit)
+ userTransaction.commit();
+ } catch (Exception e) {
+ log.error("Exception committing a user transaction", e);
+ }
+ workerPool.execute(new Worker(null, messageReceiver));
+ }
}
}
}