[Jboss-cvs] JBossAS SVN: r56548 - branches/JBoss_4_0_3_SP1_JBAS-3622/messaging/src/main/org/jboss/mq/server

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

Author: luc.texier at jboss.com
Date: 2006-09-04 11:07:35 -0400 (Mon, 04 Sep 2006)
New Revision: 56548

Modified:
   branches/JBoss_4_0_3_SP1_JBAS-3622/messaging/src/main/org/jboss/mq/server/BasicQueue.java
Log:
JBAS-3622 added fix to ack in case of a remove failure 

Modified: branches/JBoss_4_0_3_SP1_JBAS-3622/messaging/src/main/org/jboss/mq/server/BasicQueue.java
===================================================================
--- branches/JBoss_4_0_3_SP1_JBAS-3622/messaging/src/main/org/jboss/mq/server/BasicQueue.java	2006-09-04 15:02:07 UTC (rev 56547)
+++ branches/JBoss_4_0_3_SP1_JBAS-3622/messaging/src/main/org/jboss/mq/server/BasicQueue.java	2006-09-04 15:07:35 UTC (rev 56548)
@@ -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,21 @@
       }
       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);
+            server.getPersistenceManager().getTxManager().addPostCommitTask(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 +588,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