Mailing-List: contact commits-help@activemq.apache.org; run by ezmlm
Precedence: bulk
Reply-To: dev@activemq.apache.org
Content-Type: text/plain; charset="utf-8"
MIME-Version: 1.0
Content-Transfer-Encoding: 7bit
Subject: svn commit: r813927 -
 /activemq/trunk/activemq-core/src/main/java/org/apache/activemq/TransactionContext.java
Date: Fri, 11 Sep 2009 17:10:49 -0000
To: commits@activemq.apache.org
From: rajdavies@apache.org
Message-Id: <20090911171049.DE52F238889D@eris.apache.org>

Author: rajdavies
Date: Fri Sep 11 17:10:49 2009
New Revision: 813927

URL: http://svn.apache.org/viewvc?rev=813927&view=rev
Log:
patch applied for https://issues.apache.org/activemq/browse/AMQ-2191

Modified:
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/TransactionContext.java

Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/TransactionContext.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/TransactionContext.java?rev=813927&r1=813926&r2=813927&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/TransactionContext.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/TransactionContext.java Fri Sep 11 17:10:49 2009
@@ -16,6 +16,7 @@
  */
 package org.apache.activemq;
 
+import java.io.InterruptedIOException;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
@@ -27,11 +28,13 @@
 import javax.transaction.xa.XAResource;
 import javax.transaction.xa.Xid;
 
+import org.apache.activemq.command.Command;
 import org.apache.activemq.command.ConnectionId;
 import org.apache.activemq.command.DataArrayResponse;
 import org.apache.activemq.command.DataStructure;
 import org.apache.activemq.command.IntegerResponse;
 import org.apache.activemq.command.LocalTransactionId;
+import org.apache.activemq.command.Response;
 import org.apache.activemq.command.TransactionId;
 import org.apache.activemq.command.TransactionInfo;
 import org.apache.activemq.command.XATransactionId;
@@ -280,7 +283,7 @@
             TransactionInfo info = new TransactionInfo(getConnectionId(), transactionId, TransactionInfo.COMMIT_ONE_PHASE);
             this.transactionId = null;
             // Notify the listener that the tx was committed back
-            this.connection.syncSendPacket(info);
+            syncSendPacketWithInterruptionHandling(info);
             if (localTransactionEventListener != null) {
                 localTransactionEventListener.commitEvent();
             }
@@ -399,7 +402,7 @@
             TransactionInfo info = new TransactionInfo(getConnectionId(), x, TransactionInfo.PREPARE);
 
             // Find out if the server wants to commit or rollback.
-            IntegerResponse response = (IntegerResponse)this.connection.syncSendPacket(info);
+            IntegerResponse response = (IntegerResponse)syncSendPacketWithInterruptionHandling(info);
             return response.getResult();
 
         } catch (JMSException e) {
@@ -433,7 +436,7 @@
 
             // Let the server know that the tx is rollback.
             TransactionInfo info = new TransactionInfo(getConnectionId(), x, TransactionInfo.ROLLBACK);
-            this.connection.syncSendPacket(info);
+            syncSendPacketWithInterruptionHandling(info);
 
             List<TransactionContext> l = ENDED_XA_TRANSACTION_CONTEXTS.remove(x);
             if (l != null && !l.isEmpty()) {
@@ -472,7 +475,7 @@
             // Notify the server that the tx was committed back
             TransactionInfo info = new TransactionInfo(getConnectionId(), x, onePhase ? TransactionInfo.COMMIT_ONE_PHASE : TransactionInfo.COMMIT_TWO_PHASE);
 
-            this.connection.syncSendPacket(info);
+            syncSendPacketWithInterruptionHandling(info);
 
             List<TransactionContext> l = ENDED_XA_TRANSACTION_CONTEXTS.remove(x);
             if (l != null && !l.isEmpty()) {
@@ -509,7 +512,7 @@
 
         try {
             // Tell the server to forget the transaction.
-            this.connection.syncSendPacket(info);
+            syncSendPacketWithInterruptionHandling(info);
         } catch (JMSException e) {
             throw toXAException(e);
         }
@@ -601,7 +604,7 @@
             if (transactionId != null) {
                 TransactionInfo info = new TransactionInfo(connectionId, transactionId, TransactionInfo.END);
                 try {
-                    this.connection.syncSendPacket(info);
+                    syncSendPacketWithInterruptionHandling(info);
                     if (LOG.isDebugEnabled()) {
                         LOG.debug("Ended XA transaction: " + transactionId);
                     }
@@ -628,6 +631,31 @@
     }
 
     /**
+     * Sends the given command. Also sends the command in case of interruption,
+     * so that important commands like rollback and commit are never interrupted.
+     * If interruption occurred, set the interruption state of the current 
+     * after performing the action again. 
+     * 
+     * @return the response
+     */
+    private Response syncSendPacketWithInterruptionHandling(Command command) throws JMSException {
+    	try {
+			return this.connection.syncSendPacket(command);
+		} catch (JMSException e) {
+			if (e.getLinkedException() instanceof InterruptedIOException) {
+				try {
+					Thread.interrupted();
+					return this.connection.syncSendPacket(command);
+				} finally {
+					Thread.currentThread().interrupt();
+				}				
+			}
+			
+			throw e;
+		}
+    }
+
+    /**
      * Converts a JMSException from the server to an XAException. if the
      * JMSException contained a linked XAException that is returned instead.
      *