You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by ro...@apache.org on 2012/04/17 17:04:38 UTC
svn commit: r1327128 [2/5] - in
/qpid/branches/java-config-and-management/qpid/java: ./
amqp-1-0-client-jms/ amqp-1-0-client/ amqp-1-0-common/
bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/
bdbstore/src/main/java/org/apache/qpid/server...
Modified: qpid/branches/java-config-and-management/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgradeFrom5To6.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgradeFrom5To6.java?rev=1327128&r1=1327127&r2=1327128&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgradeFrom5To6.java (original)
+++ qpid/branches/java-config-and-management/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgradeFrom5To6.java Tue Apr 17 15:04:34 2012
@@ -24,13 +24,30 @@ import static org.apache.qpid.server.sto
import static org.apache.qpid.server.store.berkeleydb.upgrade.UpgradeInteractionResponse.NO;
import static org.apache.qpid.server.store.berkeleydb.upgrade.UpgradeInteractionResponse.YES;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
+import java.util.Set;
import java.util.SortedMap;
import java.util.TreeMap;
+import java.util.UUID;
import org.apache.log4j.Logger;
import org.apache.qpid.AMQStoreException;
+import org.apache.qpid.exchange.ExchangeDefaults;
+import org.apache.qpid.framing.AMQShortString;
+import org.apache.qpid.framing.FieldTable;
+import org.apache.qpid.server.model.Binding;
+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.store.berkeleydb.AMQShortStringEncoding;
+import org.apache.qpid.server.store.berkeleydb.FieldTableEncoding;
+import org.apache.qpid.server.util.MapJsonSerializer;
import com.sleepycat.bind.tuple.LongBinding;
import com.sleepycat.bind.tuple.TupleBinding;
@@ -51,26 +68,71 @@ public class UpgradeFrom5To6 extends Abs
private static final Logger _logger = Logger.getLogger(UpgradeFrom5To6.class);
- private static final String OLD_CONTENT_DB_NAME = "messageContentDb_v5";
- private static final String NEW_CONTENT_DB_NAME = "MESSAGE_CONTENT";
- private static final String META_DATA_DB_NAME = "messageMetaDataDb_v5";
-
- private static final String NEW_DB_NAMES[] = { "EXCHANGES", "QUEUES", "QUEUE_BINDINGS", "DELIVERIES",
- "MESSAGE_METADATA", NEW_CONTENT_DB_NAME, "BRIDGES", "LINKS", "XIDS" };
- private static final String OLD_DB_NAMES[] = { "exchangeDb_v5", "queueDb_v5", "queueBindingsDb_v5", "deliveryDb_v5",
- META_DATA_DB_NAME, OLD_CONTENT_DB_NAME, "bridges_v5", "links_v5", "xids_v5" };
+ static final String OLD_CONTENT_DB_NAME = "messageContentDb_v5";
+ static final String NEW_CONTENT_DB_NAME = "MESSAGE_CONTENT";
+ static final String NEW_METADATA_DB_NAME = "MESSAGE_METADATA";
+ static final String OLD_META_DATA_DB_NAME = "messageMetaDataDb_v5";
+ static final String OLD_EXCHANGE_DB_NAME = "exchangeDb_v5";
+ static final String OLD_QUEUE_DB_NAME = "queueDb_v5";
+ static final String OLD_DELIVERY_DB_NAME = "deliveryDb_v5";
+ static final String OLD_QUEUE_BINDINGS_DB_NAME = "queueBindingsDb_v5";
+ static final String OLD_XID_DB_NAME = "xids_v5";
+ static final String NEW_XID_DB_NAME = "XIDS";
+ static final String CONFIGURED_OBJECTS_DB_NAME = "CONFIGURED_OBJECTS";
+ static final String NEW_DELIVERY_DB_NAME = "QUEUE_ENTRIES";
+ static final String NEW_BRIDGES_DB_NAME = "BRIDGES";
+ static final String NEW_LINKS_DB_NAME = "LINKS";
+ static final String OLD_BRIDGES_DB_NAME = "bridges_v5";
+ static final String OLD_LINKS_DB_NAME = "links_v5";
+
+ static final String[] DEFAULT_EXCHANGES = { ExchangeDefaults.DEFAULT_EXCHANGE_NAME.asString(),
+ ExchangeDefaults.DEFAULT_EXCHANGE_NAME.asString(), ExchangeDefaults.FANOUT_EXCHANGE_NAME.asString(),
+ ExchangeDefaults.HEADERS_EXCHANGE_NAME.asString(), ExchangeDefaults.TOPIC_EXCHANGE_NAME.asString(),
+ ExchangeDefaults.DIRECT_EXCHANGE_NAME.asString() };
+ private static final Set<String> DEFAULT_EXCHANGES_SET = new HashSet<String>(Arrays.asList(DEFAULT_EXCHANGES));
- public void performUpgrade(final Environment environment, final UpgradeInteractionHandler handler) throws DatabaseException, AMQStoreException
+ private MapJsonSerializer _serializer = new MapJsonSerializer();
+
+ /**
+ * Upgrades from a v5 database to a v6 database
+ *
+ * v6 is the first "new style" schema where we don't version every table,
+ * and the upgrade is re-runnable
+ *
+ * Change in this version:
+ *
+ * Message content is moved from the database messageContentDb_v5 to
+ * MESSAGE_CONTENT. The structure of the database changes from ( message-id:
+ * long, chunk-id: int ) -> ( size: int, byte[] data ) to ( message-id:
+ * long) -> ( byte[] data )
+ *
+ * That is we keep only one record per message, which contains all the
+ * message content
+ *
+ * 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)
+ throws DatabaseException, AMQStoreException
+ {
+ reportStarting(environment, 5);
+ upgradeMessages(environment, handler);
+ upgradeConfiguredObjectsAndDependencies(environment, handler, virtualHostName);
+ renameDatabases(environment, null);
+ reportFinished(environment, 6);
+ }
+
+ private void upgradeConfiguredObjectsAndDependencies(Environment environment, UpgradeInteractionHandler handler, String virtualHostName)
+ throws AMQStoreException
{
- _logger.info("Starting store upgrade from version 5");
Transaction transaction = null;
try
{
- reportStarting(environment, OLD_DB_NAMES, USER_FRIENDLY_NAMES);
transaction = environment.beginTransaction(null, null);
- performUpgradeInternal(environment, handler, transaction);
+ upgradeConfiguredObjects(environment, handler, transaction, virtualHostName);
+ upgradeQueueEntries(environment, transaction, virtualHostName);
+ upgradeXidEntries(environment, transaction, virtualHostName);
transaction.commit();
- reportFinished(environment, NEW_DB_NAMES, USER_FRIENDLY_NAMES);
}
catch (Exception e)
{
@@ -90,22 +152,53 @@ public class UpgradeFrom5To6 extends Abs
}
}
- /**
- * Upgrades from a v5 database to a v6 database
- *
- * v6 is the first "new style" schema where we don't version every table, and the upgrade is re-runnable
- *
- * Change in this version:
- *
- * Message content is moved from the database messageContentDb_v5 to MESSAGE_CONTENT.
- * The structure of the database changes from
- * ( message-id: long, chunk-id: int ) -> ( size: int, byte[] data )
- * to
- * ( message-id: long) -> ( byte[] data )
- *
- * That is we keep only one record per message, which contains all the message content
- */
- public void performUpgradeInternal(final Environment environment, final UpgradeInteractionHandler handler,
+ private void upgradeMessages(final Environment environment, final UpgradeInteractionHandler handler)
+ throws AMQStoreException
+ {
+ Transaction transaction = null;
+ try
+ {
+ transaction = environment.beginTransaction(null, null);
+ upgradeMessages(environment, handler, transaction);
+ transaction.commit();
+ }
+ catch (Exception e)
+ {
+ transaction.abort();
+ if (e instanceof DatabaseException)
+ {
+ throw (DatabaseException) e;
+ }
+ else if (e instanceof AMQStoreException)
+ {
+ throw (AMQStoreException) e;
+ }
+ else
+ {
+ throw new AMQStoreException("Unexpected exception", e);
+ }
+ }
+ }
+
+ private void renameDatabases(Environment environment, Transaction transaction)
+ {
+ List<String> databases = environment.getDatabaseNames();
+ String[] oldDatabases = { OLD_META_DATA_DB_NAME, OLD_BRIDGES_DB_NAME, OLD_LINKS_DB_NAME };
+ String[] newDatabases = { NEW_METADATA_DB_NAME, NEW_BRIDGES_DB_NAME, NEW_LINKS_DB_NAME };
+
+ for (int i = 0; i < oldDatabases.length; i++)
+ {
+ String oldName = oldDatabases[i];
+ String newName = newDatabases[i];
+ if (databases.contains(oldName))
+ {
+ _logger.info("Renaming " + oldName + " into " + newName);
+ environment.renameDatabase(transaction, oldName, newName);
+ }
+ }
+ }
+
+ private void upgradeMessages(final Environment environment, final UpgradeInteractionHandler handler,
final Transaction transaction) throws AMQStoreException
{
_logger.info("Message Contents");
@@ -129,27 +222,13 @@ public class UpgradeFrom5To6 extends Abs
metadataDatabase);
}
};
- new DatabaseTemplate(environment, META_DATA_DB_NAME, contentTransaction).run(metaDataDatabaseOperation);
+ new DatabaseTemplate(environment, OLD_META_DATA_DB_NAME, contentTransaction)
+ .run(metaDataDatabaseOperation);
_logger.info(metaDataDatabaseOperation.getRowCount() + " Message Content Entries");
}
};
new DatabaseTemplate(environment, OLD_CONTENT_DB_NAME, NEW_CONTENT_DB_NAME, transaction).run(contentOperation);
- }
- renameDatabases(environment, transaction);
- }
-
- private void renameDatabases(Environment environment, Transaction transaction)
- {
- List<String> databases = environment.getDatabaseNames();
- for (int i = 0; i < OLD_DB_NAMES.length; i++)
- {
- String oldName = OLD_DB_NAMES[i];
- String newName = NEW_DB_NAMES[i];
- if (databases.contains(oldName))
- {
- _logger.info("Renaming " + oldName + " into " + newName);
- environment.renameDatabase(transaction, oldName, newName);
- }
+ environment.removeDatabase(transaction, OLD_CONTENT_DB_NAME);
}
}
@@ -221,7 +300,7 @@ public class UpgradeFrom5To6 extends Abs
DatabaseEntry value = new DatabaseEntry();
dataBinding.objectToEntry(consolidatedData, value);
- newDatabase.put(txn, key, value);
+ put(newDatabase, txn, key, value);
}
/**
@@ -268,6 +347,264 @@ public class UpgradeFrom5To6 extends Abs
return data;
}
+ private void upgradeConfiguredObjects(Environment environment, UpgradeInteractionHandler handler, Transaction transaction, String virtualHostName)
+ throws AMQStoreException
+ {
+ upgradeQueues(environment, transaction, virtualHostName);
+ upgradeExchanges(environment, transaction, virtualHostName);
+ upgradeQueueBindings(environment, transaction, handler, virtualHostName);
+ }
+
+ private void upgradeXidEntries(Environment environment, Transaction transaction, final String virtualHostName)
+ {
+ if (environment.getDatabaseNames().contains(OLD_XID_DB_NAME))
+ {
+ _logger.info("Xid Records");
+ final OldPreparedTransactionBinding oldTransactionBinding = new OldPreparedTransactionBinding();
+ final NewPreparedTransactionBinding newTransactionBinding = new NewPreparedTransactionBinding();
+ CursorOperation xidEntriesCursor = new CursorOperation()
+ {
+ @Override
+ public void processEntry(Database oldXidDatabase, Database newXidDatabase, Transaction transaction,
+ DatabaseEntry key, DatabaseEntry value)
+ {
+ OldPreparedTransaction oldPreparedTransaction = oldTransactionBinding.entryToObject(value);
+ OldRecordImpl[] oldDequeues = oldPreparedTransaction.getDequeues();
+ OldRecordImpl[] oldEnqueues = oldPreparedTransaction.getEnqueues();
+
+ NewRecordImpl[] newEnqueues = null;
+ NewRecordImpl[] newDequeues = null;
+ if (oldDequeues != null)
+ {
+ newDequeues = new NewRecordImpl[oldDequeues.length];
+ for (int i = 0; i < newDequeues.length; i++)
+ {
+ OldRecordImpl dequeue = oldDequeues[i];
+ UUID id = UUIDGenerator.generateUUID(dequeue.getQueueName(), virtualHostName);
+ newDequeues[i] = new NewRecordImpl(id, dequeue.getMessageNumber());
+ }
+ }
+ if (oldEnqueues != null)
+ {
+ newEnqueues = new NewRecordImpl[oldEnqueues.length];
+ for (int i = 0; i < newEnqueues.length; i++)
+ {
+ OldRecordImpl enqueue = oldEnqueues[i];
+ UUID id = UUIDGenerator.generateUUID(enqueue.getQueueName(), virtualHostName);
+ newEnqueues[i] = new NewRecordImpl(id, enqueue.getMessageNumber());
+ }
+ }
+ NewPreparedTransaction newPreparedTransaction = new NewPreparedTransaction(newEnqueues, newDequeues);
+ DatabaseEntry newValue = new DatabaseEntry();
+ newTransactionBinding.objectToEntry(newPreparedTransaction, newValue);
+ put(newXidDatabase, transaction, key, newValue);
+ }
+ };
+ new DatabaseTemplate(environment, OLD_XID_DB_NAME, NEW_XID_DB_NAME, transaction).run(xidEntriesCursor);
+ environment.removeDatabase(transaction, OLD_XID_DB_NAME);
+ _logger.info(xidEntriesCursor.getRowCount() + " Xid Entries");
+ }
+ }
+
+ private void upgradeQueueEntries(Environment environment, Transaction transaction, final String virtualHostName)
+ {
+ _logger.info("Queue Delivery Records");
+ if (environment.getDatabaseNames().contains(OLD_DELIVERY_DB_NAME))
+ {
+ final OldQueueEntryBinding oldBinding = new OldQueueEntryBinding();
+ final NewQueueEntryBinding newBinding = new NewQueueEntryBinding();
+ CursorOperation queueEntriesCursor = new CursorOperation()
+ {
+ @Override
+ public void processEntry(Database oldDeliveryDatabase, Database newDeliveryDatabase,
+ Transaction transaction, DatabaseEntry key, DatabaseEntry value)
+ {
+ OldQueueEntryKey oldEntryRecord = oldBinding.entryToObject(key);
+ UUID queueId = UUIDGenerator.generateUUID(oldEntryRecord.getQueueName().asString(), virtualHostName);
+
+ NewQueueEntryKey newEntryRecord = new NewQueueEntryKey(queueId, oldEntryRecord.getMessageId());
+ DatabaseEntry newKey = new DatabaseEntry();
+ newBinding.objectToEntry(newEntryRecord, newKey);
+ put(newDeliveryDatabase, transaction, newKey, value);
+ }
+ };
+ new DatabaseTemplate(environment, OLD_DELIVERY_DB_NAME, NEW_DELIVERY_DB_NAME, transaction)
+ .run(queueEntriesCursor);
+ environment.removeDatabase(transaction, OLD_DELIVERY_DB_NAME);
+ _logger.info(queueEntriesCursor.getRowCount() + " Queue Delivery Record Entries");
+ }
+ }
+
+ private void upgradeQueueBindings(Environment environment, Transaction transaction, final UpgradeInteractionHandler handler, final String virtualHostName)
+ {
+ _logger.info("Queue Bindings");
+ if (environment.getDatabaseNames().contains(OLD_QUEUE_BINDINGS_DB_NAME))
+ {
+ final QueueBindingBinding binding = new QueueBindingBinding();
+ CursorOperation bindingCursor = new CursorOperation()
+ {
+ @Override
+ public void processEntry(Database exchangeDatabase, Database configuredObjectsDatabase,
+ Transaction transaction, DatabaseEntry key, DatabaseEntry value)
+ {
+ // TODO: check and remove orphaned bindings
+ BindingRecord bindingRecord = binding.entryToObject(key);
+ String exchangeName = bindingRecord.getExchangeName() == null ? ExchangeDefaults.DEFAULT_EXCHANGE_NAME
+ .asString() : bindingRecord.getExchangeName().asString();
+ String queueName = bindingRecord.getQueueName().asString();
+ String routingKey = bindingRecord.getRoutingKey().asString();
+ FieldTable arguments = bindingRecord.getArguments();
+
+ UUID bindingId = UUIDGenerator.generateUUID();
+ UpgradeConfiguredObjectRecord configuredObject = createBindingConfiguredObjectRecord(exchangeName, queueName,
+ routingKey, arguments, virtualHostName);
+ storeConfiguredObjectEntry(configuredObjectsDatabase, bindingId, configuredObject, transaction);
+ }
+
+ };
+ new DatabaseTemplate(environment, OLD_QUEUE_BINDINGS_DB_NAME, CONFIGURED_OBJECTS_DB_NAME, transaction)
+ .run(bindingCursor);
+ environment.removeDatabase(transaction, OLD_QUEUE_BINDINGS_DB_NAME);
+ _logger.info(bindingCursor.getRowCount() + " Queue Binding Entries");
+ }
+ }
+
+ private List<String> upgradeExchanges(Environment environment, Transaction transaction, final String virtualHostName)
+ {
+ final List<String> exchangeNames = new ArrayList<String>();
+ _logger.info("Exchanges");
+ if (environment.getDatabaseNames().contains(OLD_EXCHANGE_DB_NAME))
+ {
+ final ExchangeBinding exchangeBinding = new ExchangeBinding();
+ CursorOperation exchangeCursor = new CursorOperation()
+ {
+ @Override
+ public void processEntry(Database exchangeDatabase, Database configuredObjectsDatabase,
+ Transaction transaction, DatabaseEntry key, DatabaseEntry value)
+ {
+ ExchangeRecord exchangeRecord = exchangeBinding.entryToObject(value);
+ String exchangeName = exchangeRecord.getNameShortString().asString();
+ if (!DEFAULT_EXCHANGES_SET.contains(exchangeName))
+ {
+ String exchangeType = exchangeRecord.getType().asString();
+ boolean autoDelete = exchangeRecord.isAutoDelete();
+
+ UUID exchangeId = UUIDGenerator.generateUUID(exchangeName, virtualHostName);
+
+ UpgradeConfiguredObjectRecord configuredObject = createExchangeConfiguredObjectRecord(exchangeName,
+ exchangeType, autoDelete);
+ storeConfiguredObjectEntry(configuredObjectsDatabase, exchangeId, configuredObject, transaction);
+ exchangeNames.add(exchangeName);
+ }
+ }
+ };
+ new DatabaseTemplate(environment, OLD_EXCHANGE_DB_NAME, CONFIGURED_OBJECTS_DB_NAME, transaction)
+ .run(exchangeCursor);
+ environment.removeDatabase(transaction, OLD_EXCHANGE_DB_NAME);
+ _logger.info(exchangeCursor.getRowCount() + " Exchange Entries");
+ }
+ return exchangeNames;
+ }
+
+ private List<String> upgradeQueues(Environment environment, Transaction transaction, final String virtualHostName)
+ {
+ final List<String> queueNames = new ArrayList<String>();
+ _logger.info("Queues");
+ if (environment.getDatabaseNames().contains(OLD_QUEUE_DB_NAME))
+ {
+ final UpgradeQueueBinding queueBinding = new UpgradeQueueBinding();
+ CursorOperation queueCursor = new CursorOperation()
+ {
+ @Override
+ public void processEntry(Database queueDatabase, Database configuredObjectsDatabase,
+ Transaction transaction, DatabaseEntry key, DatabaseEntry value)
+ {
+ OldQueueRecord queueRecord = queueBinding.entryToObject(value);
+ String queueName = queueRecord.getNameShortString().asString();
+ queueNames.add(queueName);
+ String owner = queueRecord.getOwner() == null ? null : queueRecord.getOwner().asString();
+ boolean exclusive = queueRecord.isExclusive();
+ FieldTable arguments = queueRecord.getArguments();
+
+ UUID queueId = UUIDGenerator.generateUUID(queueName, virtualHostName);
+ UpgradeConfiguredObjectRecord configuredObject = createQueueConfiguredObjectRecord(queueName, owner, exclusive,
+ arguments);
+ storeConfiguredObjectEntry(configuredObjectsDatabase, queueId, configuredObject, transaction);
+ }
+ };
+ new DatabaseTemplate(environment, OLD_QUEUE_DB_NAME, CONFIGURED_OBJECTS_DB_NAME, transaction).run(queueCursor);
+ environment.removeDatabase(transaction, OLD_QUEUE_DB_NAME);
+ _logger.info(queueCursor.getRowCount() + " Queue Entries");
+ }
+ return queueNames;
+ }
+
+ private void storeConfiguredObjectEntry(Database configuredObjectsDatabase, UUID id,
+ UpgradeConfiguredObjectRecord configuredObject, Transaction transaction)
+ {
+ DatabaseEntry key = new DatabaseEntry();
+ DatabaseEntry value = new DatabaseEntry();
+ UpgradeUUIDBinding uuidBinding = new UpgradeUUIDBinding();
+ uuidBinding.objectToEntry(id, key);
+ ConfiguredObjectBinding configuredBinding = new ConfiguredObjectBinding();
+ configuredBinding.objectToEntry(configuredObject, value);
+ put(configuredObjectsDatabase, transaction, key, value);
+ }
+
+ private UpgradeConfiguredObjectRecord createQueueConfiguredObjectRecord(String queueName, String owner, boolean exclusive,
+ FieldTable arguments)
+ {
+ Map<String, Object> attributesMap = new HashMap<String, Object>();
+ attributesMap.put(Queue.NAME, queueName);
+ attributesMap.put(Queue.OWNER, owner);
+ attributesMap.put(Queue.EXCLUSIVE, exclusive);
+ if (arguments != null)
+ {
+ attributesMap.put("ARGUMENTS", FieldTable.convertToMap(arguments));
+ }
+ String json = _serializer.serialize(attributesMap);
+ UpgradeConfiguredObjectRecord configuredObject = new UpgradeConfiguredObjectRecord(Queue.class.getName(), json);
+ return configuredObject;
+ }
+
+ private UpgradeConfiguredObjectRecord createExchangeConfiguredObjectRecord(String exchangeName, String exchangeType,
+ boolean autoDelete)
+ {
+ Map<String, Object> attributesMap = new HashMap<String, Object>();
+ attributesMap.put(Exchange.NAME, exchangeName);
+ attributesMap.put(Exchange.TYPE, exchangeType);
+ attributesMap.put(Exchange.LIFETIME_POLICY, autoDelete ? LifetimePolicy.AUTO_DELETE.name()
+ : LifetimePolicy.PERMANENT.name());
+ String json = _serializer.serialize(attributesMap);
+ UpgradeConfiguredObjectRecord configuredObject = new UpgradeConfiguredObjectRecord(Exchange.class.getName(), json);
+ return configuredObject;
+ }
+
+ private UpgradeConfiguredObjectRecord createBindingConfiguredObjectRecord(String exchangeName, String queueName,
+ String routingKey, FieldTable arguments, String virtualHostName)
+ {
+ Map<String, Object> attributesMap = new HashMap<String, Object>();
+ attributesMap.put(Binding.NAME, routingKey);
+ attributesMap.put(Binding.EXCHANGE, UUIDGenerator.generateUUID(exchangeName, virtualHostName));
+ attributesMap.put(Binding.QUEUE, UUIDGenerator.generateUUID(queueName, virtualHostName));
+ if (arguments != null)
+ {
+ attributesMap.put(Binding.ARGUMENTS, FieldTable.convertToMap(arguments));
+ }
+ String json = _serializer.serialize(attributesMap);
+ UpgradeConfiguredObjectRecord configuredObject = new UpgradeConfiguredObjectRecord(Binding.class.getName(), json);
+ return configuredObject;
+ }
+
+ private void put(final Database database, Transaction txn, DatabaseEntry key, DatabaseEntry value)
+ {
+ OperationStatus status = database.put(txn, key, value);
+ if (status != OperationStatus.SUCCESS)
+ {
+ throw new RuntimeException("Cannot add record into " + database.getDatabaseName() + ":" + status);
+ }
+ }
+
static final class CompoundKey
{
public final long _messageId;
@@ -367,4 +704,504 @@ public class UpgradeFrom5To6 extends Abs
}
}
+ static class OldQueueRecord extends Object
+ {
+ private final AMQShortString _queueName;
+ private final AMQShortString _owner;
+ private final FieldTable _arguments;
+ private boolean _exclusive;
+
+ public OldQueueRecord(AMQShortString queueName, AMQShortString owner, boolean exclusive, FieldTable arguments)
+ {
+ _queueName = queueName;
+ _owner = owner;
+ _exclusive = exclusive;
+ _arguments = arguments;
+ }
+
+ public AMQShortString getNameShortString()
+ {
+ return _queueName;
+ }
+
+ public AMQShortString getOwner()
+ {
+ return _owner;
+ }
+
+ public boolean isExclusive()
+ {
+ return _exclusive;
+ }
+
+ public void setExclusive(boolean exclusive)
+ {
+ _exclusive = exclusive;
+ }
+
+ public FieldTable getArguments()
+ {
+ return _arguments;
+ }
+
+ }
+
+ static class UpgradeConfiguredObjectRecord
+ {
+ private String _attributes;
+ private String _type;
+
+ public UpgradeConfiguredObjectRecord(String type, String attributes)
+ {
+ super();
+ _attributes = attributes;
+ _type = type;
+ }
+
+ public String getAttributes()
+ {
+ return _attributes;
+ }
+
+ public String getType()
+ {
+ return _type;
+ }
+
+ }
+
+ static class UpgradeQueueBinding extends TupleBinding<OldQueueRecord>
+ {
+ public OldQueueRecord entryToObject(TupleInput tupleInput)
+ {
+ AMQShortString name = AMQShortStringEncoding.readShortString(tupleInput);
+ AMQShortString owner = AMQShortStringEncoding.readShortString(tupleInput);
+ FieldTable arguments = FieldTableEncoding.readFieldTable(tupleInput);
+ boolean exclusive = tupleInput.readBoolean();
+ return new OldQueueRecord(name, owner, exclusive, arguments);
+ }
+
+ public void objectToEntry(OldQueueRecord queue, TupleOutput tupleOutput)
+ {
+ AMQShortStringEncoding.writeShortString(queue.getNameShortString(), tupleOutput);
+ AMQShortStringEncoding.writeShortString(queue.getOwner(), tupleOutput);
+ FieldTableEncoding.writeFieldTable(queue.getArguments(), tupleOutput);
+ tupleOutput.writeBoolean(queue.isExclusive());
+ }
+ }
+
+ static class UpgradeUUIDBinding extends TupleBinding<UUID>
+ {
+ public UUID entryToObject(final TupleInput tupleInput)
+ {
+ return new UUID(tupleInput.readLong(), tupleInput.readLong());
+ }
+
+ public void objectToEntry(final UUID uuid, final TupleOutput tupleOutput)
+ {
+ tupleOutput.writeLong(uuid.getMostSignificantBits());
+ tupleOutput.writeLong(uuid.getLeastSignificantBits());
+ }
+ }
+
+ static class ConfiguredObjectBinding extends TupleBinding<UpgradeConfiguredObjectRecord>
+ {
+
+ public UpgradeConfiguredObjectRecord entryToObject(TupleInput tupleInput)
+ {
+ String type = tupleInput.readString();
+ String json = tupleInput.readString();
+ UpgradeConfiguredObjectRecord configuredObject = new UpgradeConfiguredObjectRecord(type, json);
+ return configuredObject;
+ }
+
+ public void objectToEntry(UpgradeConfiguredObjectRecord object, TupleOutput tupleOutput)
+ {
+ tupleOutput.writeString(object.getType());
+ tupleOutput.writeString(object.getAttributes());
+ }
+
+ }
+
+ static class ExchangeRecord extends Object
+ {
+ private final AMQShortString _exchangeName;
+ private final AMQShortString _exchangeType;
+ private final boolean _autoDelete;
+
+ public ExchangeRecord(AMQShortString exchangeName, AMQShortString exchangeType, boolean autoDelete)
+ {
+ _exchangeName = exchangeName;
+ _exchangeType = exchangeType;
+ _autoDelete = autoDelete;
+ }
+
+ public AMQShortString getNameShortString()
+ {
+ return _exchangeName;
+ }
+
+ public AMQShortString getType()
+ {
+ return _exchangeType;
+ }
+
+ public boolean isAutoDelete()
+ {
+ return _autoDelete;
+ }
+
+ }
+
+ static class ExchangeBinding extends TupleBinding<ExchangeRecord>
+ {
+
+ public ExchangeRecord entryToObject(TupleInput tupleInput)
+ {
+ AMQShortString name = AMQShortStringEncoding.readShortString(tupleInput);
+ AMQShortString typeName = AMQShortStringEncoding.readShortString(tupleInput);
+
+ boolean autoDelete = tupleInput.readBoolean();
+
+ return new ExchangeRecord(name, typeName, autoDelete);
+ }
+
+ public void objectToEntry(ExchangeRecord exchange, TupleOutput tupleOutput)
+ {
+ AMQShortStringEncoding.writeShortString(exchange.getNameShortString(), tupleOutput);
+ AMQShortStringEncoding.writeShortString(exchange.getType(), tupleOutput);
+
+ tupleOutput.writeBoolean(exchange.isAutoDelete());
+ }
+ }
+
+ static class BindingRecord extends Object
+ {
+ private final AMQShortString _exchangeName;
+ private final AMQShortString _queueName;
+ private final AMQShortString _routingKey;
+ private final FieldTable _arguments;
+
+ public BindingRecord(AMQShortString exchangeName, AMQShortString queueName, AMQShortString routingKey,
+ FieldTable arguments)
+ {
+ _exchangeName = exchangeName;
+ _queueName = queueName;
+ _routingKey = routingKey;
+ _arguments = arguments;
+ }
+
+ public AMQShortString getExchangeName()
+ {
+ return _exchangeName;
+ }
+
+ public AMQShortString getQueueName()
+ {
+ return _queueName;
+ }
+
+ public AMQShortString getRoutingKey()
+ {
+ return _routingKey;
+ }
+
+ public FieldTable getArguments()
+ {
+ return _arguments;
+ }
+
+ }
+
+ static class QueueBindingBinding extends TupleBinding<BindingRecord>
+ {
+
+ public BindingRecord entryToObject(TupleInput tupleInput)
+ {
+ AMQShortString exchangeName = AMQShortStringEncoding.readShortString(tupleInput);
+ AMQShortString queueName = AMQShortStringEncoding.readShortString(tupleInput);
+ AMQShortString routingKey = AMQShortStringEncoding.readShortString(tupleInput);
+
+ FieldTable arguments = FieldTableEncoding.readFieldTable(tupleInput);
+
+ return new BindingRecord(exchangeName, queueName, routingKey, arguments);
+ }
+
+ public void objectToEntry(BindingRecord binding, TupleOutput tupleOutput)
+ {
+ AMQShortStringEncoding.writeShortString(binding.getExchangeName(), tupleOutput);
+ AMQShortStringEncoding.writeShortString(binding.getQueueName(), tupleOutput);
+ AMQShortStringEncoding.writeShortString(binding.getRoutingKey(), tupleOutput);
+
+ FieldTableEncoding.writeFieldTable(binding.getArguments(), tupleOutput);
+ }
+ }
+
+ static class OldQueueEntryKey
+ {
+ private AMQShortString _queueName;
+ private long _messageId;
+
+ public OldQueueEntryKey(AMQShortString queueName, long messageId)
+ {
+ _queueName = queueName;
+ _messageId = messageId;
+ }
+
+ public AMQShortString getQueueName()
+ {
+ return _queueName;
+ }
+
+ public long getMessageId()
+ {
+ return _messageId;
+ }
+ }
+
+ static class OldQueueEntryBinding extends TupleBinding<OldQueueEntryKey>
+ {
+
+ public OldQueueEntryKey entryToObject(TupleInput tupleInput)
+ {
+ AMQShortString queueName = AMQShortStringEncoding.readShortString(tupleInput);
+ long messageId = tupleInput.readLong();
+
+ return new OldQueueEntryKey(queueName, messageId);
+ }
+
+ public void objectToEntry(OldQueueEntryKey mk, TupleOutput tupleOutput)
+ {
+ AMQShortStringEncoding.writeShortString(mk.getQueueName(), tupleOutput);
+ tupleOutput.writeLong(mk.getMessageId());
+ }
+ }
+
+ static class NewQueueEntryKey
+ {
+ private UUID _queueId;
+ private long _messageId;
+
+ public NewQueueEntryKey(UUID queueId, long messageId)
+ {
+ _queueId = queueId;
+ _messageId = messageId;
+ }
+
+ public UUID getQueueId()
+ {
+ return _queueId;
+ }
+
+ public long getMessageId()
+ {
+ return _messageId;
+ }
+ }
+
+ static class NewQueueEntryBinding extends TupleBinding<NewQueueEntryKey>
+ {
+
+ public NewQueueEntryKey entryToObject(TupleInput tupleInput)
+ {
+ UUID queueId = new UUID(tupleInput.readLong(), tupleInput.readLong());
+ long messageId = tupleInput.readLong();
+
+ return new NewQueueEntryKey(queueId, messageId);
+ }
+
+ public void objectToEntry(NewQueueEntryKey mk, TupleOutput tupleOutput)
+ {
+ UUID uuid = mk.getQueueId();
+ tupleOutput.writeLong(uuid.getMostSignificantBits());
+ tupleOutput.writeLong(uuid.getLeastSignificantBits());
+ tupleOutput.writeLong(mk.getMessageId());
+ }
+ }
+
+ static class NewPreparedTransaction
+ {
+ private final NewRecordImpl[] _enqueues;
+ private final NewRecordImpl[] _dequeues;
+
+ public NewPreparedTransaction(NewRecordImpl[] enqueues, NewRecordImpl[] dequeues)
+ {
+ _enqueues = enqueues;
+ _dequeues = dequeues;
+ }
+
+ public NewRecordImpl[] getEnqueues()
+ {
+ return _enqueues;
+ }
+
+ public NewRecordImpl[] getDequeues()
+ {
+ return _dequeues;
+ }
+ }
+
+ static class NewRecordImpl
+ {
+
+ private long _messageNumber;
+ private UUID _queueId;
+
+ public NewRecordImpl(UUID queueId, long messageNumber)
+ {
+ _messageNumber = messageNumber;
+ _queueId = queueId;
+ }
+
+ public long getMessageNumber()
+ {
+ return _messageNumber;
+ }
+
+ public UUID getId()
+ {
+ return _queueId;
+ }
+ }
+
+ static class NewPreparedTransactionBinding extends TupleBinding<NewPreparedTransaction>
+ {
+ @Override
+ public NewPreparedTransaction entryToObject(TupleInput input)
+ {
+ NewRecordImpl[] enqueues = readRecords(input);
+
+ NewRecordImpl[] dequeues = readRecords(input);
+
+ return new NewPreparedTransaction(enqueues, dequeues);
+ }
+
+ private NewRecordImpl[] readRecords(TupleInput input)
+ {
+ NewRecordImpl[] records = new NewRecordImpl[input.readInt()];
+ for (int i = 0; i < records.length; i++)
+ {
+ records[i] = new NewRecordImpl(new UUID(input.readLong(), input.readLong()), input.readLong());
+ }
+ return records;
+ }
+
+ @Override
+ public void objectToEntry(NewPreparedTransaction preparedTransaction, TupleOutput output)
+ {
+ writeRecords(preparedTransaction.getEnqueues(), output);
+ writeRecords(preparedTransaction.getDequeues(), output);
+ }
+
+ private void writeRecords(NewRecordImpl[] records, TupleOutput output)
+ {
+ if (records == null)
+ {
+ output.writeInt(0);
+ }
+ else
+ {
+ output.writeInt(records.length);
+ for (NewRecordImpl record : records)
+ {
+ UUID id = record.getId();
+ output.writeLong(id.getMostSignificantBits());
+ output.writeLong(id.getLeastSignificantBits());
+ output.writeLong(record.getMessageNumber());
+ }
+ }
+ }
+ }
+
+ static class OldRecordImpl
+ {
+
+ private long _messageNumber;
+ private String _queueName;
+
+ public OldRecordImpl(String queueName, long messageNumber)
+ {
+ _messageNumber = messageNumber;
+ _queueName = queueName;
+ }
+
+ public long getMessageNumber()
+ {
+ return _messageNumber;
+ }
+
+ public String getQueueName()
+ {
+ return _queueName;
+ }
+ }
+
+ static class OldPreparedTransaction
+ {
+ private final OldRecordImpl[] _enqueues;
+ private final OldRecordImpl[] _dequeues;
+
+ public OldPreparedTransaction(OldRecordImpl[] enqueues, OldRecordImpl[] dequeues)
+ {
+ _enqueues = enqueues;
+ _dequeues = dequeues;
+ }
+
+ public OldRecordImpl[] getEnqueues()
+ {
+ return _enqueues;
+ }
+
+ public OldRecordImpl[] getDequeues()
+ {
+ return _dequeues;
+ }
+ }
+
+ static class OldPreparedTransactionBinding extends TupleBinding<OldPreparedTransaction>
+ {
+ @Override
+ public OldPreparedTransaction entryToObject(TupleInput input)
+ {
+ OldRecordImpl[] enqueues = readRecords(input);
+
+ OldRecordImpl[] dequeues = readRecords(input);
+
+ return new OldPreparedTransaction(enqueues, dequeues);
+ }
+
+ private OldRecordImpl[] readRecords(TupleInput input)
+ {
+ OldRecordImpl[] records = new OldRecordImpl[input.readInt()];
+ for (int i = 0; i < records.length; i++)
+ {
+ records[i] = new OldRecordImpl(input.readString(), input.readLong());
+ }
+ return records;
+ }
+
+ @Override
+ public void objectToEntry(OldPreparedTransaction preparedTransaction, TupleOutput output)
+ {
+ writeRecords(preparedTransaction.getEnqueues(), output);
+ writeRecords(preparedTransaction.getDequeues(), output);
+ }
+
+ private void writeRecords(OldRecordImpl[] records, TupleOutput output)
+ {
+ if (records == null)
+ {
+ output.writeInt(0);
+ }
+ else
+ {
+ output.writeInt(records.length);
+ for (OldRecordImpl record : records)
+ {
+ output.writeString(record.getQueueName());
+ output.writeLong(record.getMessageNumber());
+ }
+ }
+ }
+ }
}
\ No newline at end of file
Modified: qpid/branches/java-config-and-management/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/Upgrader.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/Upgrader.java?rev=1327128&r1=1327127&r2=1327128&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/Upgrader.java (original)
+++ qpid/branches/java-config-and-management/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/Upgrader.java Tue Apr 17 15:04:34 2012
@@ -38,13 +38,15 @@ import com.sleepycat.je.OperationStatus;
public class Upgrader
{
- static final String VERSION_DB_NAME = "VERSION";
+ static final String VERSION_DB_NAME = "DB_VERSION";
private Environment _environment;
+ private String _virtualHostName;
- public Upgrader(Environment environment)
+ public Upgrader(Environment environment, String virtualHostName)
{
_environment = environment;
+ _virtualHostName = virtualHostName;
}
public void upgradeIfNecessary() throws AMQStoreException
@@ -125,7 +127,7 @@ public class Upgrader
+ "UpgradeFrom"+fromVersion+"To"+toVersion);
Constructor<StoreUpgrade> ctr = upgradeClass.getConstructor();
StoreUpgrade upgrade = ctr.newInstance();
- upgrade.performUpgrade(_environment, UpgradeInteractionHandler.DEFAULT_HANDLER);
+ upgrade.performUpgrade(_environment, UpgradeInteractionHandler.DEFAULT_HANDLER, _virtualHostName);
}
catch (ClassNotFoundException e)
{
Modified: qpid/branches/java-config-and-management/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/BDBMessageStoreTest.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/BDBMessageStoreTest.java?rev=1327128&r1=1327127&r2=1327128&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/BDBMessageStoreTest.java (original)
+++ qpid/branches/java-config-and-management/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/BDBMessageStoreTest.java Tue Apr 17 15:04:34 2012
@@ -23,6 +23,7 @@ package org.apache.qpid.server.store.ber
import java.nio.ByteBuffer;
import java.util.Arrays;
import java.util.List;
+import java.util.UUID;
import org.apache.qpid.AMQStoreException;
import org.apache.qpid.framing.AMQShortString;
import org.apache.qpid.framing.BasicContentHeaderProperties;
@@ -36,6 +37,7 @@ import org.apache.qpid.server.message.Me
import org.apache.qpid.server.message.MessageMetaData_0_10;
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.store.MessageMetaDataType;
import org.apache.qpid.server.store.MessageStore;
import org.apache.qpid.server.store.StorableMessageMetaData;
@@ -405,13 +407,13 @@ public class BDBMessageStoreTest extends
BDBMessageStore bdbStore = assertBDBStore(log);
- final AMQShortString mockQueueName = new AMQShortString("queueName");
-
+ final UUID mockQueueId = UUIDGenerator.generateUUID();
TransactionLogResource mockQueue = new TransactionLogResource()
{
- public String getResourceName()
+ @Override
+ public UUID getId()
{
- return mockQueueName.asString();
+ return mockQueueId;
}
};
@@ -421,7 +423,7 @@ public class BDBMessageStoreTest extends
txn.enqueueMessage(mockQueue, new MockMessage(5L));
txn.commitTran();
- List<Long> enqueuedIds = bdbStore.getEnqueuedMessages(mockQueueName);
+ List<Long> enqueuedIds = bdbStore.getEnqueuedMessages(mockQueueId);
assertEquals("Number of enqueued messages is incorrect", 2, enqueuedIds.size());
Long val = enqueuedIds.get(0);
@@ -443,13 +445,13 @@ public class BDBMessageStoreTest extends
BDBMessageStore bdbStore = assertBDBStore(log);
- final AMQShortString mockQueueName = new AMQShortString("queueName");
-
+ final UUID mockQueueId = UUIDGenerator.generateUUID();
TransactionLogResource mockQueue = new TransactionLogResource()
{
- public String getResourceName()
+ @Override
+ public UUID getId()
{
- return mockQueueName.asString();
+ return mockQueueId;
}
};
@@ -463,7 +465,7 @@ public class BDBMessageStoreTest extends
txn.enqueueMessage(mockQueue, new MockMessage(23L));
txn.commitTran();
- List<Long> enqueuedIds = bdbStore.getEnqueuedMessages(mockQueueName);
+ List<Long> enqueuedIds = bdbStore.getEnqueuedMessages(mockQueueId);
assertEquals("Number of enqueued messages is incorrect", 2, enqueuedIds.size());
Long val = enqueuedIds.get(0);
@@ -484,13 +486,13 @@ public class BDBMessageStoreTest extends
BDBMessageStore bdbStore = assertBDBStore(log);
- final AMQShortString mockQueueName = new AMQShortString("queueName");
-
+ final UUID mockQueueId = UUIDGenerator.generateUUID();
TransactionLogResource mockQueue = new TransactionLogResource()
{
- public String getResourceName()
+ @Override
+ public UUID getId()
{
- return mockQueueName.asString();
+ return mockQueueId;
}
};
@@ -507,7 +509,7 @@ public class BDBMessageStoreTest extends
txn.enqueueMessage(mockQueue, new MockMessage(32L));
txn.commitTran();
- List<Long> enqueuedIds = bdbStore.getEnqueuedMessages(mockQueueName);
+ List<Long> enqueuedIds = bdbStore.getEnqueuedMessages(mockQueueId);
assertEquals("Number of enqueued messages is incorrect", 2, enqueuedIds.size());
Long val = enqueuedIds.get(0);
Modified: qpid/branches/java-config-and-management/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/upgrade/AbstractUpgradeTestCase.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/upgrade/AbstractUpgradeTestCase.java?rev=1327128&r1=1327127&r2=1327128&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/upgrade/AbstractUpgradeTestCase.java (original)
+++ qpid/branches/java-config-and-management/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/upgrade/AbstractUpgradeTestCase.java Tue Apr 17 15:04:34 2012
@@ -22,10 +22,9 @@ package org.apache.qpid.server.store.ber
import java.io.File;
-import junit.framework.TestCase;
-
import org.apache.qpid.server.logging.LogSubject;
import org.apache.qpid.server.logging.subjects.TestBlankSubject;
+import org.apache.qpid.test.utils.QpidTestCase;
import org.apache.qpid.util.FileUtils;
import com.sleepycat.je.Database;
@@ -33,7 +32,7 @@ import com.sleepycat.je.Environment;
import com.sleepycat.je.EnvironmentConfig;
import com.sleepycat.je.Transaction;
-public abstract class AbstractUpgradeTestCase extends TestCase
+public abstract class AbstractUpgradeTestCase extends QpidTestCase
{
protected static final class StaticAnswerHandler implements UpgradeInteractionHandler
{
@@ -57,7 +56,6 @@ public abstract class AbstractUpgradeTes
public static int[] QUEUE_SIZES = { 1, 1, 10, 3 };
public static int TOTAL_MESSAGE_NUMBER = 15;
protected static final LogSubject LOG_SUBJECT = new TestBlankSubject();
- protected static final String TMP_FOLDER = System.getProperty("java.io.tmpdir");
// one binding per exchange
protected static final int TOTAL_BINDINGS = QUEUE_NAMES.length * 2;
@@ -148,4 +146,8 @@ public abstract class AbstractUpgradeTes
return count.longValue();
}
+ public String getVirtualHostName()
+ {
+ return getName();
+ }
}
Modified: qpid/branches/java-config-and-management/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgradeFrom4to5Test.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgradeFrom4to5Test.java?rev=1327128&r1=1327127&r2=1327128&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgradeFrom4to5Test.java (original)
+++ qpid/branches/java-config-and-management/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgradeFrom4to5Test.java Tue Apr 17 15:04:34 2012
@@ -66,7 +66,7 @@ public class UpgradeFrom4to5Test extends
public void testPerformUpgradeWithHandlerAnsweringYes() throws Exception
{
UpgradeFrom4To5 upgrade = new UpgradeFrom4To5();
- upgrade.performUpgrade(_environment, new StaticAnswerHandler(UpgradeInteractionResponse.YES));
+ upgrade.performUpgrade(_environment, new StaticAnswerHandler(UpgradeInteractionResponse.YES), getVirtualHostName());
assertQueues(new HashSet<String>(Arrays.asList(QUEUE_NAMES)));
@@ -93,7 +93,7 @@ public class UpgradeFrom4to5Test extends
public void testPerformUpgradeWithHandlerAnsweringNo() throws Exception
{
UpgradeFrom4To5 upgrade = new UpgradeFrom4To5();
- upgrade.performUpgrade(_environment, new StaticAnswerHandler(UpgradeInteractionResponse.NO));
+ upgrade.performUpgrade(_environment, new StaticAnswerHandler(UpgradeInteractionResponse.NO), getVirtualHostName());
assertQueues(new HashSet<String>(Arrays.asList(DURABLE_SUBSCRIPTION_QUEUE, DURABLE_SUBSCRIPTION_QUEUE_WITH_SELECTOR, DURABLE_QUEUE)));
assertDatabaseRecordCount(DELIVERY_DB_NAME, 12);
Modified: qpid/branches/java-config-and-management/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgradeFrom5To6Test.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgradeFrom5To6Test.java?rev=1327128&r1=1327127&r2=1327128&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgradeFrom5To6Test.java (original)
+++ qpid/branches/java-config-and-management/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgradeFrom5To6Test.java Tue Apr 17 15:04:34 2012
@@ -20,14 +20,50 @@
*/
package org.apache.qpid.server.store.berkeleydb.upgrade;
+import static org.apache.qpid.server.store.berkeleydb.upgrade.UpgradeFrom5To6.CONFIGURED_OBJECTS_DB_NAME;
+import static org.apache.qpid.server.store.berkeleydb.upgrade.UpgradeFrom5To6.NEW_CONTENT_DB_NAME;
+import static org.apache.qpid.server.store.berkeleydb.upgrade.UpgradeFrom5To6.NEW_DELIVERY_DB_NAME;
+import static org.apache.qpid.server.store.berkeleydb.upgrade.UpgradeFrom5To6.NEW_METADATA_DB_NAME;
+import static org.apache.qpid.server.store.berkeleydb.upgrade.UpgradeFrom5To6.NEW_XID_DB_NAME;
+import static org.apache.qpid.server.store.berkeleydb.upgrade.UpgradeFrom5To6.OLD_CONTENT_DB_NAME;
+import static org.apache.qpid.server.store.berkeleydb.upgrade.UpgradeFrom5To6.OLD_XID_DB_NAME;
+
+import java.io.File;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Set;
+import java.util.UUID;
+
import org.apache.log4j.Logger;
+import org.apache.qpid.server.model.Binding;
+import org.apache.qpid.server.model.Exchange;
+import org.apache.qpid.server.model.Queue;
+import org.apache.qpid.server.model.UUIDGenerator;
+import org.apache.qpid.server.store.berkeleydb.entry.Xid;
+import org.apache.qpid.server.store.berkeleydb.tuple.XidBinding;
import org.apache.qpid.server.store.berkeleydb.upgrade.UpgradeFrom5To6.CompoundKey;
import org.apache.qpid.server.store.berkeleydb.upgrade.UpgradeFrom5To6.CompoundKeyBinding;
+import org.apache.qpid.server.store.berkeleydb.upgrade.UpgradeFrom5To6.ConfiguredObjectBinding;
+import org.apache.qpid.server.store.berkeleydb.upgrade.UpgradeFrom5To6.UpgradeConfiguredObjectRecord;
import org.apache.qpid.server.store.berkeleydb.upgrade.UpgradeFrom5To6.NewDataBinding;
+import org.apache.qpid.server.store.berkeleydb.upgrade.UpgradeFrom5To6.NewPreparedTransaction;
+import org.apache.qpid.server.store.berkeleydb.upgrade.UpgradeFrom5To6.NewPreparedTransactionBinding;
+import org.apache.qpid.server.store.berkeleydb.upgrade.UpgradeFrom5To6.NewQueueEntryBinding;
+import org.apache.qpid.server.store.berkeleydb.upgrade.UpgradeFrom5To6.NewQueueEntryKey;
+import org.apache.qpid.server.store.berkeleydb.upgrade.UpgradeFrom5To6.NewRecordImpl;
+import org.apache.qpid.server.store.berkeleydb.upgrade.UpgradeFrom5To6.OldPreparedTransaction;
+import org.apache.qpid.server.store.berkeleydb.upgrade.UpgradeFrom5To6.OldPreparedTransactionBinding;
+import org.apache.qpid.server.store.berkeleydb.upgrade.UpgradeFrom5To6.OldRecordImpl;
+import org.apache.qpid.server.store.berkeleydb.upgrade.UpgradeFrom5To6.UpgradeUUIDBinding;
+import org.apache.qpid.server.util.MapJsonSerializer;
import com.sleepycat.bind.tuple.LongBinding;
import com.sleepycat.je.Database;
import com.sleepycat.je.DatabaseEntry;
+import com.sleepycat.je.Environment;
+import com.sleepycat.je.LockMode;
import com.sleepycat.je.Transaction;
public class UpgradeFrom5To6Test extends AbstractUpgradeTestCase
@@ -43,10 +79,13 @@ public class UpgradeFrom5To6Test extends
public void testPerformUpgrade() throws Exception
{
UpgradeFrom5To6 upgrade = new UpgradeFrom5To6();
- upgrade.performUpgrade(_environment, UpgradeInteractionHandler.DEFAULT_HANDLER);
+ upgrade.performUpgrade(_environment, UpgradeInteractionHandler.DEFAULT_HANDLER, getVirtualHostName());
assertDatabaseRecordCounts();
assertContent();
+
+ assertConfiguredObjects();
+ assertQueueEntries();
}
public void testPerformUpgradeWithMissingMessageChunkKeepsIncompleteMessage() throws Exception
@@ -54,9 +93,12 @@ public class UpgradeFrom5To6Test extends
corruptDatabase();
UpgradeFrom5To6 upgrade = new UpgradeFrom5To6();
- upgrade.performUpgrade(_environment, new StaticAnswerHandler(UpgradeInteractionResponse.YES));
+ upgrade.performUpgrade(_environment, new StaticAnswerHandler(UpgradeInteractionResponse.YES), getVirtualHostName());
assertDatabaseRecordCounts();
+
+ assertConfiguredObjects();
+ assertQueueEntries();
}
public void testPerformUpgradeWithMissingMessageChunkDiscardsIncompleteMessage() throws Exception
@@ -67,10 +109,117 @@ public class UpgradeFrom5To6Test extends
UpgradeInteractionHandler discardMessageInteractionHandler = new StaticAnswerHandler(UpgradeInteractionResponse.NO);
- upgrade.performUpgrade(_environment, discardMessageInteractionHandler);
+ upgrade.performUpgrade(_environment, discardMessageInteractionHandler, getVirtualHostName());
+
+ assertDatabaseRecordCount(NEW_METADATA_DB_NAME, 11);
+ assertDatabaseRecordCount(NEW_CONTENT_DB_NAME, 11);
+
+ assertConfiguredObjects();
+ assertQueueEntries();
+ }
+
+ public void testPerformXidUpgrade() throws Exception
+ {
+ File storeLocation = new File(TMP_FOLDER, getName());
+ storeLocation.mkdirs();
+ Environment environment = createEnvironment(storeLocation);
+ try
+ {
+ populateOldXidEntries(environment);
+ UpgradeFrom5To6 upgrade = new UpgradeFrom5To6();
+ upgrade.performUpgrade(environment, UpgradeInteractionHandler.DEFAULT_HANDLER, getVirtualHostName());
+ assertXidEntries(environment);
+ }
+ finally
+ {
+ try
+ {
+ environment.close();
+ }
+ finally
+ {
+ deleteDirectoryIfExists(storeLocation);
+ }
+
+ }
+ }
+
+ private void assertXidEntries(Environment environment)
+ {
+ final DatabaseEntry value = new DatabaseEntry();
+ final DatabaseEntry key = getXidKey();
+ new DatabaseTemplate(environment, NEW_XID_DB_NAME, null).run(new DatabaseRunnable()
+ {
+
+ @Override
+ public void run(Database xidDatabase, Database nullDatabase, Transaction transaction)
+ {
+ xidDatabase.get(null, key, value, LockMode.DEFAULT);
+ }
+ });
+ NewPreparedTransactionBinding newBinding = new NewPreparedTransactionBinding();
+ NewPreparedTransaction newTransaction = newBinding.entryToObject(value);
+ NewRecordImpl[] newEnqueues = newTransaction.getEnqueues();
+ NewRecordImpl[] newDequeues = newTransaction.getDequeues();
+ assertEquals("Unxpected new enqueus number", 1, newEnqueues.length);
+ NewRecordImpl enqueue = newEnqueues[0];
+ assertEquals("Unxpected queue id", UUIDGenerator.generateUUID("TEST1", getVirtualHostName()), 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.generateUUID("TEST2", getVirtualHostName()), dequeue.getId());
+ assertEquals("Unxpected message id", 2, dequeue.getMessageNumber());
+ }
+
+ private void populateOldXidEntries(Environment environment)
+ {
+
+ final DatabaseEntry value = new DatabaseEntry();
+ OldRecordImpl[] enqueues = { new OldRecordImpl("TEST1", 1) };
+ OldRecordImpl[] dequeues = { new OldRecordImpl("TEST2", 2) };
+ OldPreparedTransaction oldPreparedTransaction = new OldPreparedTransaction(enqueues, dequeues);
+ OldPreparedTransactionBinding oldPreparedTransactionBinding = new OldPreparedTransactionBinding();
+ oldPreparedTransactionBinding.objectToEntry(oldPreparedTransaction, value);
+
+ final DatabaseEntry key = getXidKey();
+ new DatabaseTemplate(environment, OLD_XID_DB_NAME, null).run(new DatabaseRunnable()
+ {
+
+ @Override
+ public void run(Database xidDatabase, Database nullDatabase, Transaction transaction)
+ {
+ xidDatabase.put(null, key, value);
+ }
+ });
+ }
+
+ protected DatabaseEntry getXidKey()
+ {
+ final DatabaseEntry value = new DatabaseEntry();
+ byte[] globalId = { 1 };
+ byte[] branchId = { 2 };
+ Xid xid = new Xid(1l, globalId, branchId);
+ XidBinding xidBinding = XidBinding.getInstance();
+ xidBinding.objectToEntry(xid, value);
+ return value;
+ }
- assertDatabaseRecordCount("MESSAGE_METADATA", 11);
- assertDatabaseRecordCount("MESSAGE_CONTENT", 11);
+ private void assertQueueEntries()
+ {
+ final Map<UUID, UpgradeConfiguredObjectRecord> configuredObjects = loadConfiguredObjects();
+ final NewQueueEntryBinding newBinding = new NewQueueEntryBinding();
+ CursorOperation cursorOperation = new CursorOperation()
+ {
+
+ @Override
+ public void processEntry(Database sourceDatabase, Database targetDatabase, Transaction transaction,
+ DatabaseEntry key, DatabaseEntry value)
+ {
+ NewQueueEntryKey newEntryRecord = newBinding.entryToObject(key);
+ assertTrue("Unexpected queue id", configuredObjects.containsKey(newEntryRecord.getQueueId()));
+ }
+ };
+ new DatabaseTemplate(_environment, NEW_DELIVERY_DB_NAME, null).run(cursorOperation);
}
/**
@@ -105,19 +254,124 @@ public class UpgradeFrom5To6Test extends
};
Transaction transaction = _environment.beginTransaction(null, null);
- new DatabaseTemplate(_environment, "messageContentDb_v5", transaction).run(cursorOperation);
+ new DatabaseTemplate(_environment, OLD_CONTENT_DB_NAME, transaction).run(cursorOperation);
transaction.commit();
}
private void assertDatabaseRecordCounts()
{
- assertDatabaseRecordCount("EXCHANGES", 5);
- assertDatabaseRecordCount("QUEUES", 3);
- assertDatabaseRecordCount("QUEUE_BINDINGS", 6);
- assertDatabaseRecordCount("DELIVERIES", 12);
+ assertDatabaseRecordCount(CONFIGURED_OBJECTS_DB_NAME, 9);
+ assertDatabaseRecordCount(NEW_DELIVERY_DB_NAME, 12);
+
+ assertDatabaseRecordCount(NEW_METADATA_DB_NAME, 12);
+ assertDatabaseRecordCount(NEW_CONTENT_DB_NAME, 12);
+ }
+
+ private void assertConfiguredObjects()
+ {
+ Map<UUID, UpgradeConfiguredObjectRecord> configuredObjects = loadConfiguredObjects();
+ assertEquals("Unexpected number of configured objects", 9, configuredObjects.size());
- assertDatabaseRecordCount("MESSAGE_METADATA", 12);
- assertDatabaseRecordCount("MESSAGE_CONTENT", 12);
+ Set<Map<String, Object>> expected = new HashSet<Map<String, Object>>(9);
+ Map<String, Object> queue1 = new HashMap<String, Object>();
+ queue1.put("exclusive", Boolean.FALSE);
+ queue1.put("name", "myUpgradeQueue");
+ queue1.put("owner", null);
+ expected.add(queue1);
+ Map<String, Object> queue2 = new HashMap<String, Object>();
+ queue2.put("exclusive", Boolean.TRUE);
+ queue2.put("name", "clientid:mySelectorDurSubName");
+ queue2.put("owner", "clientid");
+ expected.add(queue2);
+ Map<String, Object> queue3 = new HashMap<String, Object>();
+ queue3.put("exclusive", Boolean.TRUE);
+ queue3.put("name", "clientid:myDurSubName");
+ queue3.put("owner", "clientid");
+ expected.add(queue3);
+
+ Map<String, Object> queueBinding1 = new HashMap<String, Object>();
+ queueBinding1.put("queue", UUIDGenerator.generateUUID("myUpgradeQueue", getVirtualHostName()).toString());
+ queueBinding1.put("name", "myUpgradeQueue");
+ queueBinding1.put("exchange", UUIDGenerator.generateUUID("<<default>>", getVirtualHostName()).toString());
+ expected.add(queueBinding1);
+ Map<String, Object> queueBinding2 = new HashMap<String, Object>();
+ queueBinding2.put("queue", UUIDGenerator.generateUUID("myUpgradeQueue", getVirtualHostName()).toString());
+ queueBinding2.put("name", "myUpgradeQueue");
+ queueBinding2.put("exchange", UUIDGenerator.generateUUID("amq.direct", getVirtualHostName()).toString());
+ Map<String, Object> arguments2 = new HashMap<String, Object>();
+ arguments2.put("x-filter-jms-selector", "");
+ queueBinding2.put("arguments", arguments2);
+ expected.add(queueBinding2);
+ Map<String, Object> queueBinding3 = new HashMap<String, Object>();
+ queueBinding3.put("queue", UUIDGenerator.generateUUID("clientid:myDurSubName", getVirtualHostName()).toString());
+ queueBinding3.put("name", "myUpgradeTopic");
+ queueBinding3.put("exchange", UUIDGenerator.generateUUID("amq.topic", getVirtualHostName()).toString());
+ Map<String, Object> arguments3 = new HashMap<String, Object>();
+ arguments3.put("x-filter-jms-selector", "");
+ queueBinding3.put("arguments", arguments3);
+ expected.add(queueBinding3);
+ Map<String, Object> queueBinding4 = new HashMap<String, Object>();
+ queueBinding4.put("queue", UUIDGenerator.generateUUID("clientid:mySelectorDurSubName", getVirtualHostName()).toString());
+ queueBinding4.put("name", "mySelectorUpgradeTopic");
+ queueBinding4.put("exchange", UUIDGenerator.generateUUID("amq.topic", getVirtualHostName()).toString());
+ Map<String, Object> arguments4 = new HashMap<String, Object>();
+ arguments4.put("x-filter-jms-selector", "testprop='true'");
+ queueBinding4.put("arguments", arguments4);
+ expected.add(queueBinding4);
+ Map<String, Object> queueBinding5 = new HashMap<String, Object>();
+ queueBinding5.put("queue", UUIDGenerator.generateUUID("clientid:myDurSubName", getVirtualHostName()).toString());
+ queueBinding5.put("name", "clientid:myDurSubName");
+ queueBinding5.put("exchange", UUIDGenerator.generateUUID("<<default>>", getVirtualHostName()).toString());
+ expected.add(queueBinding5);
+ Map<String, Object> queueBinding6 = new HashMap<String, Object>();
+ queueBinding6.put("queue", UUIDGenerator.generateUUID("clientid:mySelectorDurSubName", getVirtualHostName()).toString());
+ queueBinding6.put("name", "clientid:mySelectorDurSubName");
+ queueBinding6.put("exchange", UUIDGenerator.generateUUID("<<default>>", getVirtualHostName()).toString());
+ expected.add(queueBinding6);
+
+ Set<String> expectedTypes = new HashSet<String>();
+ expectedTypes.add(Queue.class.getName());
+ expectedTypes.add(Exchange.class.getName());
+ expectedTypes.add(Binding.class.getName());
+ MapJsonSerializer jsonSerializer = new MapJsonSerializer();
+ for (Entry<UUID, UpgradeConfiguredObjectRecord> entry : configuredObjects.entrySet())
+ {
+ UpgradeConfiguredObjectRecord object = entry.getValue();
+ UUID key = entry.getKey();
+ Map<String, Object> deserialized = jsonSerializer.deserialize(object.getAttributes());
+ assertTrue("Unexpected entry:" + object.getAttributes(), expected.remove(deserialized));
+ String type = object.getType();
+ assertTrue("Unexpected type:" + type, expectedTypes.contains(type));
+ if (type.equals(Exchange.class.getName()) || type.equals(Queue.class.getName()))
+ {
+ assertEquals("Unexpected key", key, UUIDGenerator.generateUUID(((String) deserialized.get("name")), getVirtualHostName()));
+ }
+ else
+ {
+ assertNotNull("Key cannot be null", key);
+ }
+ }
+ assertTrue("Not all expected configured objects found:" + expected, expected.isEmpty());
+ }
+
+ private Map<UUID, UpgradeConfiguredObjectRecord> loadConfiguredObjects()
+ {
+ final Map<UUID, UpgradeConfiguredObjectRecord> configuredObjectsRecords = new HashMap<UUID, UpgradeConfiguredObjectRecord>();
+ final ConfiguredObjectBinding binding = new ConfiguredObjectBinding();
+ final UpgradeUUIDBinding uuidBinding = new UpgradeUUIDBinding();
+ CursorOperation configuredObjectsCursor = new CursorOperation()
+ {
+ @Override
+ public void processEntry(Database sourceDatabase, Database targetDatabase, Transaction transaction,
+ DatabaseEntry key, DatabaseEntry value)
+ {
+ UUID id = uuidBinding.entryToObject(key);
+ UpgradeConfiguredObjectRecord object = binding.entryToObject(value);
+ configuredObjectsRecords.put(id, object);
+ }
+ };
+ new DatabaseTemplate(_environment, CONFIGURED_OBJECTS_DB_NAME, null).run(configuredObjectsCursor);
+ return configuredObjectsRecords;
}
private void assertContent()
@@ -127,8 +381,8 @@ public class UpgradeFrom5To6Test extends
{
@Override
- public void processEntry(Database sourceDatabase, Database targetDatabase, Transaction transaction, DatabaseEntry key,
- DatabaseEntry value)
+ public void processEntry(Database sourceDatabase, Database targetDatabase, Transaction transaction,
+ DatabaseEntry key, DatabaseEntry value)
{
long id = LongBinding.entryToLong(key);
assertTrue("Unexpected id", id > 0);
@@ -136,6 +390,6 @@ public class UpgradeFrom5To6Test extends
assertNotNull("Unexpected content", content);
}
};
- new DatabaseTemplate(_environment, "MESSAGE_CONTENT", null).run(contentCursorOperation);
+ new DatabaseTemplate(_environment, NEW_CONTENT_DB_NAME, null).run(contentCursorOperation);
}
}
Modified: qpid/branches/java-config-and-management/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgraderTest.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgraderTest.java?rev=1327128&r1=1327127&r2=1327128&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgraderTest.java (original)
+++ qpid/branches/java-config-and-management/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgraderTest.java Tue Apr 17 15:04:34 2012
@@ -50,7 +50,7 @@ public class UpgraderTest extends Abstra
public void setUp() throws Exception
{
super.setUp();
- _upgrader = new Upgrader(_environment);
+ _upgrader = new Upgrader(_environment, getVirtualHostName());
}
private int getStoreVersion()
@@ -105,12 +105,12 @@ public class UpgraderTest extends Abstra
nonExistentStoreLocation.mkdir();
_environment = createEnvironment(nonExistentStoreLocation);
- _upgrader = new Upgrader(_environment);
+ _upgrader = new Upgrader(_environment, getVirtualHostName());
_upgrader.upgradeIfNecessary();
List<String> databaseNames = _environment.getDatabaseNames();
List<String> expectedDatabases = new ArrayList<String>();
- expectedDatabases.add("VERSION");
+ expectedDatabases.add(Upgrader.VERSION_DB_NAME);
assertEquals("Expectedonly VERSION table in initially empty store after upgrade: ", expectedDatabases, databaseNames);
assertEquals("Unexpected store version", BDBMessageStore.VERSION, getStoreVersion());
Propchange: qpid/branches/java-config-and-management/qpid/java/broker/
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/broker:r1327001-1327003
Modified: qpid/branches/java-config-and-management/qpid/java/broker-plugins/extras/src/main/java/org/apache/qpid/extras/exchanges/diagnostic/DiagnosticExchange.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/java/broker-plugins/extras/src/main/java/org/apache/qpid/extras/exchanges/diagnostic/DiagnosticExchange.java?rev=1327128&r1=1327127&r2=1327128&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/java/broker-plugins/extras/src/main/java/org/apache/qpid/extras/exchanges/diagnostic/DiagnosticExchange.java (original)
+++ qpid/branches/java-config-and-management/qpid/java/broker-plugins/extras/src/main/java/org/apache/qpid/extras/exchanges/diagnostic/DiagnosticExchange.java Tue Apr 17 15:04:34 2012
@@ -40,6 +40,7 @@ import javax.management.openmbean.OpenDa
import javax.management.openmbean.TabularData;
import java.util.ArrayList;
import java.util.Map;
+import java.util.UUID;
/**
* This is a special diagnostic exchange type which doesn't actually do anything
@@ -125,14 +126,14 @@ public class DiagnosticExchange extends
return DiagnosticExchange.class;
}
- public DiagnosticExchange newInstance(VirtualHost host,
+ public DiagnosticExchange newInstance(UUID id, VirtualHost host,
AMQShortString name,
boolean durable,
int ticket,
boolean autoDelete) throws AMQException
{
DiagnosticExchange exch = new DiagnosticExchange();
- exch.initialise(host,name,durable,ticket,autoDelete);
+ exch.initialise(id, host,name,durable,ticket,autoDelete);
return exch;
}
Modified: qpid/branches/java-config-and-management/qpid/java/broker-plugins/extras/src/main/java/org/apache/qpid/extras/exchanges/diagnostic/DiagnosticExchangeType.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/java/broker-plugins/extras/src/main/java/org/apache/qpid/extras/exchanges/diagnostic/DiagnosticExchangeType.java?rev=1327128&r1=1327127&r2=1327128&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/java/broker-plugins/extras/src/main/java/org/apache/qpid/extras/exchanges/diagnostic/DiagnosticExchangeType.java (original)
+++ qpid/branches/java-config-and-management/qpid/java/broker-plugins/extras/src/main/java/org/apache/qpid/extras/exchanges/diagnostic/DiagnosticExchangeType.java Tue Apr 17 15:04:34 2012
@@ -21,6 +21,8 @@
package org.apache.qpid.extras.exchanges.diagnostic;
+import java.util.UUID;
+
import org.apache.qpid.AMQException;
import org.apache.qpid.framing.AMQShortString;
import org.apache.qpid.server.exchange.ExchangeType;
@@ -42,11 +44,11 @@ public final class DiagnosticExchangeTyp
return DiagnosticExchange.class;
}
- public DiagnosticExchange newInstance(VirtualHost host, AMQShortString name, boolean durable, int ticket, boolean autoDelete)
+ public DiagnosticExchange newInstance(UUID id, VirtualHost host, AMQShortString name, boolean durable, int ticket, boolean autoDelete)
throws AMQException
{
DiagnosticExchange exch = new DiagnosticExchange();
- exch.initialise(host, name, durable, ticket, autoDelete);
+ exch.initialise(id, host, name, durable, ticket, autoDelete);
return exch;
}
Modified: qpid/branches/java-config-and-management/qpid/java/broker-plugins/extras/src/main/java/org/apache/qpid/extras/exchanges/example/TestExchange.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/java/broker-plugins/extras/src/main/java/org/apache/qpid/extras/exchanges/example/TestExchange.java?rev=1327128&r1=1327127&r2=1327128&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/java/broker-plugins/extras/src/main/java/org/apache/qpid/extras/exchanges/example/TestExchange.java (original)
+++ qpid/branches/java-config-and-management/qpid/java/broker-plugins/extras/src/main/java/org/apache/qpid/extras/exchanges/example/TestExchange.java Tue Apr 17 15:04:34 2012
@@ -261,7 +261,7 @@ public class TestExchange implements Exc
return 0;
}
- public void initialise(VirtualHost arg0, AMQShortString arg1, boolean arg2, int arg3, boolean arg4)
+ public void initialise(UUID id, VirtualHost arg0, AMQShortString arg1, boolean arg2, int arg3, boolean arg4)
throws AMQException
{
}
Modified: qpid/branches/java-config-and-management/qpid/java/broker-plugins/extras/src/main/java/org/apache/qpid/extras/exchanges/example/TestExchangeType.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/java/broker-plugins/extras/src/main/java/org/apache/qpid/extras/exchanges/example/TestExchangeType.java?rev=1327128&r1=1327127&r2=1327128&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/java/broker-plugins/extras/src/main/java/org/apache/qpid/extras/exchanges/example/TestExchangeType.java (original)
+++ qpid/branches/java-config-and-management/qpid/java/broker-plugins/extras/src/main/java/org/apache/qpid/extras/exchanges/example/TestExchangeType.java Tue Apr 17 15:04:34 2012
@@ -21,6 +21,8 @@
package org.apache.qpid.extras.exchanges.example;
+import java.util.UUID;
+
import org.apache.qpid.AMQException;
import org.apache.qpid.framing.AMQShortString;
import org.apache.qpid.server.exchange.Exchange;
@@ -40,12 +42,12 @@ public class TestExchangeType implements
return null;
}
- public Exchange newInstance(VirtualHost host, AMQShortString name, boolean durable,
+ public Exchange newInstance(UUID id, VirtualHost host, AMQShortString name, boolean durable,
int token, boolean autoDelete)
throws AMQException
{
TestExchange ex = new TestExchange();
- ex.initialise(host, name, durable, token, autoDelete);
+ ex.initialise(id, host, name, durable, token, autoDelete);
return ex;
}
Propchange: qpid/branches/java-config-and-management/qpid/java/broker/bin/
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/broker/bin:r1327001-1327003
Modified: qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/qmf/ManagementExchange.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/qmf/ManagementExchange.java?rev=1327128&r1=1327127&r2=1327128&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/qmf/ManagementExchange.java (original)
+++ qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/qmf/ManagementExchange.java Tue Apr 17 15:04:34 2012
@@ -37,6 +37,7 @@ import org.apache.qpid.server.exchange.t
import org.apache.qpid.server.exchange.topic.TopicParser;
import org.apache.qpid.server.message.InboundMessage;
import org.apache.qpid.server.message.ServerMessage;
+import org.apache.qpid.server.model.UUIDGenerator;
import org.apache.qpid.server.queue.AMQQueue;
import org.apache.qpid.server.queue.BaseQueue;
import org.apache.qpid.server.virtualhost.HouseKeepingTask;
@@ -83,7 +84,8 @@ public class ManagementExchange implemen
private class ManagementQueue implements BaseQueue
{
- private final String NAME_AS_STRING = "##__mgmt_pseudo_queue__##" + UUID.randomUUID().toString();
+ private final UUID QUEUE_ID = UUIDGenerator.generateUUID();
+ private final String NAME_AS_STRING = "##__mgmt_pseudo_queue__##" + QUEUE_ID.toString();
private final AMQShortString NAME_AS_SHORT_STRING = new AMQShortString(NAME_AS_STRING);
public void enqueue(ServerMessage message) throws AMQException
@@ -129,9 +131,10 @@ public class ManagementExchange implemen
return NAME_AS_SHORT_STRING;
}
- public String getResourceName()
+ @Override
+ public UUID getId()
{
- return NAME_AS_STRING;
+ return QUEUE_ID;
}
}
@@ -155,14 +158,14 @@ public class ManagementExchange implemen
return ManagementExchange.class;
}
- public ManagementExchange newInstance(VirtualHost host,
+ public ManagementExchange newInstance(UUID id, VirtualHost host,
AMQShortString name,
boolean durable,
int ticket,
boolean autoDelete) throws AMQException
{
ManagementExchange exch = new ManagementExchange();
- exch.initialise(host, name, durable, ticket, autoDelete);
+ exch.initialise(id, host, name, durable, ticket, autoDelete);
return exch;
}
@@ -183,7 +186,7 @@ public class ManagementExchange implemen
return QPID_MANAGEMENT_TYPE;
}
- public void initialise(VirtualHost host, AMQShortString name, boolean durable, int ticket, boolean autoDelete)
+ public void initialise(UUID id, VirtualHost host, AMQShortString name, boolean durable, int ticket, boolean autoDelete)
throws AMQException
{
if(!QPID_MANAGEMENT.equals(name))
@@ -191,7 +194,7 @@ public class ManagementExchange implemen
throw new AMQException("Can't create more than one Management exchange");
}
_virtualHost = host;
- _id = host.getConfigStore().createId();
+ _id = id;
_virtualHost.scheduleHouseKeepingTask(_virtualHost.getBroker().getManagementPublishInterval(), new UpdateTask(_virtualHost));
getConfigStore().addConfiguredObject(this);
getQMFService().addListener(this);
Modified: qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/binding/Binding.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/binding/Binding.java?rev=1327128&r1=1327127&r2=1327128&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/binding/Binding.java (original)
+++ qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/binding/Binding.java Tue Apr 17 15:04:34 2012
@@ -117,7 +117,7 @@ public class Binding
public String toString()
{
- return "Binding{bindingKey="+_bindingKey+", exchange="+_exchange+", queue="+_queue+"}";
+ return "Binding{bindingKey="+_bindingKey+", exchange="+_exchange+", queue="+_queue+", id= " + _id + " }";
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org