7:22 a.m.
Author: ritchiem
Date: 2009-03-13 08:22:07 -0400 (Fri, 13 Mar 2009)
New Revision: 3155
Modified:
store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/BDBMessageStore.java
store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/ExchangeTB.java
store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/QueueTuple_1.java
Log:
QPID-1730 : Update to make the recovery of queues and exchanges perform a look up in the
broker to see if the object already exists rather than make a new one. This does mean that
any dynamic changes will be lost, but I'm not sure they are persisted anyway.
Modified:
store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/BDBMessageStore.java
===================================================================
---
store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/BDBMessageStore.java 2009-03-12
15:04:25 UTC (rev 3154)
+++
store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/BDBMessageStore.java 2009-03-13
12:22:07 UTC (rev 3155)
@@ -41,6 +41,7 @@
import org.apache.qpid.framing.AMQShortString;
import org.apache.qpid.framing.FieldTable;
import org.apache.qpid.framing.abstraction.ContentChunk;
+import org.apache.qpid.server.configuration.VirtualHostConfiguration;
import org.apache.qpid.server.exchange.Exchange;
import org.apache.qpid.server.queue.AMQQueue;
import org.apache.qpid.server.queue.AMQQueueFactory;
@@ -57,7 +58,6 @@
import org.apache.qpid.server.txn.NonTransactionalContext;
import org.apache.qpid.server.txn.TransactionalContext;
import org.apache.qpid.server.virtualhost.VirtualHost;
-import org.apache.qpid.server.configuration.VirtualHostConfiguration;
import java.io.File;
import java.util.ArrayList;
@@ -480,7 +480,7 @@
/**
* Removes the specified message from the store in the given transactional store
context.
* Internal method that is package scoped to allow testing.
- *
+ *
* @param context The transactional context to remove the message in.
* @param messageId Identifies the message to remove.
*
@@ -649,7 +649,7 @@
private void recoverExchange(Exchange exchange) throws AMQException,
DatabaseException
{
- _log.info("Recovering durable exchange " + exchange.getName() + "
of type " + exchange.getType() + "...");
+ _log.info("Recovering bindings for durable exchange:" + exchange);
QueueRegistry queueRegistry = _virtualHost.getQueueRegistry();
@@ -732,10 +732,7 @@
{
Exchange exchange = (Exchange) binding.entryToObject(value);
- _virtualHost.getExchangeRegistry().registerExchange(exchange);
-
exchanges.add(exchange);
- _log.info("Registering exchange " + exchange.getName());
}
return exchanges;
@@ -834,10 +831,18 @@
*/
public void createQueue(AMQQueue queue, FieldTable arguments) throws AMQException
{
- _log.debug("public void createQueue(AMQQueue queue(" + queue.getName()
+ ") = " + queue + "): called");
if (_state != State.RECOVERING)
{
+
+ // Check that we don't already have this queue in our map.
+ if (_queueNameToIdMap.containsKey(queue.getName()))
+ {
+ return;
+ }
+
+ _log.debug("BDBMessageStore.createQueue(AMQQueue queue(" + queue +
"," + arguments + "): called");
+
long queueId = _queueId.getAndIncrement();
_queueNameToIdMap.put(queue.getName(), queueId);
@@ -1057,7 +1062,7 @@
{
commit(tx);
context.setPayload(null);
- }
+ }
}
catch (DatabaseException e)
{
@@ -1494,13 +1499,8 @@
while (cursor.getNext(key, value, LockMode.RMW) == OperationStatus.SUCCESS)
{
AMQQueue queue = (AMQQueue) binding.entryToObject(value);
- if (queue != null)
- {
- _virtualHost.getQueueRegistry().registerQueue(queue);
- queues.put(queue.getName(), queue);
- _log.info("Recovering queue " + queue.getName() + "
with owner:"
- + ((queue.getOwner() == null) ? " <<null>>
" : (" \"" + queue.getOwner() + "\" ")));
- }
+
+ queues.put(queue.getName(), queue);
}
return queues;
@@ -1619,7 +1619,6 @@
}
}
-
private void deliverMessages(final StoreContext context, Map<AMQShortString,
AMQQueue> queues)
throws DatabaseException, AMQException
{
Modified:
store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/ExchangeTB.java
===================================================================
---
store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/ExchangeTB.java 2009-03-12
15:04:25 UTC (rev 3154)
+++
store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/ExchangeTB.java 2009-03-13
12:22:07 UTC (rev 3155)
@@ -1,21 +1,18 @@
package org.apache.qpid.server.store.berkeleydb;
-import org.apache.qpid.server.virtualhost.VirtualHost;
-import org.apache.qpid.server.exchange.Exchange;
-import org.apache.qpid.framing.AMQShortString;
-import org.apache.qpid.AMQException;
-
import com.sleepycat.bind.tuple.TupleBinding;
import com.sleepycat.bind.tuple.TupleInput;
import com.sleepycat.bind.tuple.TupleOutput;
import org.apache.log4j.Logger;
+import org.apache.qpid.AMQException;
+import org.apache.qpid.framing.AMQShortString;
+import org.apache.qpid.server.exchange.Exchange;
+import org.apache.qpid.server.virtualhost.VirtualHost;
public class ExchangeTB extends TupleBinding
{
private static final Logger _log = Logger.getLogger(ExchangeTB.class);
-
-
private final VirtualHost _virtualHost;
public ExchangeTB(VirtualHost virtualHost)
@@ -29,12 +26,28 @@
AMQShortString name = AMQShortStringEncoding.readShortString(tupleInput);
AMQShortString typeName = AMQShortStringEncoding.readShortString(tupleInput);
-
boolean autoDelete = tupleInput.readBoolean();
try
{
- return _virtualHost.getExchangeFactory().createExchange(name, typeName, true,
autoDelete, 0);
+ Exchange exchange;
+ Exchange existing = _virtualHost.getExchangeRegistry().getExchange(name);
+
+ if (existing != null)
+ {
+ _log.info("Exchange :" + existing + ": already exists in
configured broker.");
+ exchange = existing;
+ }
+ else
+ {
+ exchange = _virtualHost.getExchangeFactory().createExchange(name,
typeName, true, autoDelete, 0);
+
+ _virtualHost.getExchangeRegistry().registerExchange(exchange);
+
+ _log.info("Registering new durable exchange from BDB:" +
exchange);
+ }
+
+ return exchange;
}
catch (AMQException e)
{
@@ -47,10 +60,9 @@
{
Exchange exchange = (Exchange) object;
+ AMQShortStringEncoding.writeShortString(exchange.getName(), tupleOutput);
+ AMQShortStringEncoding.writeShortString(exchange.getType(), tupleOutput);
- AMQShortStringEncoding.writeShortString(exchange.getName(),tupleOutput);
- AMQShortStringEncoding.writeShortString(exchange.getType(),tupleOutput);
-
tupleOutput.writeBoolean(exchange.isAutoDelete());
}
Modified:
store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/QueueTuple_1.java
===================================================================
---
store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/QueueTuple_1.java 2009-03-12
15:04:25 UTC (rev 3154)
+++
store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/QueueTuple_1.java 2009-03-13
12:22:07 UTC (rev 3155)
@@ -20,14 +20,14 @@
import com.sleepycat.bind.tuple.TupleBinding;
import com.sleepycat.bind.tuple.TupleInput;
import com.sleepycat.bind.tuple.TupleOutput;
+import org.apache.log4j.Logger;
+import org.apache.qpid.AMQException;
+import org.apache.qpid.framing.AMQShortString;
+import org.apache.qpid.framing.FieldTable;
import org.apache.qpid.server.queue.AMQQueue;
import org.apache.qpid.server.queue.AMQQueueFactory;
+import org.apache.qpid.server.store.berkeleydb.AMQShortStringEncoding;
import org.apache.qpid.server.virtualhost.VirtualHost;
-import org.apache.qpid.server.store.berkeleydb.AMQShortStringEncoding;
-import org.apache.qpid.AMQException;
-import org.apache.qpid.framing.AMQShortString;
-import org.apache.qpid.framing.FieldTable;
-import org.apache.log4j.Logger;
public class QueueTuple_1 extends TupleBinding implements QueueTuple
{
@@ -76,7 +76,26 @@
{
try
{
- return AMQQueueFactory.createAMQQueueImpl(name, true, owner, false,
_virtualHost, arguments);
+
+ AMQQueue queue;
+ AMQQueue existing = _virtualHost.getQueueRegistry().getQueue(name);
+
+ if (existing != null)
+ {
+ _logger.info("Queue :" + existing + ": already exists in
configured broker.");
+
+ queue = existing;
+ }
+ else
+ {
+ // Retrieve the existing Queue object
+ queue = AMQQueueFactory.createAMQQueueImpl(name, true, owner, false,
_virtualHost, arguments);
+ _virtualHost.getQueueRegistry().registerQueue(queue);
+ _logger.info("Recovering queue " + queue.getName() + "
with owner:"
+ + ((queue.getOwner() == null) ? " <<null>>
" : (" \"" + queue.getOwner() + "\" ")));
+ }
+
+ return queue;
}
catch (AMQException e)
{