[jboss-cvs] JBossAS SVN: r57246 - branches/JBoss_4_0_3_SP1_CP/messaging/src/main/org/jboss/mq/server

[jboss-cvs-commits at lists.jboss.org]

Author: ryan.campbell at jboss.com
Date: 2006-09-27 18:45:53 -0400 (Wed, 27 Sep 2006)
New Revision: 57246

Modified:
   branches/JBoss_4_0_3_SP1_CP/messaging/src/main/org/jboss/mq/server/BasicQueue.java
Log:
ASPATCH-66: JBAS-3622: Patch for: JBAS-3573: Failure to remove a message should result in a NACK + error

Modified: branches/JBoss_4_0_3_SP1_CP/messaging/src/main/org/jboss/mq/server/BasicQueue.java
===================================================================
--- branches/JBoss_4_0_3_SP1_CP/messaging/src/main/org/jboss/mq/server/BasicQueue.java	2006-09-27 22:14:25 UTC (rev 57245)
+++ branches/JBoss_4_0_3_SP1_CP/messaging/src/main/org/jboss/mq/server/BasicQueue.java	2006-09-27 22:45:53 UTC (rev 57246)
@@ -26,6 +26,7 @@
 import org.jboss.mq.SpyMessage;
 import org.jboss.mq.Subscription;
 import org.jboss.mq.pm.Tx;
+import org.jboss.mq.pm.TxManager;
 import org.jboss.mq.selectors.Selector;
 import org.jboss.util.timeout.Timeout;
 import org.jboss.util.timeout.TimeoutFactory;
@@ -560,14 +561,23 @@
       }
       else
       {
-         if (m.isPersistent())
-            server.getPersistenceManager().remove(m, txId);
+         try
+         {
+            if (m.isPersistent())
+               server.getPersistenceManager().remove(m, txId);
+         }
+         catch (Throwable t)
+         {
+            // Something is wrong with the persistence manager,
+            // force a NACK with a rollback/error
+            Runnable task = new RestoreMessageTask(m);
+            TxManager txManager = server.getPersistenceManager().getTxManager();
+            txManager.addPostCommitTask(txId, task);            
+            txManager.addPostRollbackTask(txId, task);
+            SpyJMSException.rethrowAsJMSException("Error during ACK ref=" + m, t);
+         }
 
-         Runnable task = new RestoreMessageTask(m);
-         server.getPersistenceManager().getTxManager().addPostRollbackTask(txId, task);
-
-         task = new RemoveMessageTask(m);
-         server.getPersistenceManager().getTxManager().addPostCommitTask(txId, task);
+         performOrPerpareAcknowledgeMessage(m, txId);
       }
 
       synchronized (receivers)
@@ -580,6 +590,27 @@
    }
 
    /**
+    * Either perform or prepare the acknowledge message 
+    * 
+    * @param mes the message reference
+    * @param txId the transaction id
+    * @throws Exception for any error
+    */
+   protected void performOrPerpareAcknowledgeMessage(MessageReference mes, Tx txId) throws JMSException
+   {
+      TxManager txManager = server.getPersistenceManager().getTxManager();
+      
+      // The message is restored to the queue on a rollback
+      Runnable task = new RestoreMessageTask(mes);
+      txManager.addPostRollbackTask(txId, task);
+
+      // The message is fully removed after the transaction commits
+      task = new RemoveMessageTask(mes);
+      txManager.addPostCommitTask(txId, task);
+   }
+   
+   
+   /**
     * Nack all messages for a subscription
     *
     * @param sub the subscription