You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by ri...@apache.org on 2009/04/11 03:57:36 UTC
svn commit: r764148 - in
/qpid/branches/0.5-release/qpid/java/broker/src/main/java/org/apache/qpid/server:
queue/SimpleAMQQueue.java transactionlog/BaseTransactionLog.java
Author: ritchiem
Date: Sat Apr 11 01:57:36 2009
New Revision: 764148
URL: http://svn.apache.org/viewvc?rev=764148&view=rev
Log:
QPID-1800: modify SAMQQ to record transactionlog etry even when queue isnt durable(consistent with restof broker) and send both current and new queue to
BTL for enqueue to ensure references are kept properly. Update BTL to check for prevous enqueues and record new enques in any existing list, despatching only new enqueues to the delegate
merged from trunk r764075
Modified:
qpid/branches/0.5-release/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java
qpid/branches/0.5-release/qpid/java/broker/src/main/java/org/apache/qpid/server/transactionlog/BaseTransactionLog.java
Modified: qpid/branches/0.5-release/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java
URL: http://svn.apache.org/viewvc/qpid/branches/0.5-release/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java?rev=764148&r1=764147&r2=764148&view=diff
==============================================================================
--- qpid/branches/0.5-release/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java (original)
+++ qpid/branches/0.5-release/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java Sat Apr 11 01:57:36 2009
@@ -823,6 +823,12 @@
AMQQueue toQueue = getVirtualHost().getQueueRegistry().getQueue(new AMQShortString(queueName));
TransactionLog transactionLog = getVirtualHost().getTransactionLog();
+
+ if (toQueue.equals(this))
+ {
+ //nothing to do here, message is already at the requested destination
+ return;
+ }
List<QueueEntry> entries = getMessagesOnTheQueue(new QueueEntryFilter()
{
@@ -848,19 +854,24 @@
// Move the messages in the transaction log.
for (QueueEntry entry : entries)
{
- if (entry.isPersistent() && toQueue.isDurable())
+ if (entry.isPersistent())
{
//FIXME
//fixme
- ArrayList list = new ArrayList();
+
+ // Creating a list with the destination queue AND the current queue.
+ // This is a hack to ensure a reference is kept in the TLog to the new destination when dequeing
+ // the old destination below, thus preventing incorrect removal of the message from the store
+ ArrayList<AMQQueue> list = new ArrayList<AMQQueue>();
list.add(toQueue);
+ list.add(this);
transactionLog.enqueueMessage(storeContext, list, entry.getMessageId());
}
// dequeue will remove the messages from the queue
entry.dequeue(storeContext);
}
- // Commit and flush the move transcations.
+ // Commit and flush the move transactions.
try
{
transactionLog.commitTran(storeContext);
@@ -891,7 +902,7 @@
toQueue.enqueue(storeContext, entry.getMessage());
// As we only did a dequeue above now that we have moved the message we should perform a delete.
// We cannot do this earlier as the message will be lost if flowed.
- //entry.delete();
+ entry.delete();
}
}
catch (MessageCleanupException e)
@@ -913,6 +924,12 @@
AMQQueue toQueue = getVirtualHost().getQueueRegistry().getQueue(new AMQShortString(queueName));
TransactionLog transactionLog = getVirtualHost().getTransactionLog();
+ if (toQueue.equals(this))
+ {
+ //nothing to do here, message is already at the requested destination
+ return;
+ }
+
List<QueueEntry> entries = getMessagesOnTheQueue(new QueueEntryFilter()
{
@@ -944,11 +961,15 @@
// Move the messages in on the transaction log.
for (QueueEntry entry : entries)
{
- if (!entry.isDeleted() && entry.isPersistent() && toQueue.isDurable())
+ if (!entry.isDeleted() && entry.isPersistent())
{
//fixme
//FIXME
+
+ // Creating a list with the destination queue AND the current queue.
+ // This is a hack to ensure a reference is kept in the TLog to the old destination when enqueing
ArrayList list = new ArrayList();
+ list.add(this);
list.add(toQueue);
transactionLog.enqueueMessage(storeContext, list, entry.getMessageId());
}
Modified: qpid/branches/0.5-release/qpid/java/broker/src/main/java/org/apache/qpid/server/transactionlog/BaseTransactionLog.java
URL: http://svn.apache.org/viewvc/qpid/branches/0.5-release/qpid/java/broker/src/main/java/org/apache/qpid/server/transactionlog/BaseTransactionLog.java?rev=764148&r1=764147&r2=764148&view=diff
==============================================================================
--- qpid/branches/0.5-release/qpid/java/broker/src/main/java/org/apache/qpid/server/transactionlog/BaseTransactionLog.java (original)
+++ qpid/branches/0.5-release/qpid/java/broker/src/main/java/org/apache/qpid/server/transactionlog/BaseTransactionLog.java Sat Apr 11 01:57:36 2009
@@ -67,12 +67,47 @@
{
_logger.info("Recording Enqueue of (" + messageId + ") on queue:" + queues);
}
+
+ //list to hold which new queues to enqueue the message on
+ ArrayList<AMQQueue> toEnqueueList = new ArrayList<AMQQueue>();
+
+ List<AMQQueue> enqueuedList = _idToQueues.get(messageId);
+ if (enqueuedList != null)
+ {
+ //There are previous enqueues for this messageId
+ synchronized (enqueuedList)
+ {
+ for(AMQQueue queue : queues)
+ {
+ if(!enqueuedList.contains(queue))
+ {
+ //update the old list.
+ enqueuedList.add(queue);
+ //keep track of new enqueues to be made
+ toEnqueueList.add(queue);
+ }
+ }
+ }
+
+ if(toEnqueueList.isEmpty())
+ {
+ //no new queues to enqueue message on
+ return;
+ }
+ }
+ else
+ {
+ //No existing list, add all provided queues (cloning toEnqueueList in case someone else changes original).
+ toEnqueueList.addAll(queues);
+ _idToQueues.put(messageId, Collections.synchronizedList((ArrayList<AMQQueue>)toEnqueueList.clone()));
+ }
- //Clone the list incase someone else changes it.
- _idToQueues.put(messageId, Collections.synchronizedList((ArrayList<AMQQueue>)queues.clone()));
+ _delegate.enqueueMessage(context, toEnqueueList, messageId);
+ }
+ else
+ {
+ _delegate.enqueueMessage(context, queues, messageId);
}
-
- _delegate.enqueueMessage(context, queues, messageId);
}
public void dequeueMessage(StoreContext context, AMQQueue queue, Long messageId) throws AMQException
---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project: http://qpid.apache.org
Use/Interact: mailto:commits-subscribe@qpid.apache.org