[rhmessaging-commits] rhmessaging commits: r3273 - store/branches/java/0.5-release/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb.

[rhmessaging-commits at lists.jboss.org]

Author: ritchiem
Date: 2009-04-09 11:40:04 -0400 (Thu, 09 Apr 2009)
New Revision: 3273

Modified:
   store/branches/java/0.5-release/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/BDBMessageStore.java
Log:
RHM-8 : Applied patch from Robert Godfrey with some additional comments
REVERTED change as performance difference was negligable
Merged reverting change from trunk r3272 


Modified: store/branches/java/0.5-release/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/BDBMessageStore.java
===================================================================
--- store/branches/java/0.5-release/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/BDBMessageStore.java	2009-04-09 15:37:17 UTC (rev 3272)
+++ store/branches/java/0.5-release/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/BDBMessageStore.java	2009-04-09 15:40:04 UTC (rev 3273)
@@ -500,7 +500,8 @@
             _log.debug("Message Id: " + messageId + " Removing");
         }
 
-        //Remove the MetaData then iterate deleting all the ContentChunks we can for the given messageId
+        // first we need to look up the header to get the chunk count
+        MessageMetaData mmd = getMessageMetaData(context, messageId);
         try
         {
             OperationStatus status = _messageMetaDataDb.delete(tx, key);
@@ -522,21 +523,45 @@
 
             DatabaseEntry contentKey = new DatabaseEntry();
             TupleBinding contentKeyBinding = new MessageContentKey.TupleBinding();
-            int i = 0;
-
-            do
+            for (int i = 0; i < mmd.getContentChunkCount(); i++)
             {
                 MessageContentKey mck = new MessageContentKey(messageId, i);
                 contentKeyBinding.objectToEntry(mck, contentKey);
+                status = _messageContentDb.get(tx, contentKey, new DatabaseEntry(), LockMode.RMW);
+                if (status == OperationStatus.NOTFOUND)
+                {
+                    if (localTx)
+                    {
+                        tx.abort();
+                        context.setPayload(null);
+                    }
+
+                    throw new AMQException("Content chunk " + i + " not found for message " + messageId);
+                }
+
                 status = _messageContentDb.delete(tx, contentKey);
-                i++;
-            } while (status != OperationStatus.NOTFOUND);
+                if (status == OperationStatus.NOTFOUND)
+                {
+                    if (localTx)
+                    {
+                        tx.abort();
+                        context.setPayload(null);
+                    }
 
+                    throw new AMQException("Content chunk " + i + " not found for message " + messageId);
+                }
+
+                if (_log.isDebugEnabled())
+                {
+                    _log.debug("Deleted content chunk " + i + " for message " + messageId);
+                }
+            }
+
             if (localTx)
             {
-                // Should we not be doing this async as
+                // ? Will this not perform the environment default commit? Should we not be doing this async as
                 // remove will only occur when message has been fully dequeued? 2009-03-31
-                commit(tx);
+                tx.commit();
                 context.setPayload(null);
             }
         }