[jboss-cvs] JBossAS SVN: r73848 - in trunk/jbossmq/src/main/org/jboss/mq/pm: jdbc2 and 1 other directory.

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

Author: adrian at jboss.org
Date: 2008-05-30 11:32:34 -0400 (Fri, 30 May 2008)
New Revision: 73848

Added:
   trunk/jbossmq/src/main/org/jboss/mq/pm/PersistenceManagerExt.java
Modified:
   trunk/jbossmq/src/main/org/jboss/mq/pm/TxManager.java
   trunk/jbossmq/src/main/org/jboss/mq/pm/jdbc2/PersistenceManager.java
Log:
[JBAS-5571] - Port from jboss-4.2

Added: trunk/jbossmq/src/main/org/jboss/mq/pm/PersistenceManagerExt.java
===================================================================
--- trunk/jbossmq/src/main/org/jboss/mq/pm/PersistenceManagerExt.java	                        (rev 0)
+++ trunk/jbossmq/src/main/org/jboss/mq/pm/PersistenceManagerExt.java	2008-05-30 15:32:34 UTC (rev 73848)
@@ -0,0 +1,41 @@
+/*
+* JBoss, Home of Professional Open Source
+* Copyright 2007, JBoss Inc., and individual contributors as indicated
+* by the @authors tag. See the copyright.txt in the distribution for a
+* full listing of individual contributors.
+*
+* This is free software; you can redistribute it and/or modify it
+* under the terms of the GNU Lesser General Public License as
+* published by the Free Software Foundation; either version 2.1 of
+* the License, or (at your option) any later version.
+*
+* This software is distributed in the hope that it will be useful,
+* but WITHOUT ANY WARRANTY; without even the implied warranty of
+* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+* Lesser General Public License for more details.
+*
+* You should have received a copy of the GNU Lesser General Public
+* License along with this software; if not, write to the Free
+* Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA
+* 02110-1301 USA, or see the FSF site: http://www.fsf.org.
+*/
+package org.jboss.mq.pm;
+
+import javax.jms.JMSException;
+
+/**
+ * PersistenceManagerExt.
+ * 
+ * @author <a href="adrian at jboss.com">Adrian Brock</a>
+ * @version $Revision: 1.1 $
+ */
+public interface PersistenceManagerExt
+{
+   /**
+    * Force the transaction to the log
+    * 
+    * @param tx the transaction
+    * @throws JMSException for any error
+    */
+   void forcePersistentTx(Tx tx) throws JMSException;
+}

Modified: trunk/jbossmq/src/main/org/jboss/mq/pm/TxManager.java
===================================================================
--- trunk/jbossmq/src/main/org/jboss/mq/pm/TxManager.java	2008-05-30 15:32:07 UTC (rev 73847)
+++ trunk/jbossmq/src/main/org/jboss/mq/pm/TxManager.java	2008-05-30 15:32:34 UTC (rev 73848)
@@ -100,6 +100,18 @@
    }
 
    /**
+     * Force the txid to the log
+     * 
+     * @param txid the transaction id
+     * @exception JMSException for any error
+     */
+   public final void forcePersistentTx(Tx txid) throws JMSException
+   {
+      if (persistenceManager instanceof PersistenceManagerExt)
+        ((PersistenceManagerExt)persistenceManager).forcePersistentTx(txid);
+   }
+
+   /**
 	 * Commit the transaction to the persistent store.
 	 * 
 	 * @param txId the transaction
@@ -303,7 +315,12 @@
       try
       {
          if (xid instanceof Xid)
+         {
+            // If we are marking this prepared, we need to log the txid even if it has no messages
+            // in the transaction
+            forcePersistentTx(txId);
             addPreparedTx(txId, (Xid) xid, false);
+         }
       }
       catch (Throwable t)
       {

Modified: trunk/jbossmq/src/main/org/jboss/mq/pm/jdbc2/PersistenceManager.java
===================================================================
--- trunk/jbossmq/src/main/org/jboss/mq/pm/jdbc2/PersistenceManager.java	2008-05-30 15:32:07 UTC (rev 73847)
+++ trunk/jbossmq/src/main/org/jboss/mq/pm/jdbc2/PersistenceManager.java	2008-05-30 15:32:34 UTC (rev 73848)
@@ -54,6 +54,7 @@
 import org.jboss.mq.SpyMessage;
 import org.jboss.mq.SpyTopic;
 import org.jboss.mq.pm.CacheStore;
+import org.jboss.mq.pm.PersistenceManagerExt;
 import org.jboss.mq.pm.Tx;
 import org.jboss.mq.pm.TxManager;
 import org.jboss.mq.server.JMSDestination;
@@ -76,7 +77,7 @@
  * @version $Revision$
  */
 public class PersistenceManager extends ServiceMBeanSupport
-   implements PersistenceManagerMBean, org.jboss.mq.pm.PersistenceManager, CacheStore
+   implements PersistenceManagerMBean, org.jboss.mq.pm.PersistenceManager, CacheStore, PersistenceManagerExt
 {
    /** FAQ about concurrency problems */
    private static String CONCURRENCY_WARNING = "\nCommon reasons for missing messages are \n1) You have multiple JBossMQs running over the same database.\n2) You are using a replicating database that is not keeping up with replication.";
@@ -822,7 +823,46 @@
          throw new IOException("Could not load the message: " + e);
       }
    }
+   
+   public void forcePersistentTx(Tx txId) throws javax.jms.JMSException
+   {
+      // No need if already done or not doing xa recovery
+      if (txId.wasPersisted() || xaRecovery == false)
+         return;
+      
+      TransactionManagerStrategy tms = new TransactionManagerStrategy();
+      tms.startTX();
+      Connection c = null;
+      boolean threadWasInterrupted = Thread.interrupted();
+      try
+      {
 
+         c = this.getConnection();
+         insertPersistentTx(tms, c, txId);
+      }
+      catch (SQLException e)
+      {
+         tms.setRollbackOnly();
+         throw new SpyJMSException("Could not commit tx: " + txId, e);
+      }
+      finally
+      {
+         try
+         {
+            if (c != null)
+               c.close();
+         }
+         catch (Throwable ignore)
+         {
+         }
+         tms.endTX();
+
+         // Restore the interrupted state of the thread
+         if (threadWasInterrupted)
+            Thread.currentThread().interrupt();
+      }
+   }
+
    /////////////////////////////////////////////////////////////////////////////////
    //
    // TX Commit