You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by ta...@apache.org on 2015/01/27 19:21:17 UTC
qpid-jms git commit: Some initial work on refactoring TX handling in
the JMS Session to deal with TX and failover.
Repository: qpid-jms
Updated Branches:
refs/heads/master e67c6ecda -> 41a99b37c
Some initial work on refactoring TX handling in the JMS Session to deal
with TX and failover.
Project: http://git-wip-us.apache.org/repos/asf/qpid-jms/repo
Commit: http://git-wip-us.apache.org/repos/asf/qpid-jms/commit/41a99b37
Tree: http://git-wip-us.apache.org/repos/asf/qpid-jms/tree/41a99b37
Diff: http://git-wip-us.apache.org/repos/asf/qpid-jms/diff/41a99b37
Branch: refs/heads/master
Commit: 41a99b37cf22db667bd88ce2b692b1bd8feff3de
Parents: e67c6ec
Author: Timothy Bish <ta...@gmail.com>
Authored: Tue Jan 27 13:19:30 2015 -0500
Committer: Timothy Bish <ta...@gmail.com>
Committed: Tue Jan 27 13:20:12 2015 -0500
----------------------------------------------------------------------
.../qpid/jms/JmsLocalTransactionContext.java | 103 ++++++++------
.../org/apache/qpid/jms/JmsMessageConsumer.java | 8 +-
.../qpid/jms/JmsNoTxTransactionContext.java | 87 ++++++++++++
.../java/org/apache/qpid/jms/JmsSession.java | 110 +++++++--------
.../apache/qpid/jms/JmsTransactionContext.java | 137 +++++++++++++++++++
5 files changed, 341 insertions(+), 104 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/qpid-jms/blob/41a99b37/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsLocalTransactionContext.java
----------------------------------------------------------------------
diff --git a/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsLocalTransactionContext.java b/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsLocalTransactionContext.java
index f0473d3..0582c7c 100644
--- a/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsLocalTransactionContext.java
+++ b/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsLocalTransactionContext.java
@@ -20,17 +20,21 @@ import java.util.ArrayList;
import java.util.List;
import javax.jms.JMSException;
+import javax.jms.TransactionRolledBackException;
import org.apache.qpid.jms.exceptions.JmsExceptionSupport;
+import org.apache.qpid.jms.message.JmsInboundMessageDispatch;
+import org.apache.qpid.jms.message.JmsOutboundMessageDispatch;
import org.apache.qpid.jms.meta.JmsTransactionId;
import org.apache.qpid.jms.meta.JmsTransactionInfo;
+import org.apache.qpid.jms.provider.ProviderConstants.ACK_TYPE;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* Manages the details of a Session operating inside of a local JMS transaction.
*/
-public class JmsLocalTransactionContext {
+public class JmsLocalTransactionContext implements JmsTransactionContext {
private static final Logger LOG = LoggerFactory.getLogger(JmsLocalTransactionContext.class);
@@ -38,6 +42,7 @@ public class JmsLocalTransactionContext {
private final JmsSession session;
private final JmsConnection connection;
private JmsTransactionId transactionId;
+ private boolean failed;
private JmsTransactionListener listener;
public JmsLocalTransactionContext(JmsSession session) {
@@ -45,12 +50,34 @@ public class JmsLocalTransactionContext {
this.connection = session.getConnection();
}
- /**
- * Adds the given Transaction synchronization to the current list.
- *
- * @param s
- * the transaction synchronization to add.
- */
+ @Override
+ public void send(JmsConnection connection, JmsOutboundMessageDispatch envelope) throws JMSException {
+ // TODO - Optional throw an exception here to give early warning that
+ // the transaction is in a failed state and must be rolled back.
+ if (!isFailed()) {
+ begin();
+ connection.send(envelope);
+ }
+ }
+
+ @Override
+ public void acknowledge(JmsConnection connection, JmsInboundMessageDispatch envelope, ACK_TYPE ackType) throws JMSException {
+ // TODO - Should we ACK as delivered if failed just to ensure that prefetch is
+ // extended and new message arrive until commit or rollback is called.
+ // A quiet consumer could be misleading and prevent the code from doing
+ // its normal batched receive / commit.
+ if (!isFailed()) {
+ // Consumed or delivered messages fall into a transaction so we must check
+ // that there is an active one and start one if not.
+ if (ackType == ACK_TYPE.CONSUMED || ackType == ACK_TYPE.DELIVERED) {
+ begin();
+ }
+
+ connection.acknowledge(envelope, ackType);
+ }
+ }
+
+ @Override
public void addSynchronization(JmsTxSynchronization s) {
if (synchronizations == null) {
synchronizations = new ArrayList<JmsTxSynchronization>(10);
@@ -58,24 +85,23 @@ public class JmsLocalTransactionContext {
synchronizations.add(s);
}
- /**
- * Clears the current Transacted state. This is usually done when the client
- * detects that a failover has occurred and needs to create a new Transaction
- * for a Session that was previously enlisted in a transaction.
- */
- public void clear() {
- this.transactionId = null;
- this.synchronizations = null;
+ @Override
+ public void markAsFailed() {
+ if (isInTransaction()) {
+ failed = true;
+ }
}
- /**
- * Start a local transaction.
- *
- * @throws javax.jms.JMSException on internal error
- */
+ @Override
+ public boolean isFailed() {
+ return failed;
+ }
+
+ @Override
public void begin() throws JMSException {
- if (transactionId == null) {
+ if (!isInTransaction()) {
synchronizations = null;
+ failed = false;
transactionId = connection.getNextTransactionId();
JmsTransactionInfo transaction = new JmsTransactionInfo(session.getSessionId(), transactionId);
@@ -89,18 +115,13 @@ public class JmsLocalTransactionContext {
}
}
- /**
- * Rolls back any work done in this transaction and releases any locks
- * currently held.
- *
- * @throws JMSException
- * if the JMS provider fails to roll back the transaction due to some internal error.
- */
+ @Override
public void rollback() throws JMSException {
- if (transactionId != null) {
+ if (isInTransaction()) {
LOG.debug("Rollback: {} syncCount: {}", transactionId,
(synchronizations != null ? synchronizations.size() : 0));
+ failed = false;
transactionId = null;
connection.rollback(session.getSessionId());
@@ -112,15 +133,13 @@ public class JmsLocalTransactionContext {
afterRollback();
}
- /**
- * Commits all work done in this transaction and releases any locks
- * currently held.
- *
- * @throws JMSException
- * if the JMS provider fails to roll back the transaction due to some internal error.
- */
+ @Override
public void commit() throws JMSException {
- if (transactionId != null) {
+ if (isFailed()) {
+ throw new TransactionRolledBackException("Transaction failed and must be rolled back.");
+ }
+
+ if (isInTransaction()) {
LOG.debug("Commit: {} syncCount: {}", transactionId,
(synchronizations != null ? synchronizations.size() : 0));
@@ -145,25 +164,29 @@ public class JmsLocalTransactionContext {
@Override
public String toString() {
- return "JmsLocalTransactionContext{transactionId=" + transactionId + "}";
+ return "JmsLocalTransactionContext{ transactionId=" + transactionId + " }";
}
//------------- Getters and Setters --------------------------------------//
+ @Override
public JmsTransactionId getTransactionId() {
- return this.transactionId;
+ return transactionId;
}
+ @Override
public JmsTransactionListener getListener() {
return listener;
}
+ @Override
public void setListener(JmsTransactionListener listener) {
this.listener = listener;
}
+ @Override
public boolean isInTransaction() {
- return this.transactionId != null;
+ return transactionId != null;
}
//------------- Implementation methods -----------------------------------//
http://git-wip-us.apache.org/repos/asf/qpid-jms/blob/41a99b37/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsMessageConsumer.java
----------------------------------------------------------------------
diff --git a/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsMessageConsumer.java b/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsMessageConsumer.java
index 9b358a2..0547440 100644
--- a/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsMessageConsumer.java
+++ b/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsMessageConsumer.java
@@ -272,7 +272,7 @@ public class JmsMessageConsumer implements MessageConsumer, JmsMessageAvailableC
private JmsInboundMessageDispatch doAckConsumed(final JmsInboundMessageDispatch envelope) throws JMSException {
checkClosed();
try {
- session.acknowledge(envelope, ACK_TYPE.CONSUMED, true);
+ session.acknowledge(envelope, ACK_TYPE.CONSUMED);
} catch (JMSException ex) {
session.onException(ex);
throw ex;
@@ -282,7 +282,7 @@ public class JmsMessageConsumer implements MessageConsumer, JmsMessageAvailableC
private JmsInboundMessageDispatch doAckDelivered(final JmsInboundMessageDispatch envelope) throws JMSException {
try {
- session.acknowledge(envelope, ACK_TYPE.DELIVERED, true);
+ session.acknowledge(envelope, ACK_TYPE.DELIVERED);
} catch (JMSException ex) {
session.onException(ex);
throw ex;
@@ -292,7 +292,7 @@ public class JmsMessageConsumer implements MessageConsumer, JmsMessageAvailableC
private void doAckReleased(final JmsInboundMessageDispatch envelope) throws JMSException {
try {
- session.acknowledge(envelope, ACK_TYPE.RELEASED, false);
+ session.acknowledge(envelope, ACK_TYPE.RELEASED);
} catch (JMSException ex) {
session.onException(ex);
throw ex;
@@ -555,7 +555,7 @@ public class JmsMessageConsumer implements MessageConsumer, JmsMessageAvailableC
try {
JmsMessage copy = null;
boolean autoAckOrDupsOk = acknowledgementMode == Session.AUTO_ACKNOWLEDGE ||
- acknowledgementMode == Session.DUPS_OK_ACKNOWLEDGE;
+ acknowledgementMode == Session.DUPS_OK_ACKNOWLEDGE;
if (autoAckOrDupsOk) {
copy = copy(doAckDelivered(envelope));
} else {
http://git-wip-us.apache.org/repos/asf/qpid-jms/blob/41a99b37/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsNoTxTransactionContext.java
----------------------------------------------------------------------
diff --git a/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsNoTxTransactionContext.java b/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsNoTxTransactionContext.java
new file mode 100644
index 0000000..97f6df7
--- /dev/null
+++ b/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsNoTxTransactionContext.java
@@ -0,0 +1,87 @@
+/**
+ * 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.qpid.jms;
+
+import javax.jms.JMSException;
+
+import org.apache.qpid.jms.message.JmsInboundMessageDispatch;
+import org.apache.qpid.jms.message.JmsOutboundMessageDispatch;
+import org.apache.qpid.jms.meta.JmsTransactionId;
+import org.apache.qpid.jms.provider.ProviderConstants.ACK_TYPE;
+
+/**
+ * Used in non-transacted JMS Sessions to throw proper errors indicating
+ * that the Session is not transacted and cannot be treated as such.
+ */
+public class JmsNoTxTransactionContext implements JmsTransactionContext {
+
+ @Override
+ public void send(JmsConnection connection, JmsOutboundMessageDispatch envelope) throws JMSException {
+ connection.send(envelope);
+ }
+
+ @Override
+ public void acknowledge(JmsConnection connection, JmsInboundMessageDispatch envelope, ACK_TYPE ackType) throws JMSException {
+ connection.acknowledge(envelope, ackType);
+ }
+
+ @Override
+ public void addSynchronization(JmsTxSynchronization sync) {
+ }
+
+ @Override
+ public void markAsFailed() {
+ }
+
+ @Override
+ public boolean isFailed() {
+ return false;
+ }
+
+ @Override
+ public void begin() throws JMSException {
+ }
+
+ @Override
+ public void rollback() throws JMSException {
+ throw new javax.jms.IllegalStateException("Not a transacted session");
+ }
+
+ @Override
+ public void commit() throws JMSException {
+ throw new javax.jms.IllegalStateException("Not a transacted session");
+ }
+
+ @Override
+ public JmsTransactionId getTransactionId() {
+ return null;
+ }
+
+ @Override
+ public JmsTransactionListener getListener() {
+ return null;
+ }
+
+ @Override
+ public void setListener(JmsTransactionListener listener) {
+ }
+
+ @Override
+ public boolean isInTransaction() {
+ return false;
+ }
+}
http://git-wip-us.apache.org/repos/asf/qpid-jms/blob/41a99b37/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsSession.java
----------------------------------------------------------------------
diff --git a/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsSession.java b/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsSession.java
index fa6ecc6..a71adb2 100644
--- a/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsSession.java
+++ b/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsSession.java
@@ -95,7 +95,7 @@ public class JmsSession implements Session, QueueSession, TopicSession, JmsMessa
private final AtomicLong consumerIdGenerator = new AtomicLong();
private final AtomicLong producerIdGenerator = new AtomicLong();
- private JmsLocalTransactionContext transactionContext;
+ private JmsTransactionContext transactionContext;
private boolean sessionRecovered;
protected JmsSession(JmsConnection connection, JmsSessionId sessionId, int acknowledgementMode) throws JMSException {
@@ -103,7 +103,11 @@ public class JmsSession implements Session, QueueSession, TopicSession, JmsMessa
this.acknowledgementMode = acknowledgementMode;
this.prefetchPolicy = new JmsPrefetchPolicy(connection.getPrefetchPolicy());
- setTransactionContext(new JmsLocalTransactionContext(this));
+ if (acknowledgementMode == SESSION_TRANSACTED) {
+ setTransactionContext(new JmsLocalTransactionContext(this));
+ } else {
+ setTransactionContext(new JmsNoTxTransactionContext());
+ }
this.sessionInfo = new JmsSessionInfo(sessionId);
this.sessionInfo.setAcknowledgementMode(acknowledgementMode);
@@ -158,27 +162,23 @@ public class JmsSession implements Session, QueueSession, TopicSession, JmsMessa
@Override
public void commit() throws JMSException {
checkClosed();
-
- if (!getTransacted()) {
- throw new javax.jms.IllegalStateException("Not a transacted session");
- }
-
- this.transactionContext.commit();
+ transactionContext.commit();
}
@Override
public void rollback() throws JMSException {
checkClosed();
+
if (!getTransacted()) {
throw new javax.jms.IllegalStateException("Not a transacted session");
}
- //Stop processing any new messages that arrive
+ // Stop processing any new messages that arrive
for (JmsMessageConsumer c : consumers.values()) {
c.suspendForRollback();
}
- this.transactionContext.rollback();
+ transactionContext.rollback();
for (JmsMessageConsumer c : consumers.values()) {
c.resumeAfterRollback();
@@ -212,8 +212,8 @@ public class JmsSession implements Session, QueueSession, TopicSession, JmsMessa
protected void doClose() throws JMSException {
boolean interrupted = Thread.interrupted();
shutdown();
- this.connection.removeSession(this);
- this.connection.destroyResource(sessionInfo);
+ connection.removeSession(this);
+ connection.destroyResource(sessionInfo);
if (interrupted) {
Thread.currentThread().interrupt();
}
@@ -441,7 +441,7 @@ public class JmsSession implements Session, QueueSession, TopicSession, JmsMessa
@Override
public void unsubscribe(String name) throws JMSException {
checkClosed();
- this.connection.unsubscribe(name);
+ connection.unsubscribe(name);
}
//////////////////////////////////////////////////////////////////////////
@@ -595,13 +595,11 @@ public class JmsSession implements Session, QueueSession, TopicSession, JmsMessa
return connection.createTemporaryTopic();
}
- //////////////////////////////////////////////////////////////////////////
- // Session Implementation methods
- //////////////////////////////////////////////////////////////////////////
+ //----- Session Implementation methods -----------------------------------//
protected void add(JmsMessageConsumer consumer) throws JMSException {
- this.consumers.put(consumer.getConsumerId(), consumer);
- this.connection.addDispatcher(consumer.getConsumerId(), this);
+ consumers.put(consumer.getConsumerId(), consumer);
+ connection.addDispatcher(consumer.getConsumerId(), this);
if (started.get()) {
consumer.start();
@@ -609,24 +607,24 @@ public class JmsSession implements Session, QueueSession, TopicSession, JmsMessa
}
protected void remove(JmsMessageConsumer consumer) throws JMSException {
- this.connection.removeDispatcher(consumer.getConsumerId());
- this.consumers.remove(consumer.getConsumerId());
+ connection.removeDispatcher(consumer.getConsumerId());
+ consumers.remove(consumer.getConsumerId());
}
protected void add(JmsMessageProducer producer) {
- this.producers.add(producer);
+ producers.add(producer);
}
protected void remove(MessageProducer producer) {
- this.producers.remove(producer);
+ producers.remove(producer);
}
protected void onException(Exception ex) {
- this.connection.onException(ex);
+ connection.onException(ex);
}
protected void onException(JMSException ex) {
- this.connection.onException(ex);
+ connection.onException(ex);
}
protected void send(JmsMessageProducer producer, Destination dest, Message msg, int deliveryMode, int priority, long timeToLive, boolean disableMsgId, boolean disableTimestamp) throws JMSException {
@@ -642,8 +640,6 @@ public class JmsSession implements Session, QueueSession, TopicSession, JmsMessa
private void send(JmsMessageProducer producer, JmsDestination destination, Message original, int deliveryMode, int priority, long timeToLive, boolean disableMsgId, boolean disableTimestamp) throws JMSException {
sendLock.lock();
try {
- startNextTransaction();
-
original.setJMSDeliveryMode(deliveryMode);
original.setJMSPriority(priority);
original.setJMSRedelivered(false);
@@ -698,18 +694,14 @@ public class JmsSession implements Session, QueueSession, TopicSession, JmsMessa
envelope.setSendAsync(!sync);
envelope.setDispatchId(msgId);
- this.connection.send(envelope);
+ transactionContext.send(connection, envelope);
} finally {
sendLock.unlock();
}
}
- void acknowledge(JmsInboundMessageDispatch envelope, ACK_TYPE ackType, boolean startNextTransaction) throws JMSException {
- if(startNextTransaction) {
- startNextTransaction();
- }
-
- this.connection.acknowledge(envelope, ackType);
+ void acknowledge(JmsInboundMessageDispatch envelope, ACK_TYPE ackType) throws JMSException {
+ transactionContext.acknowledge(connection, envelope, ackType);
}
/**
@@ -719,11 +711,15 @@ public class JmsSession implements Session, QueueSession, TopicSession, JmsMessa
* @throws JMSException if an error occurs while the acknowledge is processed.
*/
void acknowledge() throws JMSException {
+ if (isTransacted()) {
+ throw new IllegalStateException("Session acknowledge called inside a transacted Session");
+ }
+
this.connection.acknowledge(sessionInfo.getSessionId());
}
public boolean isClosed() {
- return this.closed.get();
+ return closed.get();
}
/**
@@ -732,7 +728,7 @@ public class JmsSession implements Session, QueueSession, TopicSession, JmsMessa
* @return true - if the session uses transactions.
*/
public boolean isTransacted() {
- return this.acknowledgementMode == Session.SESSION_TRANSACTED;
+ return acknowledgementMode == Session.SESSION_TRANSACTED;
}
/**
@@ -741,7 +737,7 @@ public class JmsSession implements Session, QueueSession, TopicSession, JmsMessa
* @return true - if the session uses client acknowledgment.
*/
protected boolean isClientAcknowledge() {
- return this.acknowledgementMode == Session.CLIENT_ACKNOWLEDGE;
+ return acknowledgementMode == Session.CLIENT_ACKNOWLEDGE;
}
/**
@@ -763,7 +759,7 @@ public class JmsSession implements Session, QueueSession, TopicSession, JmsMessa
}
protected void checkClosed() throws IllegalStateException {
- if (this.closed.get()) {
+ if (closed.get()) {
throw new IllegalStateException("The Session is closed");
}
}
@@ -837,11 +833,11 @@ public class JmsSession implements Session, QueueSession, TopicSession, JmsMessa
}
protected boolean isStarted() {
- return this.started.get();
+ return started.get();
}
public JmsConnection getConnection() {
- return this.connection;
+ return connection;
}
Executor getExecutor() {
@@ -861,11 +857,11 @@ public class JmsSession implements Session, QueueSession, TopicSession, JmsMessa
}
protected JmsSessionInfo getSessionInfo() {
- return this.sessionInfo;
+ return sessionInfo;
}
protected JmsSessionId getSessionId() {
- return this.sessionInfo.getSessionId();
+ return sessionInfo.getSessionId();
}
protected JmsConsumerId getNextConsumerId() {
@@ -885,12 +881,6 @@ public class JmsSession implements Session, QueueSession, TopicSession, JmsMessa
return message;
}
- private synchronized void startNextTransaction() throws JMSException {
- if (getTransacted()) {
- transactionContext.begin();
- }
- }
-
boolean isDestinationInUse(JmsDestination destination) {
for (JmsMessageConsumer consumer : consumers.values()) {
if (consumer.isUsingDestination(destination)) {
@@ -924,11 +914,18 @@ public class JmsSession implements Session, QueueSession, TopicSession, JmsMessa
if (started.get()) {
deliver(envelope);
} else {
- this.stoppedMessages.add(envelope);
+ stoppedMessages.add(envelope);
}
}
protected void onConnectionInterrupted() {
+
+ if (this.acknowledgementMode == SESSION_TRANSACTED) {
+ if (transactionContext.isInTransaction()) {
+ transactionContext.markAsFailed();
+ }
+ }
+
for (JmsMessageProducer producer : producers) {
producer.onConnectionInterrupted();
}
@@ -944,13 +941,6 @@ public class JmsSession implements Session, QueueSession, TopicSession, JmsMessa
provider.create(sessionInfo, request);
request.sync();
- if (this.acknowledgementMode == SESSION_TRANSACTED) {
- if (transactionContext.isInTransaction()) {
- transactionContext.clear();
- transactionContext.begin();
- }
- }
-
for (JmsMessageProducer producer : producers) {
producer.onConnectionRecovery(provider);
}
@@ -985,10 +975,10 @@ public class JmsSession implements Session, QueueSession, TopicSession, JmsMessa
if (id == null) {
this.connection.onException(new JMSException("No ConsumerId set for " + envelope.getMessage()));
}
- if (this.messageListener != null) {
- this.messageListener.onMessage(envelope.getMessage());
+ if (messageListener != null) {
+ messageListener.onMessage(envelope.getMessage());
} else {
- JmsMessageConsumer consumer = this.consumers.get(id);
+ JmsMessageConsumer consumer = consumers.get(id);
if (consumer != null) {
consumer.onInboundMessage(envelope);
}
@@ -1001,7 +991,7 @@ public class JmsSession implements Session, QueueSession, TopicSession, JmsMessa
* @param transactionContext
* provides the means to control a JMS transaction.
*/
- public void setTransactionContext(JmsLocalTransactionContext transactionContext) {
+ public void setTransactionContext(JmsTransactionContext transactionContext) {
this.transactionContext = transactionContext;
}
@@ -1011,7 +1001,7 @@ public class JmsSession implements Session, QueueSession, TopicSession, JmsMessa
* @return transactionContext
* session's transaction context.
*/
- public JmsLocalTransactionContext getTransactionContext() {
+ public JmsTransactionContext getTransactionContext() {
return transactionContext;
}
http://git-wip-us.apache.org/repos/asf/qpid-jms/blob/41a99b37/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsTransactionContext.java
----------------------------------------------------------------------
diff --git a/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsTransactionContext.java b/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsTransactionContext.java
new file mode 100644
index 0000000..2d5524d
--- /dev/null
+++ b/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsTransactionContext.java
@@ -0,0 +1,137 @@
+/**
+ * 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.qpid.jms;
+
+import javax.jms.JMSException;
+
+import org.apache.qpid.jms.message.JmsInboundMessageDispatch;
+import org.apache.qpid.jms.message.JmsOutboundMessageDispatch;
+import org.apache.qpid.jms.meta.JmsTransactionId;
+import org.apache.qpid.jms.provider.ProviderConstants.ACK_TYPE;
+
+/**
+ * A Transaction Context is used to track and manage the state of a
+ * Transaction in a JMS Session object.
+ */
+public interface JmsTransactionContext {
+
+ /**
+ * Allows the context to intercept a message acknowledgement and perform any
+ * additional logic prior to the acknowledge being forwarded onto the connection.
+ *
+ * @param connection
+ * the connection that the acknowledge will be forwarded to.
+ * @param envelope
+ * the envelope that contains the message to be acknowledged.
+ * @param ackType
+ * the acknowledgement type being requested.
+ *
+ * @throws JMSException if an error occurs while performing the acknowledge.
+ */
+ void acknowledge(JmsConnection connection, JmsInboundMessageDispatch envelope, ACK_TYPE ackType) throws JMSException;
+
+ /**
+ * Allows the context to intercept and perform any additional logic
+ * prior to a message being sent on to the connection and subsequently
+ * the remote peer.
+ *
+ * @param connection
+ * the connection that will be do the send of the message
+ * @param envelope
+ * the envelope that contains the message to be sent.
+ *
+ * @throws JMSException if an error occurs during the send.
+ */
+ void send(JmsConnection connection, JmsOutboundMessageDispatch envelope) throws JMSException;
+
+ /**
+ * Adds the given Transaction synchronization to the current list. The
+ * registered synchronization will be notified of various event points
+ * in the lifetime of a transaction such as before and after commit or
+ * rollback.
+ *
+ * @param sync
+ * the transaction synchronization to add.
+ */
+ void addSynchronization(JmsTxSynchronization sync);
+
+ /**
+ * Marks an currently active Transaction as being failed, usually due to a
+ * connection failure. Once failed all the transaction must be rolled back
+ * before new work can be done.
+ */
+ void markAsFailed();
+
+ /**
+ * @returns if the currently transaction has been marked as being failed.
+ */
+ boolean isFailed();
+
+ /**
+ * Start a transaction if none is currently active.
+ *
+ * @throws JMSException on internal error occurs.
+ */
+ void begin() throws JMSException;
+
+ /**
+ * Rolls back any work done in this transaction and releases any locks
+ * currently held. If the current transaction is in a failed state this
+ * resets that state and prepares the context for a new transaction to be
+ * stated via a call to <code>begin</code>.
+ *
+ * @throws JMSException
+ * if the JMS provider fails to roll back the transaction due to some internal error.
+ */
+ void rollback() throws JMSException;
+
+ /**
+ * Commits all work done in this transaction and releases any locks
+ * currently held. If the transaction is in a failed state this method
+ * throws an exception to indicate that the transaction has failed and
+ * will be rolled back.
+ *
+ * @throws JMSException
+ * if the commit fails to roll back the transaction due to some internal error.
+ */
+ void commit() throws JMSException;
+
+ /**
+ * @return the transaction ID of the currently active TX or null if none active.
+ */
+ JmsTransactionId getTransactionId();
+
+ /**
+ * @return the currently configured JMS Transaction listener which will receive TX events.
+ */
+ JmsTransactionListener getListener();
+
+ /**
+ * Sets the single JMS Transaction listener which will be notified of significant TX events
+ * such as Commit or Rollback.
+ *
+ * @param listener
+ * the JMS Transaction listener that will be sent all TX event notifications.
+ */
+ void setListener(JmsTransactionListener listener);
+
+ /**
+ * @return true if there is a transaction in progress even if the current is failed.
+ */
+ boolean isInTransaction();
+
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org