[jboss-cvs] JBoss Messaging SVN: r8537 - branches/Branch_1_4/src/main/org/jboss/messaging/core/impl.

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

Author: gaohoward
Date: 2012-05-23 07:36:02 -0400 (Wed, 23 May 2012)
New Revision: 8537

Modified:
   branches/Branch_1_4/src/main/org/jboss/messaging/core/impl/JDBCPersistenceManager.java
Log:
JBMESSAGING-1826



Modified: branches/Branch_1_4/src/main/org/jboss/messaging/core/impl/JDBCPersistenceManager.java
===================================================================
--- branches/Branch_1_4/src/main/org/jboss/messaging/core/impl/JDBCPersistenceManager.java	2012-05-09 10:41:23 UTC (rev 8536)
+++ branches/Branch_1_4/src/main/org/jboss/messaging/core/impl/JDBCPersistenceManager.java	2012-05-23 11:36:02 UTC (rev 8537)
@@ -822,6 +822,8 @@
                   }
 
                }
+               
+               deleteMessagesNonTx(references, conn);
 
                return null;
             }
@@ -833,8 +835,6 @@
       }
 
       new RemoveDepagedReferencesRunner().executeWithRetry();
-
-      deleteMessages(references);
    }
 
    // After loading paged refs this is used to update P messages to non paged
@@ -1848,6 +1848,8 @@
                   log.trace("Deleted " + rows + " references");
                }
 
+               deleteMessageNonTx(ref.getMessage().getMessageID(), conn);
+
                return null;
             }
             finally
@@ -1870,8 +1872,6 @@
          // No tx so we remove the reference directly from the db
 
          new RemoveReferenceRunner().executeWithRetry();
-
-         deleteMessage(ref.getMessage().getMessageID());
       }
    }
 
@@ -2166,6 +2166,8 @@
 
                }
 
+               deleteMessagesNonTx(refsToRemove, conn);
+
                return null;
             }
             catch (SQLException e)
@@ -2207,8 +2209,6 @@
       }
 
       new HandleBeforeCommit1PCRunner().executeWithRetry();
-
-      this.deleteMessages(refsToRemove);
    }
 
    protected void handleBeforeCommit2PC(final List refsToRemove,
@@ -2260,6 +2260,8 @@
 
                removeTXRecord(conn, tx);
 
+               deleteMessagesNonTx(refsToRemove, conn);
+
                return null;
             }
             finally
@@ -2270,8 +2272,6 @@
       }
 
       new HandleBeforeCommit2PCRunner().executeWithRetry();
-
-      this.deleteMessages(refsToRemove);
    }
 
    protected void handleBeforePrepare(final List refsToAdd,
@@ -2469,6 +2469,8 @@
 
                removeTXRecord(conn, tx);
 
+               deleteMessagesNonTx(refsToAdd, conn);
+               
                return null;
             }
             finally
@@ -2479,8 +2481,6 @@
       }
 
       new HandleBeforeRollbackRunner().executeWithRetry();
-
-      this.deleteMessages(refsToAdd);
    }
 
    protected void addTXRecord(Connection conn, Transaction tx) throws Exception
@@ -3118,97 +3118,72 @@
 
    // Private -------------------------------------------------------
 
-   private void deleteMessages(final List references) throws Exception
+   private void deleteMessagesNonTx(final List references, Connection conn) throws Exception
    {
-      class DeleteMessagesRunner extends JDBCTxRunner2
+      // Order to avoid deadlock
+      orderReferences(references);
+      
+      PreparedStatement psMessage = null;
+      try
       {
-         public Object doTransaction() throws Exception
-         {
+         psMessage = conn
+               .prepareStatement(getSQLStatement("DELETE_MESSAGE"));
 
-            PreparedStatement psMessage = null;
+         Iterator iter = references.iterator();
 
-            try
+         while (iter.hasNext())
+         {
+            Object obj = iter.next();
+
+            MessageReference ref;
+            if (obj instanceof MessageReference)
             {
-               psMessage = conn
-                     .prepareStatement(getSQLStatement("DELETE_MESSAGE"));
+               ref = (MessageReference) obj;
+            }
+            else
+            {
+               ref = ((ChannelRefPair) obj).ref;
+            }
 
-               Iterator iter = references.iterator();
+            psMessage.setLong(1, ref.getMessage().getMessageID());
+            psMessage.setLong(2, ref.getMessage().getMessageID());
 
-               while (iter.hasNext())
-               {
-                  Object obj = iter.next();
+            int rows = psMessage.executeUpdate();
 
-                  MessageReference ref;
-                  if (obj instanceof MessageReference)
-                  {
-                     ref = (MessageReference) obj;
-                  }
-                  else
-                  {
-                     ref = ((ChannelRefPair) obj).ref;
-                  }
-
-                  psMessage.setLong(1, ref.getMessage().getMessageID());
-                  psMessage.setLong(2, ref.getMessage().getMessageID());
-
-                  int rows = psMessage.executeUpdate();
-
-                  if (trace)
-                  {
-                     log.trace("Deleted " + rows + " messages");
-                  }
-
-               }
-
-               return null;
-            }
-            finally
+            if (trace)
             {
-               closeStatement(psMessage);
+               log.trace("Deleted " + rows + " messages");
             }
          }
       }
-
-      // Order to avoid deadlock
-      orderReferences(references);
-
-      new DeleteMessagesRunner().executeWithRetry();
+      finally
+      {
+         closeStatement(psMessage);
+      }
    }
 
-   private void deleteMessage(final long messageID) throws Exception
+   private void deleteMessageNonTx(final long messageID, Connection conn) throws Exception
    {
-      class DeleteMessageRunner extends JDBCTxRunner2
+      PreparedStatement psMessage = null;
+      try
       {
-         public Object doTransaction() throws Exception
-         {
+         psMessage = conn
+               .prepareStatement(getSQLStatement("DELETE_MESSAGE"));
 
-            PreparedStatement psMessage = null;
+         psMessage.setLong(1, messageID);
+         psMessage.setLong(2, messageID);
 
-            try
-            {
-               psMessage = conn
-                     .prepareStatement(getSQLStatement("DELETE_MESSAGE"));
+         int rows = psMessage.executeUpdate();
 
-               psMessage.setLong(1, messageID);
-               psMessage.setLong(2, messageID);
-
-               int rows = psMessage.executeUpdate();
-
-               if (trace)
-               {
-                  log.trace("Deleted " + rows + " messages");
-               }
-
-               return null;
-            }
-            finally
-            {
-               closeStatement(psMessage);
-            }
+         if (trace)
+         {
+            log.trace("Deleted " + rows + " messages");
          }
       }
-
-      new DeleteMessageRunner().executeWithRetry();
+      finally
+      {
+         closeStatement(psMessage);
+      }
    }
 
    private List getMessageChannelPair(String sqlQuery, long transactionId)