QPID-5073 : [Java Broker] Refactor DurableConfigurationStore recovery to allow for additional configured object children other than just Exchange/Binding/Queue

git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@1514360 13f79535-47bb-0310-9956-ffa450edef68

2 files changed, 36 insertions, 9 deletions

diff --git a/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/AbstractBDBMessageStore.java b/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/AbstractBDBMessageStore.java
index c2a530499a..2350e28ee2 100644
--- a/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/AbstractBDBMessageStore.java
+++ b/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/AbstractBDBMessageStore.java
@@ -29,6 +29,7 @@ import java.io.File;
 import java.lang.ref.SoftReference;
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
+import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.LinkedList;
@@ -809,31 +810,52 @@ public abstract class AbstractBDBMessageStore implements MessageStore, DurableCo
         {
             LOGGER.debug("public void remove(id = " + id + ", type="+type+"): called");
         }
-        OperationStatus status = removeConfiguredObject(id);
+        OperationStatus status = removeConfiguredObject(null, id);
         if (status == OperationStatus.NOTFOUND)
         {
             throw new AMQStoreException("Configured object of type " + type + " with id " + id + " not found");
         }
     }
 
+    @Override
+    public UUID[] removeConfiguredObjects(final UUID... objects) throws AMQStoreException
+    {
+        com.sleepycat.je.Transaction txn = _environment.beginTransaction(null, null);
+        Collection<UUID> removed = new ArrayList<UUID>(objects.length);
+        for(UUID id : objects)
+        {
+            if(removeConfiguredObject(txn, id) == OperationStatus.SUCCESS)
+            {
+                removed.add(id);
+            }
+        }
+
+        txn.commit();
+        return removed.toArray(new UUID[removed.size()]);
+    }
 
     @Override
     public void update(UUID id, String type, Map<String, Object> attributes) throws AMQStoreException
     {
-        update(id, type, attributes, null);
+        update(false, id, type, attributes, null);
     }
 
     public void update(ConfiguredObjectRecord... records) throws AMQStoreException
     {
+        update(false, records);
+    }
+
+    public void update(boolean createIfNecessary, ConfiguredObjectRecord... records) throws AMQStoreException
+    {
         com.sleepycat.je.Transaction txn = _environment.beginTransaction(null, null);
         for(ConfiguredObjectRecord record : records)
         {
-            update(record.getId(), record.getType(), record.getAttributes(), txn);
+            update(createIfNecessary, record.getId(), record.getType(), record.getAttributes(), txn);
         }
         txn.commit();
     }
 
-    private void update(UUID id, String type, Map<String, Object> attributes, com.sleepycat.je.Transaction txn) throws AMQStoreException
+    private void update(boolean createIfNecessary, UUID id, String type, Map<String, Object> attributes, com.sleepycat.je.Transaction txn) throws AMQStoreException
     {
         if (LOGGER.isDebugEnabled())
         {
@@ -851,7 +873,7 @@ public abstract class AbstractBDBMessageStore implements MessageStore, DurableCo
             ConfiguredObjectBinding configuredObjectBinding = ConfiguredObjectBinding.getInstance();
 
             OperationStatus status = _configuredObjectsDb.get(txn, key, value, LockMode.DEFAULT);
-            if (status == OperationStatus.SUCCESS)
+            if (status == OperationStatus.SUCCESS || (createIfNecessary && status == OperationStatus.NOTFOUND))
             {
                 ConfiguredObjectRecord newQueueRecord = new ConfiguredObjectRecord(id, type, attributes);
 
@@ -1406,14 +1428,14 @@ public abstract class AbstractBDBMessageStore implements MessageStore, DurableCo
         }
     }
 
-    private OperationStatus removeConfiguredObject(UUID id) throws AMQStoreException
+    private OperationStatus removeConfiguredObject(Transaction tx, UUID id) throws AMQStoreException
     {
         DatabaseEntry key = new DatabaseEntry();
         UUIDTupleBinding uuidBinding = UUIDTupleBinding.getInstance();
         uuidBinding.objectToEntry(id, key);
         try
         {
-            return _configuredObjectsDb.delete(null, key);
+            return _configuredObjectsDb.delete(tx, key);
         }
         catch (DatabaseException e)
         {
diff --git a/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/BDBHAVirtualHost.java b/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/BDBHAVirtualHost.java
index be91e4a484..b92a97c8cb 100644
--- a/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/BDBHAVirtualHost.java
+++ b/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/BDBHAVirtualHost.java
@@ -25,12 +25,14 @@ import org.apache.qpid.server.connection.IConnectionRegistry;
 import org.apache.qpid.server.logging.subjects.MessageStoreLogSubject;
 import org.apache.qpid.server.model.VirtualHost;
 import org.apache.qpid.server.stats.StatisticsGatherer;
+import org.apache.qpid.server.store.DurableConfigurationRecoverer;
 import org.apache.qpid.server.store.DurableConfigurationStore;
 import org.apache.qpid.server.store.Event;
 import org.apache.qpid.server.store.EventListener;
 import org.apache.qpid.server.store.MessageStore;
 import org.apache.qpid.server.store.OperationalLoggingListener;
 import org.apache.qpid.server.virtualhost.AbstractVirtualHost;
+import org.apache.qpid.server.virtualhost.DefaultUpgraderProvider;
 import org.apache.qpid.server.virtualhost.State;
 import org.apache.qpid.server.virtualhost.VirtualHostConfigRecoveryHandler;
 import org.apache.qpid.server.virtualhost.VirtualHostRegistry;
@@ -58,7 +60,7 @@ public class BDBHAVirtualHost extends AbstractVirtualHost
         _messageStore = new BDBHAMessageStore();
 
         final MessageStoreLogSubject storeLogSubject =
-                new MessageStoreLogSubject(this, _messageStore.getClass().getSimpleName());
+                new MessageStoreLogSubject(getName(), _messageStore.getClass().getSimpleName());
         OperationalLoggingListener.listen(_messageStore, storeLogSubject);
 
         _messageStore.addEventListener(new BeforeActivationListener(), Event.BEFORE_ACTIVATE);
@@ -71,9 +73,12 @@ public class BDBHAVirtualHost extends AbstractVirtualHost
         _messageStore.addEventListener(new BeforePassivationListener(), Event.BEFORE_PASSIVATE);
 
         VirtualHostConfigRecoveryHandler recoveryHandler = new VirtualHostConfigRecoveryHandler(this, getExchangeRegistry(), getExchangeFactory());
+        DurableConfigurationRecoverer configRecoverer =
+                new DurableConfigurationRecoverer(getName(), getDurableConfigurationRecoverers(),
+                                                  new DefaultUpgraderProvider(this, getExchangeRegistry()));
 
         _messageStore.configureConfigStore(getName(),
-                recoveryHandler,
+                configRecoverer,
                 virtualHost);
 
         _messageStore.configureMessageStore(getName(),