diff options
Diffstat (limited to 'qpid/java/bdbstore')
21 files changed, 613 insertions, 146 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 7d7ebd32fe..9a06f06999 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 @@ -41,8 +41,6 @@ import java.util.UUID; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicLong; -import org.apache.commons.configuration.Configuration; -import org.apache.commons.configuration.ConfigurationException; import org.apache.log4j.Logger; import org.apache.qpid.server.store.StoreException; import org.apache.qpid.server.message.EnqueueableMessage; @@ -51,16 +49,11 @@ import org.apache.qpid.server.queue.AMQQueue; import org.apache.qpid.server.store.*; import org.apache.qpid.server.store.MessageStoreRecoveryHandler.StoredMessageRecoveryHandler; import org.apache.qpid.server.store.TransactionLogRecoveryHandler.QueueEntryRecoveryHandler; +import org.apache.qpid.server.store.berkeleydb.entry.HierarchyKey; import org.apache.qpid.server.store.berkeleydb.entry.PreparedTransaction; import org.apache.qpid.server.store.berkeleydb.entry.QueueEntryKey; import org.apache.qpid.server.store.berkeleydb.entry.Xid; -import org.apache.qpid.server.store.berkeleydb.tuple.ConfiguredObjectBinding; -import org.apache.qpid.server.store.berkeleydb.tuple.ContentBinding; -import org.apache.qpid.server.store.berkeleydb.tuple.MessageMetaDataBinding; -import org.apache.qpid.server.store.berkeleydb.tuple.PreparedTransactionBinding; -import org.apache.qpid.server.store.berkeleydb.tuple.QueueEntryBinding; -import org.apache.qpid.server.store.berkeleydb.tuple.UUIDTupleBinding; -import org.apache.qpid.server.store.berkeleydb.tuple.XidBinding; +import org.apache.qpid.server.store.berkeleydb.tuple.*; import org.apache.qpid.server.store.berkeleydb.upgrade.Upgrader; import org.apache.qpid.util.FileUtils; @@ -70,7 +63,7 @@ public abstract class AbstractBDBMessageStore implements MessageStore, DurableCo private static final int LOCK_RETRY_ATTEMPTS = 5; - public static final int VERSION = 7; + public static final int VERSION = 8; private static final Map<String, String> ENVCONFIG_DEFAULTS = Collections.unmodifiableMap(new HashMap<String, String>() {{ @@ -83,6 +76,8 @@ public abstract class AbstractBDBMessageStore implements MessageStore, DurableCo private Environment _environment; private static String CONFIGURED_OBJECTS = "CONFIGURED_OBJECTS"; + private static String CONFIGURED_OBJECT_HIERARCHY = "CONFIGURED_OBJECT_HIERARCHY"; + private static String MESSAGEMETADATADB_NAME = "MESSAGE_METADATA"; private static String MESSAGECONTENTDB_NAME = "MESSAGE_CONTENT"; private static String DELIVERYDB_NAME = "QUEUE_ENTRIES"; @@ -92,6 +87,7 @@ public abstract class AbstractBDBMessageStore implements MessageStore, DurableCo private static String CONFIG_VERSION_DB = "CONFIG_VERSION"; private Database _configuredObjectsDb; + private Database _configuredObjectHierarchyDb; private Database _configVersionDb; private Database _messageMetaDataDb; private Database _messageContentDb; @@ -146,6 +142,7 @@ public abstract class AbstractBDBMessageStore implements MessageStore, DurableCo private Map<String, String> _envConfigMap; private VirtualHost _virtualHost; + public AbstractBDBMessageStore() { _stateManager = new StateManager(_eventManager); @@ -183,7 +180,7 @@ public abstract class AbstractBDBMessageStore implements MessageStore, DurableCo private void completeInitialisation() { - configure(_virtualHost); + configure(); _stateManager.attainState(State.INITIALISED); } @@ -221,27 +218,24 @@ public abstract class AbstractBDBMessageStore implements MessageStore, DurableCo /** * Called after instantiation in order to configure the message store. * - * - * - * @param virtualHost The virtual host using this store * @return whether a new store environment was created or not (to indicate whether recovery is necessary) * * @throws Exception If any error occurs that means the store is unable to configure itself. */ - public void configure(VirtualHost virtualHost) + public void configure() { - configure(virtualHost, _messageRecoveryHandler != null); + configure(_messageRecoveryHandler != null); } - public void configure(VirtualHost virtualHost, boolean isMessageStore) + public void configure(boolean isMessageStore) { - String name = virtualHost.getName(); + String name = _virtualHost.getName(); final String defaultPath = System.getProperty("QPID_WORK") + File.separator + "bdbstore" + File.separator + name; String storeLocation; if(isMessageStore) { - storeLocation = (String) virtualHost.getAttribute(VirtualHost.STORE_PATH); + storeLocation = (String) _virtualHost.getAttribute(VirtualHost.STORE_PATH); if(storeLocation == null) { storeLocation = defaultPath; @@ -249,15 +243,15 @@ public abstract class AbstractBDBMessageStore implements MessageStore, DurableCo } else // we are acting only as the durable config store { - storeLocation = (String) virtualHost.getAttribute(VirtualHost.CONFIG_STORE_PATH); + storeLocation = (String) _virtualHost.getAttribute(VirtualHost.CONFIG_STORE_PATH); if(storeLocation == null) { storeLocation = defaultPath; } } - Object overfullAttr = virtualHost.getAttribute(MessageStoreConstants.OVERFULL_SIZE_ATTRIBUTE); - Object underfullAttr = virtualHost.getAttribute(MessageStoreConstants.UNDERFULL_SIZE_ATTRIBUTE); + Object overfullAttr = _virtualHost.getAttribute(MessageStoreConstants.OVERFULL_SIZE_ATTRIBUTE); + Object underfullAttr = _virtualHost.getAttribute(MessageStoreConstants.UNDERFULL_SIZE_ATTRIBUTE); _persistentSizeHighThreshold = overfullAttr == null ? -1l : overfullAttr instanceof Number ? ((Number) overfullAttr).longValue() : Long.parseLong(overfullAttr.toString()); @@ -285,7 +279,7 @@ public abstract class AbstractBDBMessageStore implements MessageStore, DurableCo _envConfigMap = new HashMap<String, String>(); _envConfigMap.putAll(ENVCONFIG_DEFAULTS); - Object bdbEnvConfigAttr = virtualHost.getAttribute("bdbEnvironmentConfig"); + Object bdbEnvConfigAttr = _virtualHost.getAttribute("bdbEnvironmentConfig"); if(bdbEnvConfigAttr instanceof Map) { _envConfigMap.putAll((Map)bdbEnvConfigAttr); @@ -293,27 +287,7 @@ public abstract class AbstractBDBMessageStore implements MessageStore, DurableCo LOGGER.info("Configuring BDB message store"); - setupStore(environmentPath, name); - } - - protected Map<String,String> getConfigMap(Map<String, String> defaultConfig, Configuration config, String prefix) throws ConfigurationException - { - final List<Object> argumentNames = config.getList(prefix + ".name"); - final List<Object> argumentValues = config.getList(prefix + ".value"); - final int initialSize = argumentNames.size() + defaultConfig.size(); - - final Map<String,String> attributes = new HashMap<String,String>(initialSize); - attributes.putAll(defaultConfig); - - for (int i = 0; i < argumentNames.size(); i++) - { - final String argName = argumentNames.get(i).toString(); - final String argValue = argumentValues.get(i).toString(); - - attributes.put(argName, argValue); - } - - return Collections.unmodifiableMap(attributes); + setupStore(environmentPath); } @Override @@ -337,11 +311,11 @@ public abstract class AbstractBDBMessageStore implements MessageStore, DurableCo _stateManager.attainState(State.ACTIVE); } - protected void setupStore(File storePath, String name) + protected void setupStore(File storePath) { _environment = createEnvironment(storePath); - new Upgrader(_environment, name).upgradeIfNecessary(); + new Upgrader(_environment, _virtualHost).upgradeIfNecessary(); openDatabases(); @@ -355,6 +329,11 @@ public abstract class AbstractBDBMessageStore implements MessageStore, DurableCo return _environment; } + protected final VirtualHost getVirtualHost() + { + return _virtualHost; + } + private void openDatabases() throws DatabaseException { DatabaseConfig dbConfig = new DatabaseConfig(); @@ -365,6 +344,7 @@ public abstract class AbstractBDBMessageStore implements MessageStore, DurableCo dbConfig.setReadOnly(false); _configuredObjectsDb = openDatabase(CONFIGURED_OBJECTS, dbConfig); + _configuredObjectHierarchyDb = openDatabase(CONFIGURED_OBJECT_HIERARCHY, dbConfig); _configVersionDb = openDatabase(CONFIG_VERSION_DB, dbConfig); _messageMetaDataDb = openDatabase(MESSAGEMETADATADB_NAME, dbConfig); _messageContentDb = openDatabase(MESSAGECONTENTDB_NAME, dbConfig); @@ -411,11 +391,17 @@ public abstract class AbstractBDBMessageStore implements MessageStore, DurableCo _messageContentDb.close(); } - if (_configuredObjectsDb != null) - { - LOGGER.info("Closing configurable objects database"); - _configuredObjectsDb.close(); - } + if (_configuredObjectsDb != null) + { + LOGGER.info("Closing configurable objects database"); + _configuredObjectsDb.close(); + } + + if (_configuredObjectHierarchyDb != null) + { + LOGGER.info("Closing configurable object hierarchy database"); + _configuredObjectHierarchyDb.close(); + } if (_deliveryDb != null) { @@ -555,24 +541,60 @@ public abstract class AbstractBDBMessageStore implements MessageStore, DurableCo private void loadConfiguredObjects(ConfigurationRecoveryHandler crh) throws DatabaseException { - Cursor cursor = null; + Cursor objectsCursor = null; + Cursor hierarchyCursor = null; try { - cursor = _configuredObjectsDb.openCursor(null, null); + objectsCursor = _configuredObjectsDb.openCursor(null, null); DatabaseEntry key = new DatabaseEntry(); DatabaseEntry value = new DatabaseEntry(); - while (cursor.getNext(key, value, LockMode.RMW) == OperationStatus.SUCCESS) + + Map<UUID, BDBConfiguredObjectRecord> configuredObjects = + new HashMap<UUID, BDBConfiguredObjectRecord>(); + + while (objectsCursor.getNext(key, value, LockMode.RMW) == OperationStatus.SUCCESS) { UUID id = UUIDTupleBinding.getInstance().entryToObject(key); - ConfiguredObjectRecord configuredObject = new ConfiguredObjectBinding(id).entryToObject(value); - crh.configuredObject(configuredObject.getId(),configuredObject.getType(),configuredObject.getAttributes()); + BDBConfiguredObjectRecord configuredObject = + (BDBConfiguredObjectRecord) new ConfiguredObjectBinding(id).entryToObject(value); + configuredObjects.put(configuredObject.getId(), configuredObject); } + // set parents + hierarchyCursor = _configuredObjectHierarchyDb.openCursor(null, null); + while (hierarchyCursor.getNext(key, value, LockMode.RMW) == OperationStatus.SUCCESS) + { + HierarchyKey hk = HierarchyKeyBinding.getInstance().entryToObject(key); + UUID parentId = UUIDTupleBinding.getInstance().entryToObject(value); + BDBConfiguredObjectRecord child = configuredObjects.get(hk.getChildId()); + if(child != null) + { + ConfiguredObjectRecord parent = configuredObjects.get(parentId); + if(parent != null) + { + child.addParent(hk.getParentType(), parent); + } + else if(hk.getParentType().equals("Exchange")) + { + // TODO - remove this hack for the pre-defined exchanges + child.addParent(hk.getParentType(), new BDBConfiguredObjectRecord(parentId, "Exchange", Collections.<String,Object>emptyMap())); + } + } + } + + for (ConfiguredObjectRecord record : configuredObjects.values()) + { + crh.configuredObject(record); + } + + + } finally { - closeCursorSafely(cursor); + closeCursorSafely(objectsCursor); + closeCursorSafely(hierarchyCursor); } } @@ -842,39 +864,28 @@ public abstract class AbstractBDBMessageStore implements MessageStore, DurableCo } @Override - public void create(UUID id, String type, Map<String, Object> attributes) throws StoreException + public void create(ConfiguredObjectRecord configuredObject) throws StoreException { + if (_stateManager.isInState(State.ACTIVE)) { - ConfiguredObjectRecord configuredObject = new ConfiguredObjectRecord(id, type, attributes); - storeConfiguredObjectEntry(configuredObject); - } - } + com.sleepycat.je.Transaction txn = _environment.beginTransaction(null, null); - @Override - public void remove(UUID id, String type) throws StoreException - { - if (LOGGER.isDebugEnabled()) - { - LOGGER.debug("public void remove(id = " + id + ", type="+type+"): called"); - } - OperationStatus status = removeConfiguredObject(null, id); - if (status == OperationStatus.NOTFOUND) - { - throw new StoreException("Configured object of type " + type + " with id " + id + " not found"); + storeConfiguredObjectEntry(txn, configuredObject); + + txn.commit(); } } - @Override - public UUID[] removeConfiguredObjects(final UUID... objects) throws StoreException + public UUID[] remove(final ConfiguredObjectRecord... objects) throws StoreException { com.sleepycat.je.Transaction txn = _environment.beginTransaction(null, null); Collection<UUID> removed = new ArrayList<UUID>(objects.length); - for(UUID id : objects) + for(ConfiguredObjectRecord record : objects) { - if(removeConfiguredObject(txn, id) == OperationStatus.SUCCESS) + if(removeConfiguredObject(txn, record) == OperationStatus.SUCCESS) { - removed.add(id); + removed.add(record.getId()); } } @@ -882,52 +893,50 @@ public abstract class AbstractBDBMessageStore implements MessageStore, DurableCo return removed.toArray(new UUID[removed.size()]); } - @Override - public void update(UUID id, String type, Map<String, Object> attributes) throws StoreException - { - update(false, id, type, attributes, null); - } - public void update(boolean createIfNecessary, ConfiguredObjectRecord... records) throws StoreException { com.sleepycat.je.Transaction txn = _environment.beginTransaction(null, null); for(ConfiguredObjectRecord record : records) { - update(createIfNecessary, record.getId(), record.getType(), record.getAttributes(), txn); + update(createIfNecessary, record, txn); } txn.commit(); } - private void update(boolean createIfNecessary, UUID id, String type, Map<String, Object> attributes, com.sleepycat.je.Transaction txn) throws + private void update(boolean createIfNecessary, ConfiguredObjectRecord record, com.sleepycat.je.Transaction txn) throws StoreException { if (LOGGER.isDebugEnabled()) { - LOGGER.debug("Updating " +type + ", id: " + id); + LOGGER.debug("Updating " + record.getType() + ", id: " + record.getId()); } try { DatabaseEntry key = new DatabaseEntry(); UUIDTupleBinding keyBinding = UUIDTupleBinding.getInstance(); - keyBinding.objectToEntry(id, key); + keyBinding.objectToEntry(record.getId(), key); DatabaseEntry value = new DatabaseEntry(); DatabaseEntry newValue = new DatabaseEntry(); ConfiguredObjectBinding configuredObjectBinding = ConfiguredObjectBinding.getInstance(); OperationStatus status = _configuredObjectsDb.get(txn, key, value, LockMode.DEFAULT); - if (status == OperationStatus.SUCCESS || (createIfNecessary && status == OperationStatus.NOTFOUND)) + final boolean isNewRecord = status == OperationStatus.NOTFOUND; + if (status == OperationStatus.SUCCESS || (createIfNecessary && isNewRecord)) { - ConfiguredObjectRecord newQueueRecord = new ConfiguredObjectRecord(id, type, attributes); // write the updated entry to the store - configuredObjectBinding.objectToEntry(newQueueRecord, newValue); + configuredObjectBinding.objectToEntry(record, newValue); status = _configuredObjectsDb.put(txn, key, newValue); if (status != OperationStatus.SUCCESS) { throw new StoreException("Error updating queue details within the store: " + status); } + if(isNewRecord) + { + writeHierarchyRecords(txn, record); + } } else if (status != OperationStatus.NOTFOUND) { @@ -1441,17 +1450,19 @@ public abstract class AbstractBDBMessageStore implements MessageStore, DurableCo /** * Makes the specified configured object persistent. * + * + * @param txn * @param configuredObject Details of the configured object to store. * @throws org.apache.qpid.server.store.StoreException If the operation fails for any reason. */ - private void storeConfiguredObjectEntry(ConfiguredObjectRecord configuredObject) throws StoreException + private void storeConfiguredObjectEntry(final Transaction txn, ConfiguredObjectRecord configuredObject) throws StoreException { if (_stateManager.isInState(State.ACTIVE)) { LOGGER.debug("Storing configured object: " + configuredObject); DatabaseEntry key = new DatabaseEntry(); - UUIDTupleBinding keyBinding = UUIDTupleBinding.getInstance(); - keyBinding.objectToEntry(configuredObject.getId(), key); + UUIDTupleBinding uuidBinding = UUIDTupleBinding.getInstance(); + uuidBinding.objectToEntry(configuredObject.getId(), key); DatabaseEntry value = new DatabaseEntry(); ConfiguredObjectBinding queueBinding = ConfiguredObjectBinding.getInstance(); @@ -1459,12 +1470,13 @@ public abstract class AbstractBDBMessageStore implements MessageStore, DurableCo queueBinding.objectToEntry(configuredObject, value); try { - OperationStatus status = _configuredObjectsDb.put(null, key, value); + OperationStatus status = _configuredObjectsDb.put(txn, key, value); if (status != OperationStatus.SUCCESS) { throw new StoreException("Error writing configured object " + configuredObject + " to database: " + status); } + writeHierarchyRecords(txn, configuredObject); } catch (DatabaseException e) { @@ -1474,16 +1486,49 @@ public abstract class AbstractBDBMessageStore implements MessageStore, DurableCo } } - private OperationStatus removeConfiguredObject(Transaction tx, UUID id) throws StoreException + private void writeHierarchyRecords(final Transaction txn, final ConfiguredObjectRecord configuredObject) { + OperationStatus status; + HierarchyKeyBinding hierarchyBinding = HierarchyKeyBinding.getInstance(); + DatabaseEntry hierarchyKey = new DatabaseEntry(); + DatabaseEntry hierarchyValue = new DatabaseEntry(); + + for(Map.Entry<String, ConfiguredObjectRecord> parent : configuredObject.getParents().entrySet()) + { + + hierarchyBinding.objectToEntry(new HierarchyKey(configuredObject.getId(), parent.getKey()), hierarchyKey); + UUIDTupleBinding.getInstance().objectToEntry(parent.getValue().getId(), hierarchyValue); + status = _configuredObjectHierarchyDb.put(txn, hierarchyKey, hierarchyValue); + if (status != OperationStatus.SUCCESS) + { + throw new StoreException("Error writing configured object " + configuredObject + " parent record to database: " + + status); + } + } + } + private OperationStatus removeConfiguredObject(Transaction tx, ConfiguredObjectRecord record) throws StoreException + { + UUID id = record.getId(); + Map<String, ConfiguredObjectRecord> parents = record.getParents(); LOGGER.debug("Removing configured object: " + id); DatabaseEntry key = new DatabaseEntry(); UUIDTupleBinding uuidBinding = UUIDTupleBinding.getInstance(); uuidBinding.objectToEntry(id, key); try { - return _configuredObjectsDb.delete(tx, key); + OperationStatus status = _configuredObjectsDb.delete(tx, key); + if(status == OperationStatus.SUCCESS) + { + for(String parentType : parents.keySet()) + { + DatabaseEntry hierarchyKey = new DatabaseEntry(); + HierarchyKeyBinding keyBinding = HierarchyKeyBinding.getInstance(); + keyBinding.objectToEntry(new HierarchyKey(record.getId(), parentType), hierarchyKey); + _configuredObjectHierarchyDb.delete(tx, hierarchyKey); + } + } + return status; } catch (DatabaseException e) { @@ -1859,4 +1904,10 @@ public abstract class AbstractBDBMessageStore implements MessageStore, DurableCo } } } + + void setVirtualHost(final VirtualHost virtualHost) + { + _virtualHost = virtualHost; + } + } diff --git a/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/BDBConfiguredObjectRecord.java b/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/BDBConfiguredObjectRecord.java new file mode 100644 index 0000000000..f13e4dd08b --- /dev/null +++ b/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/BDBConfiguredObjectRecord.java @@ -0,0 +1,107 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + * + */ +package org.apache.qpid.server.store.berkeleydb; + +import org.apache.qpid.server.store.ConfiguredObjectRecord; + +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; +import java.util.UUID; + +public class BDBConfiguredObjectRecord implements ConfiguredObjectRecord +{ + private final UUID _id; + private final String _type; + private final Map<String,Object> _attributes; + private Map<String, ConfiguredObjectRecord> _parents = new HashMap<String, ConfiguredObjectRecord>(); + + public BDBConfiguredObjectRecord(final UUID id, final String type, final Map<String, Object> attributes) + { + _id = id; + _type = type; + _attributes = Collections.unmodifiableMap(attributes); + } + + public UUID getId() + { + return _id; + } + + public String getType() + { + return _type; + } + + public Map<String, Object> getAttributes() + { + return _attributes; + } + + void addParent(String parentType, ConfiguredObjectRecord parent) + { + _parents.put(parentType, parent); + } + + @Override + public Map<String, ConfiguredObjectRecord> getParents() + { + return Collections.unmodifiableMap(_parents); + } + + @Override + public boolean equals(final Object o) + { + if (this == o) + { + return true; + } + if (o == null || getClass() != o.getClass()) + { + return false; + } + + final BDBConfiguredObjectRecord that = (BDBConfiguredObjectRecord) o; + + if (_attributes != null ? !_attributes.equals(that._attributes) : that._attributes != null) + { + return false; + } + if (_id != null ? !_id.equals(that._id) : that._id != null) + { + return false; + } + if (_type != null ? !_type.equals(that._type) : that._type != null) + { + return false; + } + + return true; + } + + @Override + public int hashCode() + { + int result = _id != null ? _id.hashCode() : 0; + result = 31 * result + (_type != null ? _type.hashCode() : 0); + return result; + } +} diff --git a/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/BDBHAMessageStore.java b/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/BDBHAMessageStore.java index d99733acf0..bdd48e99bb 100644 --- a/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/BDBHAMessageStore.java +++ b/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/BDBHAMessageStore.java @@ -37,7 +37,6 @@ import org.apache.commons.configuration.Configuration; import org.apache.commons.configuration.ConfigurationException; import org.apache.log4j.Logger; import org.apache.qpid.server.security.SecurityManager; -import org.apache.qpid.server.security.auth.TaskPrincipal; import org.apache.qpid.server.store.StoreException; import org.apache.qpid.server.model.VirtualHost; import org.apache.qpid.server.store.HAMessageStore; @@ -124,9 +123,10 @@ public class BDBHAMessageStore extends AbstractBDBMessageStore implements HAMess private Map<String, String> _repConfig; @Override - public void configure(VirtualHost virtualHost) + public void configure() { //Mandatory configuration + VirtualHost virtualHost = getVirtualHost(); _groupName = getValidatedStringAttribute(virtualHost, "haGroupName"); _nodeName = getValidatedStringAttribute(virtualHost, "haNodeName"); _nodeHostPort = getValidatedStringAttribute(virtualHost, "haNodeAddress"); @@ -159,7 +159,7 @@ public class BDBHAMessageStore extends AbstractBDBMessageStore implements HAMess + "! Please set highAvailability.coalescingSync to false in store configuration."); } - super.configure(virtualHost); + super.configure(); } @@ -207,13 +207,13 @@ public class BDBHAMessageStore extends AbstractBDBMessageStore implements HAMess @Override - protected void setupStore(File storePath, String name) throws DatabaseException + protected void setupStore(File storePath) throws DatabaseException { - super.setupStore(storePath, name); + super.setupStore(storePath); if(_coalescingSync) { - _commitThreadWrapper = new CommitThreadWrapper("Commit-Thread-" + name, getEnvironment()); + _commitThreadWrapper = new CommitThreadWrapper("Commit-Thread-" + getVirtualHost().getName(), getEnvironment()); _commitThreadWrapper.startCommitThread(); } } diff --git a/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/BDBMessageStore.java b/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/BDBMessageStore.java index acff8e2b21..0fd5060397 100644 --- a/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/BDBMessageStore.java +++ b/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/BDBMessageStore.java @@ -46,11 +46,11 @@ public class BDBMessageStore extends AbstractBDBMessageStore private CommitThreadWrapper _commitThreadWrapper; @Override - protected void setupStore(File storePath, String name) throws DatabaseException + protected void setupStore(File storePath) throws DatabaseException { - super.setupStore(storePath, name); + super.setupStore(storePath); - _commitThreadWrapper = new CommitThreadWrapper("Commit-Thread-" + name, getEnvironment()); + _commitThreadWrapper = new CommitThreadWrapper("Commit-Thread-" + getVirtualHost().getName(), getEnvironment()); _commitThreadWrapper.startCommitThread(); } diff --git a/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/entry/HierarchyKey.java b/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/entry/HierarchyKey.java new file mode 100644 index 0000000000..d1c341447e --- /dev/null +++ b/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/entry/HierarchyKey.java @@ -0,0 +1,79 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + * + */ +package org.apache.qpid.server.store.berkeleydb.entry; + +import java.util.UUID; + +public class HierarchyKey +{ + private final UUID _childId; + private final String _parentType; + + public HierarchyKey(final UUID childId, final String parentType) + { + _childId = childId; + _parentType = parentType; + } + + public UUID getChildId() + { + return _childId; + } + + public String getParentType() + { + return _parentType; + } + + @Override + public boolean equals(final Object o) + { + if (this == o) + { + return true; + } + if (o == null || getClass() != o.getClass()) + { + return false; + } + + final HierarchyKey that = (HierarchyKey) o; + + if (!_childId.equals(that._childId)) + { + return false; + } + if (!_parentType.equals(that._parentType)) + { + return false; + } + + return true; + } + + @Override + public int hashCode() + { + int result = _childId.hashCode(); + result = 31 * result + _parentType.hashCode(); + return result; + } +} diff --git a/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuple/ConfiguredObjectBinding.java b/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuple/ConfiguredObjectBinding.java index bc3beeb78b..38a2215fe7 100644 --- a/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuple/ConfiguredObjectBinding.java +++ b/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuple/ConfiguredObjectBinding.java @@ -32,6 +32,7 @@ import com.sleepycat.bind.tuple.TupleBinding; import com.sleepycat.bind.tuple.TupleInput; import com.sleepycat.bind.tuple.TupleOutput; import org.apache.qpid.server.store.StoreException; +import org.apache.qpid.server.store.berkeleydb.BDBConfiguredObjectRecord; import org.codehaus.jackson.JsonGenerationException; import org.codehaus.jackson.JsonGenerator; import org.codehaus.jackson.JsonProcessingException; @@ -80,7 +81,7 @@ public class ConfiguredObjectBinding extends TupleBinding<ConfiguredObjectRecord _uuid = uuid; } - public ConfiguredObjectRecord entryToObject(TupleInput tupleInput) + public BDBConfiguredObjectRecord entryToObject(TupleInput tupleInput) { String type = tupleInput.readString(); String json = tupleInput.readString(); @@ -88,7 +89,7 @@ public class ConfiguredObjectBinding extends TupleBinding<ConfiguredObjectRecord try { Map<String,Object> value = mapper.readValue(json, Map.class); - ConfiguredObjectRecord configuredObject = new ConfiguredObjectRecord(_uuid, type, value); + BDBConfiguredObjectRecord configuredObject = new BDBConfiguredObjectRecord(_uuid, type, value); return configuredObject; } catch (IOException e) diff --git a/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuple/HierarchyKeyBinding.java b/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuple/HierarchyKeyBinding.java new file mode 100644 index 0000000000..13adaabfc8 --- /dev/null +++ b/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuple/HierarchyKeyBinding.java @@ -0,0 +1,59 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + * + */ +package org.apache.qpid.server.store.berkeleydb.tuple; + +import com.sleepycat.bind.tuple.TupleBinding; +import com.sleepycat.bind.tuple.TupleInput; +import com.sleepycat.bind.tuple.TupleOutput; +import org.apache.qpid.server.store.berkeleydb.entry.HierarchyKey; +import org.apache.qpid.server.store.berkeleydb.entry.QueueEntryKey; + +import java.util.UUID; + +public class HierarchyKeyBinding extends TupleBinding<HierarchyKey> +{ + + private static final HierarchyKeyBinding INSTANCE = new HierarchyKeyBinding(); + + public static HierarchyKeyBinding getInstance() + { + return INSTANCE; + } + + /** private constructor forces getInstance instead */ + private HierarchyKeyBinding() { } + + public HierarchyKey entryToObject(TupleInput tupleInput) + { + UUID childId = new UUID(tupleInput.readLong(), tupleInput.readLong()); + String parentType = tupleInput.readString(); + + return new HierarchyKey(childId, parentType); + } + + public void objectToEntry(HierarchyKey hk, TupleOutput tupleOutput) + { + UUID uuid = hk.getChildId(); + tupleOutput.writeLong(uuid.getMostSignificantBits()); + tupleOutput.writeLong(uuid.getLeastSignificantBits()); + tupleOutput.writeString(hk.getParentType()); + } +}
\ No newline at end of file diff --git a/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/StoreUpgrade.java b/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/StoreUpgrade.java index adcaef35ef..b06b6d533b 100644 --- a/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/StoreUpgrade.java +++ b/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/StoreUpgrade.java @@ -21,8 +21,9 @@ package org.apache.qpid.server.store.berkeleydb.upgrade; import com.sleepycat.je.Environment; +import org.apache.qpid.server.model.VirtualHost; public interface StoreUpgrade { - void performUpgrade(Environment environment, UpgradeInteractionHandler handler, String virtualHostName); + void performUpgrade(Environment environment, UpgradeInteractionHandler handler, VirtualHost virtualHost); } diff --git a/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgradeFrom4To5.java b/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgradeFrom4To5.java index 87f8afde4a..3eac47c81b 100644 --- a/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgradeFrom4To5.java +++ b/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgradeFrom4To5.java @@ -39,6 +39,7 @@ import org.apache.qpid.framing.AMQShortString; import org.apache.qpid.framing.ContentHeaderBody; import org.apache.qpid.framing.FieldTable; import org.apache.qpid.framing.abstraction.MessagePublishInfo; +import org.apache.qpid.server.model.VirtualHost; import org.apache.qpid.server.protocol.v0_8.MessageMetaData; import org.apache.qpid.server.store.StoreException; import org.apache.qpid.server.store.StorableMessageMetaData; @@ -74,7 +75,7 @@ public class UpgradeFrom4To5 extends AbstractStoreUpgrade private static final Logger _logger = Logger.getLogger(UpgradeFrom4To5.class); - public void performUpgrade(final Environment environment, final UpgradeInteractionHandler handler, String virtualHostName) + public void performUpgrade(final Environment environment, final UpgradeInteractionHandler handler, VirtualHost virtualHost) { Transaction transaction = null; reportStarting(environment, 4); diff --git a/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgradeFrom5To6.java b/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgradeFrom5To6.java index 46f2afd741..dea8421a33 100644 --- a/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgradeFrom5To6.java +++ b/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgradeFrom5To6.java @@ -44,6 +44,7 @@ import org.apache.qpid.server.model.Exchange; import org.apache.qpid.server.model.LifetimePolicy; import org.apache.qpid.server.model.Queue; import org.apache.qpid.server.model.UUIDGenerator; +import org.apache.qpid.server.model.VirtualHost; import org.apache.qpid.server.queue.QueueArgumentsConverter; import org.apache.qpid.server.store.StoreException; import org.apache.qpid.server.store.berkeleydb.AMQShortStringEncoding; @@ -118,11 +119,11 @@ public class UpgradeFrom5To6 extends AbstractStoreUpgrade * Queue, Exchange, Bindings entries are stored now as configurable objects * in "CONFIGURED_OBJECTS" table. */ - public void performUpgrade(final Environment environment, final UpgradeInteractionHandler handler, String virtualHostName) + public void performUpgrade(final Environment environment, final UpgradeInteractionHandler handler, VirtualHost virtualHost) { reportStarting(environment, 5); upgradeMessages(environment, handler); - upgradeConfiguredObjectsAndDependencies(environment, handler, virtualHostName); + upgradeConfiguredObjectsAndDependencies(environment, handler, virtualHost.getName()); renameDatabases(environment, null); reportFinished(environment, 6); } diff --git a/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgradeFrom6To7.java b/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgradeFrom6To7.java index ce00fd1a48..79314ae098 100644 --- a/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgradeFrom6To7.java +++ b/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgradeFrom6To7.java @@ -27,6 +27,7 @@ import com.sleepycat.je.DatabaseConfig; import com.sleepycat.je.DatabaseEntry; import com.sleepycat.je.Environment; import com.sleepycat.je.OperationStatus; +import org.apache.qpid.server.model.VirtualHost; import org.apache.qpid.server.store.StoreException; public class UpgradeFrom6To7 extends AbstractStoreUpgrade @@ -35,7 +36,7 @@ public class UpgradeFrom6To7 extends AbstractStoreUpgrade private static final int DEFAULT_CONFIG_VERSION = 0; @Override - public void performUpgrade(Environment environment, UpgradeInteractionHandler handler, String virtualHostName) + public void performUpgrade(Environment environment, UpgradeInteractionHandler handler, VirtualHost virtualHost) { reportStarting(environment, 6); DatabaseConfig dbConfig = new DatabaseConfig(); diff --git a/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgradeFrom7To8.java b/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgradeFrom7To8.java new file mode 100644 index 0000000000..3756c11d0c --- /dev/null +++ b/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgradeFrom7To8.java @@ -0,0 +1,159 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + * + */ +package org.apache.qpid.server.store.berkeleydb.upgrade; + +import com.sleepycat.bind.tuple.ByteBinding; +import com.sleepycat.bind.tuple.IntegerBinding; +import com.sleepycat.bind.tuple.TupleBinding; +import com.sleepycat.bind.tuple.TupleInput; +import com.sleepycat.bind.tuple.TupleOutput; +import com.sleepycat.je.*; +import org.apache.qpid.server.model.VirtualHost; +import org.apache.qpid.server.store.ConfiguredObjectRecord; +import org.apache.qpid.server.store.StoreException; +import org.apache.qpid.server.store.berkeleydb.BDBConfiguredObjectRecord; +import org.apache.qpid.server.store.berkeleydb.entry.HierarchyKey; +import org.apache.qpid.server.store.berkeleydb.tuple.ConfiguredObjectBinding; +import org.apache.qpid.server.store.berkeleydb.tuple.HierarchyKeyBinding; +import org.apache.qpid.server.store.berkeleydb.tuple.UUIDTupleBinding; +import org.codehaus.jackson.map.ObjectMapper; + +import java.io.IOException; +import java.io.StringWriter; +import java.util.HashMap; +import java.util.Map; +import java.util.UUID; + +public class UpgradeFrom7To8 extends AbstractStoreUpgrade +{ + + @Override + public void performUpgrade(Environment environment, UpgradeInteractionHandler handler, VirtualHost virtualHost) + { + reportStarting(environment, 7); + + DatabaseConfig dbConfig = new DatabaseConfig(); + dbConfig.setTransactional(true); + dbConfig.setAllowCreate(true); + + Database hierarchyDb = environment.openDatabase(null, "CONFIGURED_OBJECT_HIERARCHY", dbConfig); + Database configuredObjectsDb = environment.openDatabase(null, "CONFIGURED_OBJECTS", dbConfig); + + Cursor objectsCursor = null; + + Transaction txn = environment.beginTransaction(null, null); + + try + { + objectsCursor = configuredObjectsDb.openCursor(txn, null); + DatabaseEntry key = new DatabaseEntry(); + DatabaseEntry value = new DatabaseEntry(); + + Map<UUID, BDBConfiguredObjectRecord> configuredObjects = + new HashMap<UUID, BDBConfiguredObjectRecord>(); + + while (objectsCursor.getNext(key, value, LockMode.RMW) == OperationStatus.SUCCESS) + { + UUID id = UUIDTupleBinding.getInstance().entryToObject(key); + TupleInput input = TupleBinding.entryToInput(value); + String type = input.readString(); + + if(!type.endsWith("Binding")) + { + UUIDTupleBinding.getInstance().objectToEntry(virtualHost.getId(),value); + TupleOutput tupleOutput = new TupleOutput(); + tupleOutput.writeLong(id.getMostSignificantBits()); + tupleOutput.writeLong(id.getLeastSignificantBits()); + tupleOutput.writeString("VirtualHost"); + TupleBinding.outputToEntry(tupleOutput, key); + hierarchyDb.put(txn, key, value); + } + else + { + String json = input.readString(); + ObjectMapper mapper = new ObjectMapper(); + try + { + DatabaseEntry hierarchyKey = new DatabaseEntry(); + DatabaseEntry hierarchyValue = new DatabaseEntry(); + + Map<String,Object> attributes = mapper.readValue(json, Map.class); + Object queueIdString = attributes.remove("queue"); + if(queueIdString instanceof String) + { + UUID queueId = UUID.fromString(queueIdString.toString()); + UUIDTupleBinding.getInstance().objectToEntry(queueId,hierarchyValue); + TupleOutput tupleOutput = new TupleOutput(); + tupleOutput.writeLong(id.getMostSignificantBits()); + tupleOutput.writeLong(id.getLeastSignificantBits()); + tupleOutput.writeString("Queue"); + TupleBinding.outputToEntry(tupleOutput, hierarchyKey); + hierarchyDb.put(txn, hierarchyKey, hierarchyValue); + } + Object exchangeIdString = attributes.remove("exchange"); + if(exchangeIdString instanceof String) + { + UUID exchangeId = UUID.fromString(exchangeIdString.toString()); + UUIDTupleBinding.getInstance().objectToEntry(exchangeId,hierarchyValue); + TupleOutput tupleOutput = new TupleOutput(); + tupleOutput.writeLong(id.getMostSignificantBits()); + tupleOutput.writeLong(id.getLeastSignificantBits()); + tupleOutput.writeString("Exchange"); + TupleBinding.outputToEntry(tupleOutput, hierarchyKey); + hierarchyDb.put(txn, hierarchyKey, hierarchyValue); + } + TupleOutput tupleOutput = new TupleOutput(); + tupleOutput.writeString(type); + StringWriter writer = new StringWriter(); + mapper.writeValue(writer,attributes); + tupleOutput.writeString(writer.getBuffer().toString()); + TupleBinding.outputToEntry(tupleOutput, value); + objectsCursor.putCurrent(value); + } + catch (IOException e) + { + throw new StoreException(e); + } + + } + + + } + + + } + finally + { + if(objectsCursor != null) + { + objectsCursor.close(); + } + } + txn.commit(); + + hierarchyDb.close(); + configuredObjectsDb.close(); + + + + reportFinished(environment, 8); + } +} diff --git a/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/Upgrader.java b/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/Upgrader.java index 4d536a2f95..e960518719 100644 --- a/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/Upgrader.java +++ b/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/Upgrader.java @@ -24,6 +24,7 @@ import com.sleepycat.je.Cursor; import java.lang.reflect.Constructor; import java.lang.reflect.InvocationTargetException; +import org.apache.qpid.server.model.VirtualHost; import org.apache.qpid.server.store.StoreException; import org.apache.qpid.server.store.berkeleydb.AbstractBDBMessageStore; @@ -41,12 +42,12 @@ public class Upgrader static final String VERSION_DB_NAME = "DB_VERSION"; private Environment _environment; - private String _virtualHostName; + private VirtualHost _virtualHost; - public Upgrader(Environment environment, String virtualHostName) + public Upgrader(Environment environment, VirtualHost virtualHost) { _environment = environment; - _virtualHostName = virtualHostName; + _virtualHost = virtualHost; } public void upgradeIfNecessary() @@ -146,7 +147,7 @@ public class Upgrader + "UpgradeFrom"+fromVersion+"To"+toVersion); Constructor<StoreUpgrade> ctr = upgradeClass.getConstructor(); StoreUpgrade upgrade = ctr.newInstance(); - upgrade.performUpgrade(_environment, UpgradeInteractionHandler.DEFAULT_HANDLER, _virtualHostName); + upgrade.performUpgrade(_environment, UpgradeInteractionHandler.DEFAULT_HANDLER, _virtualHost); } catch (ClassNotFoundException e) { diff --git a/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/HAMessageStoreSmokeTest.java b/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/HAMessageStoreSmokeTest.java index 7f7b65f315..6a42b2691f 100644 --- a/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/HAMessageStoreSmokeTest.java +++ b/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/HAMessageStoreSmokeTest.java @@ -19,7 +19,6 @@ */ package org.apache.qpid.server.store.berkeleydb; -import org.apache.commons.configuration.ConfigurationException; import org.apache.qpid.server.model.VirtualHost; import org.apache.qpid.server.util.ServerScopedRuntimeException; import org.apache.qpid.test.utils.QpidTestCase; @@ -34,7 +33,8 @@ public class HAMessageStoreSmokeTest extends QpidTestCase { try { - _store.configure(mock(VirtualHost.class)); + _store.setVirtualHost(mock(VirtualHost.class)); + _store.configure(); fail("Expected an exception to be thrown"); } catch (ServerScopedRuntimeException ce) diff --git a/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/tuple/ConfiguredObjectBindingTest.java b/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/tuple/ConfiguredObjectBindingTest.java index 5a5d39081c..965cad1cb5 100644 --- a/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/tuple/ConfiguredObjectBindingTest.java +++ b/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/tuple/ConfiguredObjectBindingTest.java @@ -29,6 +29,7 @@ import org.apache.qpid.server.store.ConfiguredObjectRecord; import com.sleepycat.bind.tuple.TupleInput; import com.sleepycat.bind.tuple.TupleOutput; +import org.apache.qpid.server.store.ConfiguredObjectRecordImpl; public class ConfiguredObjectBindingTest extends TestCase { @@ -46,7 +47,7 @@ public class ConfiguredObjectBindingTest extends TestCase { super.setUp(); _configuredObjectBinding = ConfiguredObjectBinding.getInstance(); - _object = new ConfiguredObjectRecord(UUIDGenerator.generateRandomUUID(), DUMMY_TYPE_STRING, + _object = new ConfiguredObjectRecordImpl(UUIDGenerator.generateRandomUUID(), DUMMY_TYPE_STRING, DUMMY_ATTRIBUTES_MAP); } diff --git a/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/upgrade/AbstractUpgradeTestCase.java b/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/upgrade/AbstractUpgradeTestCase.java index b2b28b3c2d..ce143aba1b 100644 --- a/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/upgrade/AbstractUpgradeTestCase.java +++ b/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/upgrade/AbstractUpgradeTestCase.java @@ -25,12 +25,16 @@ import static org.apache.qpid.server.store.berkeleydb.BDBStoreUpgradeTestPrepare import static org.apache.qpid.server.store.berkeleydb.BDBStoreUpgradeTestPreparer.PRIORITY_QUEUE_NAME; import static org.apache.qpid.server.store.berkeleydb.BDBStoreUpgradeTestPreparer.QUEUE_NAME; import static org.apache.qpid.server.store.berkeleydb.BDBStoreUpgradeTestPreparer.QUEUE_WITH_DLQ_NAME; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; import java.io.File; import java.io.InputStream; +import java.util.UUID; import org.apache.qpid.server.logging.LogSubject; import org.apache.qpid.server.logging.subjects.TestBlankSubject; +import org.apache.qpid.server.model.VirtualHost; import org.apache.qpid.test.utils.QpidTestCase; import org.apache.qpid.util.FileUtils; @@ -167,8 +171,11 @@ public abstract class AbstractUpgradeTestCase extends QpidTestCase return count.longValue(); } - public String getVirtualHostName() + public VirtualHost getVirtualHost() { - return getName(); + VirtualHost virtualHost = mock(VirtualHost.class); + when(virtualHost.getName()).thenReturn(getName()); + when(virtualHost.getId()).thenReturn(UUID.randomUUID()); + return virtualHost; } } diff --git a/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgradeFrom4to5Test.java b/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgradeFrom4to5Test.java index 500fb0a919..d0f9455d9a 100644 --- a/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgradeFrom4to5Test.java +++ b/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgradeFrom4to5Test.java @@ -73,7 +73,7 @@ public class UpgradeFrom4to5Test extends AbstractUpgradeTestCase public void testPerformUpgradeWithHandlerAnsweringYes() throws Exception { UpgradeFrom4To5 upgrade = new UpgradeFrom4To5(); - upgrade.performUpgrade(_environment, new StaticAnswerHandler(UpgradeInteractionResponse.YES), getVirtualHostName()); + upgrade.performUpgrade(_environment, new StaticAnswerHandler(UpgradeInteractionResponse.YES), getVirtualHost()); assertQueues(new HashSet<String>(Arrays.asList(QUEUE_NAMES))); @@ -103,7 +103,7 @@ public class UpgradeFrom4to5Test extends AbstractUpgradeTestCase public void testPerformUpgradeWithHandlerAnsweringNo() throws Exception { UpgradeFrom4To5 upgrade = new UpgradeFrom4To5(); - upgrade.performUpgrade(_environment, new StaticAnswerHandler(UpgradeInteractionResponse.NO), getVirtualHostName()); + upgrade.performUpgrade(_environment, new StaticAnswerHandler(UpgradeInteractionResponse.NO), getVirtualHost()); HashSet<String> queues = new HashSet<String>(Arrays.asList(QUEUE_NAMES)); assertTrue(NON_DURABLE_QUEUE_NAME + " should be in the list of queues" , queues.remove(NON_DURABLE_QUEUE_NAME)); diff --git a/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgradeFrom5To6Test.java b/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgradeFrom5To6Test.java index 701fd94115..0460b1ce4c 100644 --- a/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgradeFrom5To6Test.java +++ b/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgradeFrom5To6Test.java @@ -87,7 +87,7 @@ public class UpgradeFrom5To6Test extends AbstractUpgradeTestCase public void testPerformUpgrade() throws Exception { UpgradeFrom5To6 upgrade = new UpgradeFrom5To6(); - upgrade.performUpgrade(_environment, UpgradeInteractionHandler.DEFAULT_HANDLER, getVirtualHostName()); + upgrade.performUpgrade(_environment, UpgradeInteractionHandler.DEFAULT_HANDLER, getVirtualHost()); assertDatabaseRecordCounts(); assertContent(); @@ -101,7 +101,7 @@ public class UpgradeFrom5To6Test extends AbstractUpgradeTestCase corruptDatabase(); UpgradeFrom5To6 upgrade = new UpgradeFrom5To6(); - upgrade.performUpgrade(_environment, new StaticAnswerHandler(UpgradeInteractionResponse.YES), getVirtualHostName()); + upgrade.performUpgrade(_environment, new StaticAnswerHandler(UpgradeInteractionResponse.YES), getVirtualHost()); assertDatabaseRecordCounts(); @@ -117,7 +117,7 @@ public class UpgradeFrom5To6Test extends AbstractUpgradeTestCase UpgradeInteractionHandler discardMessageInteractionHandler = new StaticAnswerHandler(UpgradeInteractionResponse.NO); - upgrade.performUpgrade(_environment, discardMessageInteractionHandler, getVirtualHostName()); + upgrade.performUpgrade(_environment, discardMessageInteractionHandler, getVirtualHost()); assertDatabaseRecordCount(NEW_METADATA_DB_NAME, 12); assertDatabaseRecordCount(NEW_CONTENT_DB_NAME, 12); @@ -135,7 +135,7 @@ public class UpgradeFrom5To6Test extends AbstractUpgradeTestCase { populateOldXidEntries(environment); UpgradeFrom5To6 upgrade = new UpgradeFrom5To6(); - upgrade.performUpgrade(environment, UpgradeInteractionHandler.DEFAULT_HANDLER, getVirtualHostName()); + upgrade.performUpgrade(environment, UpgradeInteractionHandler.DEFAULT_HANDLER, getVirtualHost()); assertXidEntries(environment); } finally @@ -171,11 +171,11 @@ public class UpgradeFrom5To6Test extends AbstractUpgradeTestCase NewRecordImpl[] newDequeues = newTransaction.getDequeues(); assertEquals("Unxpected new enqueus number", 1, newEnqueues.length); NewRecordImpl enqueue = newEnqueues[0]; - assertEquals("Unxpected queue id", UUIDGenerator.generateQueueUUID("TEST1", getVirtualHostName()), enqueue.getId()); + assertEquals("Unxpected queue id", UUIDGenerator.generateQueueUUID("TEST1", getVirtualHost().getName()), enqueue.getId()); assertEquals("Unxpected message id", 1, enqueue.getMessageNumber()); assertEquals("Unxpected new dequeues number", 1, newDequeues.length); NewRecordImpl dequeue = newDequeues[0]; - assertEquals("Unxpected queue id", UUIDGenerator.generateQueueUUID("TEST2", getVirtualHostName()), dequeue.getId()); + assertEquals("Unxpected queue id", UUIDGenerator.generateQueueUUID("TEST2", getVirtualHost().getName()), dequeue.getId()); assertEquals("Unxpected message id", 2, dequeue.getMessageNumber()); } @@ -347,13 +347,13 @@ public class UpgradeFrom5To6Test extends AbstractUpgradeTestCase { String exchangeName = (String) deserialized.get(Exchange.NAME); assertNotNull(exchangeName); - assertEquals("Unexpected key", key, UUIDGenerator.generateExchangeUUID(exchangeName, getVirtualHostName())); + assertEquals("Unexpected key", key, UUIDGenerator.generateExchangeUUID(exchangeName, getVirtualHost().getName())); } else if (type.equals(Queue.class.getName())) { String queueName = (String) deserialized.get(Queue.NAME); assertNotNull(queueName); - assertEquals("Unexpected key", key, UUIDGenerator.generateQueueUUID(queueName, getVirtualHostName())); + assertEquals("Unexpected key", key, UUIDGenerator.generateQueueUUID(queueName, getVirtualHost().getName())); } else if (type.equals(Binding.class.getName())) { @@ -368,15 +368,15 @@ public class UpgradeFrom5To6Test extends AbstractUpgradeTestCase private Map<String, Object> createExpectedQueueBindingMapAndID(String queue, String bindingName, String exchangeName, Map<String, String> argumentMap, List<UUID> expectedBindingIDs) { Map<String, Object> expectedQueueBinding = new HashMap<String, Object>(); - expectedQueueBinding.put(Binding.QUEUE, UUIDGenerator.generateQueueUUID(queue, getVirtualHostName()).toString()); + expectedQueueBinding.put(Binding.QUEUE, UUIDGenerator.generateQueueUUID(queue, getVirtualHost().getName()).toString()); expectedQueueBinding.put(Binding.NAME, bindingName); - expectedQueueBinding.put(Binding.EXCHANGE, UUIDGenerator.generateExchangeUUID(exchangeName, getVirtualHostName()).toString()); + expectedQueueBinding.put(Binding.EXCHANGE, UUIDGenerator.generateExchangeUUID(exchangeName, getVirtualHost().getName()).toString()); if (argumentMap != null) { expectedQueueBinding.put(Binding.ARGUMENTS, argumentMap); } - expectedBindingIDs.add(UUIDGenerator.generateBindingUUID(exchangeName, queue, bindingName, getVirtualHostName())); + expectedBindingIDs.add(UUIDGenerator.generateBindingUUID(exchangeName, queue, bindingName, getVirtualHost().getName())); return expectedQueueBinding; } diff --git a/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgraderFailOnNewerVersionTest.java b/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgraderFailOnNewerVersionTest.java index 400ac12792..c0de884a28 100644 --- a/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgraderFailOnNewerVersionTest.java +++ b/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgraderFailOnNewerVersionTest.java @@ -43,7 +43,7 @@ public class UpgraderFailOnNewerVersionTest extends AbstractUpgradeTestCase public void setUp() throws Exception { super.setUp(); - _upgrader = new Upgrader(_environment, getVirtualHostName()); + _upgrader = new Upgrader(_environment, getVirtualHost()); } private int getStoreVersion() diff --git a/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgraderTest.java b/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgraderTest.java index 3465f3582f..4b9a8d19a8 100644 --- a/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgraderTest.java +++ b/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgraderTest.java @@ -51,7 +51,7 @@ public class UpgraderTest extends AbstractUpgradeTestCase public void setUp() throws Exception { super.setUp(); - _upgrader = new Upgrader(_environment, getVirtualHostName()); + _upgrader = new Upgrader(_environment, getVirtualHost()); } private int getStoreVersion(Environment environment) @@ -108,7 +108,7 @@ public class UpgraderTest extends AbstractUpgradeTestCase Environment emptyEnvironment = createEnvironment(nonExistentStoreLocation); try { - _upgrader = new Upgrader(emptyEnvironment, getVirtualHostName()); + _upgrader = new Upgrader(emptyEnvironment, getVirtualHost()); _upgrader.upgradeIfNecessary(); List<String> databaseNames = emptyEnvironment.getDatabaseNames(); diff --git a/qpid/java/bdbstore/systests/src/main/java/org/apache/qpid/server/store/berkeleydb/BDBMessageStoreTest.java b/qpid/java/bdbstore/systests/src/main/java/org/apache/qpid/server/store/berkeleydb/BDBMessageStoreTest.java index bd0411619e..29e9ae19d7 100644 --- a/qpid/java/bdbstore/systests/src/main/java/org/apache/qpid/server/store/berkeleydb/BDBMessageStoreTest.java +++ b/qpid/java/bdbstore/systests/src/main/java/org/apache/qpid/server/store/berkeleydb/BDBMessageStoreTest.java @@ -25,7 +25,8 @@ import java.nio.ByteBuffer; import java.util.Arrays; import java.util.List; import java.util.UUID; -import org.apache.qpid.server.store.StoreException; + +import org.apache.qpid.server.store.*; import org.apache.qpid.framing.AMQShortString; import org.apache.qpid.framing.BasicContentHeaderProperties; import org.apache.qpid.framing.ContentHeaderBody; @@ -41,12 +42,6 @@ import org.apache.qpid.server.message.MessageReference; import org.apache.qpid.server.message.ServerMessage; import org.apache.qpid.server.model.UUIDGenerator; import org.apache.qpid.server.protocol.v0_8.MessageMetaDataType_0_8; -import org.apache.qpid.server.store.MessageStoreTest; -import org.apache.qpid.server.store.MessageStore; -import org.apache.qpid.server.store.StorableMessageMetaData; -import org.apache.qpid.server.store.StoredMessage; -import org.apache.qpid.server.store.Transaction; -import org.apache.qpid.server.store.TransactionLogResource; import org.apache.qpid.transport.DeliveryProperties; import org.apache.qpid.transport.Header; import org.apache.qpid.transport.MessageAcceptMode; @@ -56,6 +51,8 @@ import org.apache.qpid.transport.MessageDeliveryPriority; import org.apache.qpid.transport.MessageProperties; import org.apache.qpid.transport.MessageTransfer; +import static org.mockito.Mockito.mock; + /** * Subclass of MessageStoreTest which runs the standard tests from the superclass against * the BDB Store as well as additional tests specific to the BDB store-implementation. @@ -230,7 +227,8 @@ public class BDBMessageStoreTest extends MessageStoreTest messageStore.close(); AbstractBDBMessageStore newStore = new BDBMessageStore(); - newStore.configure(getVirtualHostModel(),true); + newStore.setVirtualHost(getVirtualHostModel()); + newStore.configure(true); newStore.startWithNoRecover(); |