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 2011/09/24 22:16:02 UTC
svn commit: r1175235 [3/5] - in /qpid/trunk: ./ qpid/java/
qpid/java/bdbstore/ qpid/java/bdbstore/bin/ qpid/java/bdbstore/etc/
qpid/java/bdbstore/etc/scripts/ qpid/java/bdbstore/src/
qpid/java/bdbstore/src/main/ qpid/java/bdbstore/src/main/java/ qpid/j...
Added: qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/BDBStoreUpgrade.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/BDBStoreUpgrade.java?rev=1175235&view=auto
==============================================================================
--- qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/BDBStoreUpgrade.java (added)
+++ qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/BDBStoreUpgrade.java Sat Sep 24 20:16:00 2011
@@ -0,0 +1,1125 @@
+/*
+ *
+ * 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.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.apache.qpid.exchange.ExchangeDefaults;
+import org.apache.qpid.framing.AMQShortString;
+import org.apache.qpid.server.store.berkeleydb.BDBMessageStore;
+import org.apache.qpid.server.store.berkeleydb.BindingKey;
+import org.apache.qpid.server.store.berkeleydb.ContentTB;
+import org.apache.qpid.server.store.berkeleydb.DatabaseVisitor;
+import org.apache.qpid.server.store.berkeleydb.ExchangeTB;
+import org.apache.qpid.server.store.berkeleydb.keys.MessageContentKey_4;
+import org.apache.qpid.server.store.berkeleydb.keys.MessageContentKey_5;
+import org.apache.qpid.server.store.berkeleydb.records.ExchangeRecord;
+import org.apache.qpid.server.store.berkeleydb.records.QueueRecord;
+import org.apache.qpid.server.store.berkeleydb.tuples.MessageContentKeyTB_4;
+import org.apache.qpid.server.store.berkeleydb.tuples.MessageContentKeyTB_5;
+import org.apache.qpid.server.store.berkeleydb.tuples.QueueEntryTB;
+import org.apache.qpid.server.logging.actors.CurrentActor;
+import org.apache.qpid.server.logging.actors.BrokerActor;
+import org.apache.qpid.server.logging.NullRootMessageLogger;
+import org.apache.qpid.server.message.MessageMetaData;
+import org.apache.qpid.AMQException;
+import org.apache.qpid.AMQStoreException;
+import org.apache.qpid.util.FileUtils;
+import org.apache.commons.cli.PosixParser;
+import org.apache.commons.cli.Options;
+import org.apache.commons.cli.CommandLine;
+import org.apache.commons.cli.ParseException;
+import org.apache.commons.cli.Option;
+import org.apache.commons.cli.OptionBuilder;
+
+import java.io.File;
+import java.io.BufferedReader;
+import java.io.InputStreamReader;
+import java.io.IOException;
+import java.io.FileNotFoundException;
+import java.nio.ByteBuffer;
+import java.text.MessageFormat;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map.Entry;
+
+import com.sleepycat.je.DatabaseEntry;
+import com.sleepycat.je.DatabaseException;
+import com.sleepycat.je.Database;
+import com.sleepycat.je.Environment;
+import com.sleepycat.je.EnvironmentConfig;
+import com.sleepycat.bind.tuple.TupleBinding;
+
+/**
+ * This is a simple BerkeleyDB Store upgrade tool that will upgrade a V4 Store to a V5 Store.
+ *
+ * Currently upgrade is fixed from v4 -> v5
+ *
+ * Improvments:
+ * - Add List BDBMessageStore.getDatabases(); This can the be iterated to guard against new DBs being added.
+ * - A version in the store would allow automated upgrade or later with more available versions interactive upgrade.
+ * - Add process logging and disable all Store and Qpid logging.
+ */
+public class BDBStoreUpgrade
+{
+ private static final Logger _logger = LoggerFactory.getLogger(BDBStoreUpgrade.class);
+ /** The Store Directory that needs upgrading */
+ File _fromDir;
+ /** The Directory that will be made to contain the upgraded store */
+ File _toDir;
+ /** The Directory that will be made to backup the original store if required */
+ File _backupDir;
+
+ /** The Old Store */
+ BDBMessageStore _oldMessageStore;
+ /** The New Store */
+ BDBMessageStore _newMessageStore;
+ /** The file ending that is used by BDB Store Files */
+ private static final String BDB_FILE_ENDING = ".jdb";
+
+ static final Options _options = new Options();
+ static CommandLine _commandLine;
+ private boolean _interactive;
+ private boolean _force;
+
+ private static final String VERSION = "3.0";
+ private static final String USER_ABORTED_PROCESS = "User aborted process";
+ private static final int LOWEST_SUPPORTED_STORE_VERSION = 4;
+ private static final String PREVIOUS_STORE_VERSION_UNSUPPORTED = "Store upgrade from version {0} is not supported."
+ + " You must first run the previous store upgrade tool.";
+ private static final String FOLLOWING_STORE_VERSION_UNSUPPORTED = "Store version {0} is newer than this tool supports. "
+ + "You must use a newer version of the store upgrade tool";
+ private static final String STORE_ALREADY_UPGRADED = "Store has already been upgraded to version {0}.";
+
+ private static final String OPTION_INPUT_SHORT = "i";
+ private static final String OPTION_INPUT = "input";
+ private static final String OPTION_OUTPUT_SHORT = "o";
+ private static final String OPTION_OUTPUT = "output";
+ private static final String OPTION_BACKUP_SHORT = "b";
+ private static final String OPTION_BACKUP = "backup";
+ private static final String OPTION_QUIET_SHORT = "q";
+ private static final String OPTION_QUIET = "quiet";
+ private static final String OPTION_FORCE_SHORT = "f";
+ private static final String OPTION_FORCE = "force";
+ private boolean _inplace = false;
+
+ public BDBStoreUpgrade(String fromDir, String toDir, String backupDir, boolean interactive, boolean force)
+ {
+ _interactive = interactive;
+ _force = force;
+
+ _fromDir = new File(fromDir);
+ if (!_fromDir.exists())
+ {
+ throw new IllegalArgumentException("BDBStore path '" + fromDir + "' could not be read. "
+ + "Ensure the path is correct and that the permissions are correct.");
+ }
+
+ if (!isDirectoryAStoreDir(_fromDir))
+ {
+ throw new IllegalArgumentException("Specified directory '" + fromDir + "' does not contain a valid BDBMessageStore.");
+ }
+
+ if (toDir == null)
+ {
+ _inplace = true;
+ _toDir = new File(fromDir+"-Inplace");
+ }
+ else
+ {
+ _toDir = new File(toDir);
+ }
+
+ if (_toDir.exists())
+ {
+ if (_interactive)
+ {
+ if (toDir == null)
+ {
+ System.out.println("Upgrading in place:" + fromDir);
+ }
+ else
+ {
+ System.out.println("Upgrade destination: '" + toDir + "'");
+ }
+
+ if (userInteract("Upgrade destination exists do you wish to replace it?"))
+ {
+ if (!FileUtils.delete(_toDir, true))
+ {
+ throw new IllegalArgumentException("Unable to remove upgrade destination '" + _toDir + "'");
+ }
+ }
+ else
+ {
+ throw new IllegalArgumentException("Upgrade destination '" + _toDir + "' already exists. ");
+ }
+ }
+ else
+ {
+ if (_force)
+ {
+ if (!FileUtils.delete(_toDir, true))
+ {
+ throw new IllegalArgumentException("Unable to remove upgrade destination '" + _toDir + "'");
+ }
+ }
+ else
+ {
+ throw new IllegalArgumentException("Upgrade destination '" + _toDir + "' already exists. ");
+ }
+ }
+ }
+
+ if (!_toDir.mkdirs())
+ {
+ throw new IllegalArgumentException("Upgrade destination '" + _toDir + "' could not be created. "
+ + "Ensure the path is correct and that the permissions are correct.");
+ }
+
+ if (backupDir != null)
+ {
+ if (backupDir.equals(""))
+ {
+ _backupDir = new File(_fromDir.getAbsolutePath().toString() + "-Backup");
+ }
+ else
+ {
+ _backupDir = new File(backupDir);
+ }
+ }
+ else
+ {
+ _backupDir = null;
+ }
+ }
+
+ private static String ANSWER_OPTIONS = " Yes/No/Abort? ";
+ private static String ANSWER_NO = "no";
+ private static String ANSWER_N = "n";
+ private static String ANSWER_YES = "yes";
+ private static String ANSWER_Y = "y";
+ private static String ANSWER_ABORT = "abort";
+ private static String ANSWER_A = "a";
+
+ /**
+ * Interact with the user via System.in and System.out. If the user wishes to Abort then a RuntimeException is thrown.
+ * Otherwise the method will return based on their response true=yes false=no.
+ *
+ * @param message Message to print out
+ *
+ * @return boolean response from user if they wish to proceed
+ */
+ private boolean userInteract(String message)
+ {
+ System.out.print(message + ANSWER_OPTIONS);
+ BufferedReader br = new BufferedReader(new InputStreamReader(System.in));
+
+ String input = "";
+ try
+ {
+ input = br.readLine();
+ }
+ catch (IOException e)
+ {
+ input = "";
+ }
+
+ if (input.equalsIgnoreCase(ANSWER_Y) || input.equalsIgnoreCase(ANSWER_YES))
+ {
+ return true;
+ }
+ else
+ {
+ if (input.equalsIgnoreCase(ANSWER_N) || input.equalsIgnoreCase(ANSWER_NO))
+ {
+ return false;
+ }
+ else
+ {
+ if (input.equalsIgnoreCase(ANSWER_A) || input.equalsIgnoreCase(ANSWER_ABORT))
+ {
+ throw new RuntimeException(USER_ABORTED_PROCESS);
+ }
+ }
+ }
+
+ return userInteract(message);
+ }
+
+ /**
+ * Upgrade a Store of a specified version to the latest version.
+ *
+ * @param version the version of the current store
+ *
+ * @throws Exception
+ */
+ public void upgradeFromVersion(int version) throws Exception
+ {
+ upgradeFromVersion(version, _fromDir, _toDir, _backupDir, _force,
+ _inplace);
+ }
+
+ /**
+ * Upgrade a Store of a specified version to the latest version.
+ *
+ * @param version the version of the current store
+ * @param fromDir the directory with the old Store
+ * @param toDir the directrory to hold the newly Upgraded Store
+ * @param backupDir the directrory to backup to if required
+ * @param force suppress all questions
+ * @param inplace replace the from dir with the upgraded result in toDir
+ *
+ * @throws Exception due to Virtualhost/MessageStore.close() being
+ * rather poor at exception handling
+ * @throws DatabaseException if there is a problem with the store formats
+ * @throws AMQException if there is an issue creating Qpid data structures
+ */
+ public void upgradeFromVersion(int version, File fromDir, File toDir,
+ File backupDir, boolean force,
+ boolean inplace) throws Exception
+ {
+ _logger.info("Located store to upgrade at '" + fromDir + "'");
+
+ // Verify user has created a backup, giving option to perform backup
+ if (_interactive)
+ {
+ if (!userInteract("Have you performed a DB backup of this store."))
+ {
+ File backup;
+ if (backupDir == null)
+ {
+ backup = new File(fromDir.getAbsolutePath().toString() + "-Backup");
+ }
+ else
+ {
+ backup = backupDir;
+ }
+
+ if (userInteract("Do you wish to perform a DB backup now? " +
+ "(Store will be backed up to '" + backup.getName() + "')"))
+ {
+ performDBBackup(fromDir, backup, force);
+ }
+ else
+ {
+ if (!userInteract("Are you sure wish to proceed with DB migration without backup? " +
+ "(For more details of the consequences check the Qpid/BDB Message Store Wiki)."))
+ {
+ throw new IllegalArgumentException("Upgrade stopped at user request as no DB Backup performed.");
+ }
+ }
+ }
+ else
+ {
+ if (!inplace)
+ {
+ _logger.info("Upgrade will create a new store at '" + toDir + "'");
+ }
+
+ _logger.info("Using the contents in the Message Store '" + fromDir + "'");
+
+ if (!userInteract("Do you wish to proceed?"))
+ {
+ throw new IllegalArgumentException("Upgrade stopped as did not wish to proceed");
+ }
+ }
+ }
+ else
+ {
+ if (backupDir != null)
+ {
+ performDBBackup(fromDir, backupDir, force);
+ }
+ }
+
+ CurrentActor.set(new BrokerActor(new NullRootMessageLogger()));
+
+ //Create a new messageStore
+ _newMessageStore = new BDBMessageStore();
+ _newMessageStore.configure(toDir, false);
+ _newMessageStore.start();
+
+ try
+ {
+ //Load the old MessageStore
+ switch (version)
+ {
+ default:
+ case 4:
+ _oldMessageStore = new BDBMessageStore(4);
+ _oldMessageStore.configure(fromDir, true);
+ _oldMessageStore.start();
+ upgradeFromVersion_4();
+ break;
+ case 3:
+ case 2:
+ case 1:
+ throw new IllegalArgumentException(MessageFormat.format(PREVIOUS_STORE_VERSION_UNSUPPORTED,
+ new Object[] { Integer.toString(version) }));
+ }
+ }
+ finally
+ {
+ _newMessageStore.close();
+ if (_oldMessageStore != null)
+ {
+ _oldMessageStore.close();
+ }
+ // if we are running inplace then swap fromDir and toDir
+ if (inplace)
+ {
+ // Remove original copy
+ if (FileUtils.delete(fromDir, true))
+ {
+ // Rename upgraded store
+ toDir.renameTo(fromDir);
+ }
+ else
+ {
+ throw new RuntimeException("Unable to upgrade inplace as " +
+ "unable to delete source '"
+ +fromDir+"', Store upgrade " +
+ "successfully performed to :"
+ +toDir);
+ }
+ }
+ }
+ }
+
+ private void upgradeFromVersion_4() throws AMQException, DatabaseException
+ {
+ _logger.info("Starting store upgrade from version 4");
+
+ //Migrate _exchangeDb;
+ _logger.info("Exchanges");
+
+ moveContents(_oldMessageStore.getExchangesDb(), _newMessageStore.getExchangesDb(), "Exchange");
+
+ final List<AMQShortString> topicExchanges = new ArrayList<AMQShortString>();
+ final TupleBinding exchangeTB = new ExchangeTB();
+
+ DatabaseVisitor exchangeListVisitor = new DatabaseVisitor()
+ {
+ public void visit(DatabaseEntry key, DatabaseEntry value) throws DatabaseException
+ {
+ ExchangeRecord exchangeRec = (ExchangeRecord) exchangeTB.entryToObject(value);
+ AMQShortString type = exchangeRec.getType();
+
+ if (ExchangeDefaults.TOPIC_EXCHANGE_CLASS.equals(type))
+ {
+ topicExchanges.add(exchangeRec.getNameShortString());
+ }
+ }
+ };
+ _oldMessageStore.visitExchanges(exchangeListVisitor);
+
+
+ //Migrate _queueBindingsDb;
+ _logger.info("Queue Bindings");
+ moveContents(_oldMessageStore.getBindingsDb(), _newMessageStore.getBindingsDb(), "Queue Binding");
+
+ //Inspect the bindings to gather a list of queues which are probably durable subscriptions, i.e. those
+ //which have a colon in their name and are bound to the Topic exchanges above
+ final List<AMQShortString> durableSubQueues = new ArrayList<AMQShortString>();
+ final TupleBinding<BindingKey> bindingTB = _oldMessageStore.getBindingTupleBindingFactory().getInstance();
+
+ DatabaseVisitor durSubQueueListVisitor = new DatabaseVisitor()
+ {
+ public void visit(DatabaseEntry key, DatabaseEntry value) throws DatabaseException
+ {
+ BindingKey bindingRec = (BindingKey) bindingTB.entryToObject(key);
+ AMQShortString queueName = bindingRec.getQueueName();
+ AMQShortString exchangeName = bindingRec.getExchangeName();
+
+ if (topicExchanges.contains(exchangeName) && queueName.asString().contains(":"))
+ {
+ durableSubQueues.add(queueName);
+ }
+ }
+ };
+ _oldMessageStore.visitBindings(durSubQueueListVisitor);
+
+
+ //Migrate _queueDb;
+ _logger.info("Queues");
+
+ // hold the list of existing queue names
+ final List<AMQShortString> existingQueues = new ArrayList<AMQShortString>();
+
+ final TupleBinding<QueueRecord> queueTupleBinding = _oldMessageStore.getQueueTupleBindingFactory().getInstance();
+
+ DatabaseVisitor queueVisitor = new DatabaseVisitor()
+ {
+ public void visit(DatabaseEntry key, DatabaseEntry value) throws AMQStoreException
+ {
+ QueueRecord queueRec = (QueueRecord) queueTupleBinding.entryToObject(value);
+ AMQShortString queueName = queueRec.getNameShortString();
+
+ //if the queue name is in the gathered list then set its exclusivity true
+ if (durableSubQueues.contains(queueName))
+ {
+ _logger.info("Marking as possible DurableSubscription backing queue: " + queueName);
+ queueRec.setExclusive(true);
+ }
+
+ //The simple call to createQueue with the QueueRecord object is sufficient for a v2->v3 upgrade as
+ //the extra 'exclusive' property in v3 will be defaulted to false in the record creation.
+ _newMessageStore.createQueue(queueRec);
+
+ _count++;
+ existingQueues.add(queueName);
+ }
+ };
+ _oldMessageStore.visitQueues(queueVisitor);
+
+ logCount(queueVisitor.getVisitedCount(), "Queue");
+
+
+ // Look for persistent messages stored for non-durable queues
+ _logger.info("Checking for messages previously sent to non-durable queues");
+
+ // track all message delivery to existing queues
+ final HashSet<Long> queueMessages = new HashSet<Long>();
+
+ // hold all non existing queues and their messages IDs
+ final HashMap<String, HashSet<Long>> phantomMessageQueues = new HashMap<String, HashSet<Long>>();
+
+ // delivery DB visitor to check message delivery and identify non existing queues
+ final QueueEntryTB queueEntryTB = new QueueEntryTB();
+ DatabaseVisitor messageDeliveryCheckVisitor = new DatabaseVisitor()
+ {
+ public void visit(DatabaseEntry key, DatabaseEntry value) throws DatabaseException
+ {
+ QueueEntryKey entryKey = (QueueEntryKey) queueEntryTB.entryToObject(key);
+ Long messageId = entryKey.getMessageId();
+ AMQShortString queueName = entryKey.getQueueName();
+ if (!existingQueues.contains(queueName))
+ {
+ String name = queueName.asString();
+ HashSet<Long> messages = phantomMessageQueues.get(name);
+ if (messages == null)
+ {
+ messages = new HashSet<Long>();
+ phantomMessageQueues.put(name, messages);
+ }
+ messages.add(messageId);
+ _count++;
+ }
+ else
+ {
+ queueMessages.add(messageId);
+ }
+ }
+ };
+ _oldMessageStore.visitDelivery(messageDeliveryCheckVisitor);
+
+ if (phantomMessageQueues.isEmpty())
+ {
+ _logger.info("No such messages were found");
+ }
+ else
+ {
+ _logger.info("Found " + messageDeliveryCheckVisitor.getVisitedCount()+ " such messages in total");
+
+ for (Entry<String, HashSet<Long>> phantomQueue : phantomMessageQueues.entrySet())
+ {
+ String queueName = phantomQueue.getKey();
+ HashSet<Long> messages = phantomQueue.getValue();
+
+ _logger.info(MessageFormat.format("There are {0} messages which were previously delivered to non-durable queue ''{1}''",messages.size(), queueName));
+
+ boolean createQueue;
+ if(!_interactive)
+ {
+ createQueue = true;
+ _logger.info("Running in batch-mode, marking queue as durable to ensure retention of the messages.");
+ }
+ else
+ {
+ createQueue = userInteract("Do you want to make this queue durable?\n"
+ + "NOTE: Answering No will result in these messages being discarded!");
+ }
+
+ if (createQueue)
+ {
+ for (Long messageId : messages)
+ {
+ queueMessages.add(messageId);
+ }
+ AMQShortString name = new AMQShortString(queueName);
+ existingQueues.add(name);
+ QueueRecord record = new QueueRecord(name, null, false, null);
+ _newMessageStore.createQueue(record);
+ }
+ }
+ }
+
+
+ //Migrate _messageMetaDataDb;
+ _logger.info("Message MetaData");
+
+ final Database newMetaDataDB = _newMessageStore.getMetaDataDb();
+ final TupleBinding<Object> oldMetaDataTupleBinding = _oldMessageStore.getMetaDataTupleBindingFactory().getInstance();
+ final TupleBinding<Object> newMetaDataTupleBinding = _newMessageStore.getMetaDataTupleBindingFactory().getInstance();
+
+ DatabaseVisitor metaDataVisitor = new DatabaseVisitor()
+ {
+ public void visit(DatabaseEntry key, DatabaseEntry value) throws DatabaseException
+ {
+ _count++;
+ MessageMetaData metaData = (MessageMetaData) oldMetaDataTupleBinding.entryToObject(value);
+
+ // get message id
+ Long messageId = TupleBinding.getPrimitiveBinding(Long.class).entryToObject(key);
+
+ // ONLY copy data if message is delivered to existing queue
+ if (!queueMessages.contains(messageId))
+ {
+ return;
+ }
+ DatabaseEntry newValue = new DatabaseEntry();
+ newMetaDataTupleBinding.objectToEntry(metaData, newValue);
+
+ newMetaDataDB.put(null, key, newValue);
+ }
+ };
+ _oldMessageStore.visitMetaDataDb(metaDataVisitor);
+
+ logCount(metaDataVisitor.getVisitedCount(), "Message MetaData");
+
+
+ //Migrate _messageContentDb;
+ _logger.info("Message Contents");
+ final Database newContentDB = _newMessageStore.getContentDb();
+
+ final TupleBinding<MessageContentKey> oldContentKeyTupleBinding = new MessageContentKeyTB_4();
+ final TupleBinding<MessageContentKey> newContentKeyTupleBinding = new MessageContentKeyTB_5();
+ final TupleBinding contentTB = new ContentTB();
+
+ DatabaseVisitor contentVisitor = new DatabaseVisitor()
+ {
+ long _prevMsgId = -1; //Initialise to invalid value
+ int _bytesSeenSoFar = 0;
+
+ public void visit(DatabaseEntry key, DatabaseEntry value) throws DatabaseException
+ {
+ _count++;
+
+ //determine the msgId of the current entry
+ MessageContentKey_4 contentKey = (MessageContentKey_4) oldContentKeyTupleBinding.entryToObject(key);
+ long msgId = contentKey.getMessageId();
+
+ // ONLY copy data if message is delivered to existing queue
+ if (!queueMessages.contains(msgId))
+ {
+ return;
+ }
+ //if this is a new message, restart the byte offset count.
+ if(_prevMsgId != msgId)
+ {
+ _bytesSeenSoFar = 0;
+ }
+
+ //determine the content size
+ ByteBuffer content = (ByteBuffer) contentTB.entryToObject(value);
+ int contentSize = content.limit();
+
+ //create the new key: id + previously seen data count
+ MessageContentKey_5 newKey = new MessageContentKey_5(msgId, _bytesSeenSoFar);
+ DatabaseEntry newKeyEntry = new DatabaseEntry();
+ newContentKeyTupleBinding.objectToEntry(newKey, newKeyEntry);
+
+ DatabaseEntry newValueEntry = new DatabaseEntry();
+ contentTB.objectToEntry(content, newValueEntry);
+
+ newContentDB.put(null, newKeyEntry, newValueEntry);
+
+ _prevMsgId = msgId;
+ _bytesSeenSoFar += contentSize;
+ }
+ };
+ _oldMessageStore.visitContentDb(contentVisitor);
+
+ logCount(contentVisitor.getVisitedCount(), "Message Content");
+
+
+ //Migrate _deliveryDb;
+ _logger.info("Delivery Records");
+ final Database deliveryDb =_newMessageStore.getDeliveryDb();
+ DatabaseVisitor deliveryDbVisitor = new DatabaseVisitor()
+ {
+
+ public void visit(DatabaseEntry key, DatabaseEntry value) throws DatabaseException
+ {
+ _count++;
+
+ // get message id from entry key
+ QueueEntryKey entryKey = (QueueEntryKey) queueEntryTB.entryToObject(key);
+ AMQShortString queueName = entryKey.getQueueName();
+
+ // ONLY copy data if message queue exists
+ if (existingQueues.contains(queueName))
+ {
+ deliveryDb.put(null, key, value);
+ }
+ }
+ };
+ _oldMessageStore.visitDelivery(deliveryDbVisitor);
+ logCount(contentVisitor.getVisitedCount(), "Delivery Record");
+ }
+
+ /**
+ * Log the specified count for item in a user friendly way.
+ *
+ * @param count of items to log
+ * @param item description of what is being logged.
+ */
+ private void logCount(int count, String item)
+ {
+ _logger.info(" " + count + " " + item + " " + (count == 1 ? "entry" : "entries"));
+ }
+
+ /**
+ * @param oldDatabase The old MessageStoreDB to perform the visit on
+ * @param newDatabase The new MessageStoreDB to copy the data to.
+ * @param contentName The string name of the content for display purposes.
+ *
+ * @throws AMQException Due to createQueue thorwing AMQException
+ * @throws DatabaseException If there is a problem with the loading of the data
+ */
+ private void moveContents(Database oldDatabase, final Database newDatabase, String contentName) throws AMQException, DatabaseException
+ {
+
+ DatabaseVisitor moveVisitor = new DatabaseVisitor()
+ {
+ public void visit(DatabaseEntry key, DatabaseEntry value) throws DatabaseException
+ {
+ _count++;
+ newDatabase.put(null, key, value);
+ }
+ };
+
+ _oldMessageStore.visitDatabase(oldDatabase, moveVisitor);
+
+ logCount(moveVisitor.getVisitedCount(), contentName);
+ }
+
+ private static void usage()
+ {
+ System.out.println("usage: BDBStoreUpgrade:\n [-h|--help] [-q|--quiet] [-f|--force] [-b|--backup <Path to backup-db>] " +
+ "-i|--input <Path to input-db> [-o|--output <Path to upgraded-db>]");
+ }
+
+ private static void help()
+ {
+ System.out.println("usage: BDBStoreUpgrade:");
+ System.out.println("Required:");
+ for (Object obj : _options.getOptions())
+ {
+ Option option = (Option) obj;
+ if (option.isRequired())
+ {
+ System.out.println("-" + option.getOpt() + "|--" + option.getLongOpt() + "\t\t-\t" + option.getDescription());
+ }
+ }
+
+ System.out.println("\nOptions:");
+ for (Object obj : _options.getOptions())
+ {
+ Option option = (Option) obj;
+ if (!option.isRequired())
+ {
+ System.out.println("--" + option.getLongOpt() + "|-" + option.getOpt() + "\t\t-\t" + option.getDescription());
+ }
+ }
+ }
+
+ static boolean isDirectoryAStoreDir(File directory)
+ {
+ if (directory.isFile())
+ {
+ return false;
+ }
+
+ for (File file : directory.listFiles())
+ {
+ if (file.isFile())
+ {
+ if (file.getName().endsWith(BDB_FILE_ENDING))
+ {
+ return true;
+ }
+ }
+ }
+ return false;
+ }
+
+ static File[] discoverDBStores(File fromDir)
+ {
+ if (!fromDir.exists())
+ {
+ throw new IllegalArgumentException("'" + fromDir + "' does not exist unable to upgrade.");
+ }
+
+ // Ensure we are given a directory
+ if (fromDir.isFile())
+ {
+ throw new IllegalArgumentException("'" + fromDir + "' is not a directory unable to upgrade.");
+ }
+
+ // Check to see if we have been given a single directory
+ if (isDirectoryAStoreDir(fromDir))
+ {
+ return new File[]{fromDir};
+ }
+
+ // Check to see if we have been give a directory containing stores.
+ List<File> stores = new LinkedList<File>();
+
+ for (File directory : fromDir.listFiles())
+ {
+ if (directory.isDirectory())
+ {
+ if (isDirectoryAStoreDir(directory))
+ {
+ stores.add(directory);
+ }
+ }
+ }
+
+ return stores.toArray(new File[stores.size()]);
+ }
+
+ private static void performDBBackup(File source, File backup, boolean force) throws Exception
+ {
+ if (backup.exists())
+ {
+ if (force)
+ {
+ _logger.info("Backup location exists. Forced to remove.");
+ FileUtils.delete(backup, true);
+ }
+ else
+ {
+ throw new IllegalArgumentException("Unable to perform backup a backup already exists.");
+ }
+ }
+
+ try
+ {
+ _logger.info("Backing up '" + source + "' to '" + backup + "'");
+ FileUtils.copyRecursive(source, backup);
+ }
+ catch (FileNotFoundException e)
+ {
+ //Throwing IAE here as this will be reported as a Backup not started
+ throw new IllegalArgumentException("Unable to perform backup:" + e.getMessage());
+ }
+ catch (FileUtils.UnableToCopyException e)
+ {
+ //Throwing exception here as this will be reported as a Failed Backup
+ throw new Exception("Unable to perform backup due to:" + e.getMessage());
+ }
+ }
+
+ public static void main(String[] args) throws ParseException
+ {
+ setOptions(_options);
+
+ final Options helpOptions = new Options();
+ setHelpOptions(helpOptions);
+
+ //Display help
+ boolean displayHelp = false;
+ try
+ {
+ if (new PosixParser().parse(helpOptions, args).hasOption("h"))
+ {
+ showHelp();
+ }
+ }
+ catch (ParseException pe)
+ {
+ displayHelp = true;
+ }
+
+ //Parse commandline for required arguments
+ try
+ {
+ _commandLine = new PosixParser().parse(_options, args);
+ }
+ catch (ParseException mae)
+ {
+ if (displayHelp)
+ {
+ showHelp();
+ }
+ else
+ {
+ fatalError(mae.getMessage());
+ }
+ }
+
+ String fromDir = _commandLine.getOptionValue(OPTION_INPUT_SHORT);
+ String toDir = _commandLine.getOptionValue(OPTION_OUTPUT_SHORT);
+ String backupDir = _commandLine.getOptionValue(OPTION_BACKUP_SHORT);
+
+ if (backupDir == null && _commandLine.hasOption(OPTION_BACKUP_SHORT))
+ {
+ backupDir = "";
+ }
+
+ //Attempt to locate possible Store to upgrade on input path
+ File[] stores = new File[0];
+ try
+ {
+ stores = discoverDBStores(new File(fromDir));
+ }
+ catch (IllegalArgumentException iae)
+ {
+ fatalError(iae.getMessage());
+ }
+
+ boolean interactive = !_commandLine.hasOption(OPTION_QUIET_SHORT);
+ boolean force = _commandLine.hasOption(OPTION_FORCE_SHORT);
+
+ try{
+ for (File store : stores)
+ {
+
+ // if toDir is null then we are upgrading inplace so we don't need
+ // to provide an upgraded toDir when upgrading multiple stores.
+ if (toDir == null ||
+ // Check to see if we are upgrading a store specified in
+ // fromDir or if the directories are nested.
+ (stores.length > 0
+ && stores[0].toString().length() == fromDir.length()))
+ {
+ upgrade(store, toDir, backupDir, interactive, force);
+ }
+ else
+ {
+ // Add the extra part of path from store to the toDir
+ upgrade(store, toDir + File.separator + store.toString().substring(fromDir.length()), backupDir, interactive, force);
+ }
+ }
+ }
+ catch (RuntimeException re)
+ {
+ if (!(USER_ABORTED_PROCESS).equals(re.getMessage()))
+ {
+ re.printStackTrace();
+ _logger.error("Upgrade Failed: " + re.getMessage());
+ }
+ else
+ {
+ _logger.error("Upgrade stopped : User aborted");
+ }
+ }
+
+ }
+
+ @SuppressWarnings("static-access")
+ private static void setOptions(Options options)
+ {
+ Option input =
+ OptionBuilder.isRequired().hasArg().withDescription("Location (Path) of store to upgrade.").withLongOpt(OPTION_INPUT)
+ .create(OPTION_INPUT_SHORT);
+
+ Option output =
+ OptionBuilder.hasArg().withDescription("Location (Path) for the upgraded-db to be written.").withLongOpt(OPTION_OUTPUT)
+ .create(OPTION_OUTPUT_SHORT);
+
+ Option quiet = new Option(OPTION_QUIET_SHORT, OPTION_QUIET, false, "Disable interactive options.");
+
+ Option force = new Option(OPTION_FORCE_SHORT, OPTION_FORCE, false, "Force upgrade removing any existing upgrade target.");
+ Option backup =
+ OptionBuilder.hasOptionalArg().withDescription("Location (Path) for the backup-db to be written.").withLongOpt(OPTION_BACKUP)
+ .create(OPTION_BACKUP_SHORT);
+
+ options.addOption(input);
+ options.addOption(output);
+ options.addOption(quiet);
+ options.addOption(force);
+ options.addOption(backup);
+ setHelpOptions(options);
+ }
+
+ private static void setHelpOptions(Options options)
+ {
+ options.addOption(new Option("h", "help", false, "Show this help."));
+ }
+
+ static void upgrade(File fromDir, String toDir, String backupDir, boolean interactive, boolean force)
+ {
+
+ _logger.info("Running BDB Message Store upgrade tool: v" + VERSION);
+ int version = getStoreVersion(fromDir);
+ if (!isVersionUpgradable(version))
+ {
+ return;
+ }
+ try
+ {
+ new BDBStoreUpgrade(fromDir.toString(), toDir, backupDir, interactive, force).upgradeFromVersion(version);
+
+ _logger.info("Upgrade complete.");
+ }
+ catch (IllegalArgumentException iae)
+ {
+ _logger.error("Upgrade not started due to: " + iae.getMessage());
+ }
+ catch (DatabaseException de)
+ {
+ de.printStackTrace();
+ _logger.error("Upgrade Failed: " + de.getMessage());
+ }
+ catch (RuntimeException re)
+ {
+ if (!(USER_ABORTED_PROCESS).equals(re.getMessage()))
+ {
+ re.printStackTrace();
+ _logger.error("Upgrade Failed: " + re.getMessage());
+ }
+ else
+ {
+ throw re;
+ }
+ }
+ catch (Exception e)
+ {
+ e.printStackTrace();
+ _logger.error("Upgrade Failed: " + e.getMessage());
+ }
+ }
+
+ /**
+ * Utility method to verify if store of given version can be upgraded.
+ *
+ * @param version
+ * store version to verify
+ * @return true if store can be upgraded, false otherwise
+ */
+ protected static boolean isVersionUpgradable(int version)
+ {
+ boolean storeUpgradable = false;
+ if (version == 0)
+ {
+ _logger.error("Existing store version is undefined!");
+ }
+ else if (version < LOWEST_SUPPORTED_STORE_VERSION)
+ {
+ _logger.error(MessageFormat.format(PREVIOUS_STORE_VERSION_UNSUPPORTED,
+ new Object[] { Integer.toString(version) }));
+ }
+ else if (version == BDBMessageStore.DATABASE_FORMAT_VERSION)
+ {
+ _logger.error(MessageFormat.format(STORE_ALREADY_UPGRADED, new Object[] { Integer.toString(version) }));
+ }
+ else if (version > BDBMessageStore.DATABASE_FORMAT_VERSION)
+ {
+ _logger.error(MessageFormat.format(FOLLOWING_STORE_VERSION_UNSUPPORTED,
+ new Object[] { Integer.toString(version) }));
+ }
+ else
+ {
+ _logger.info("Existing store version is " + version);
+ storeUpgradable = true;
+ }
+ return storeUpgradable;
+ }
+
+ /**
+ * Detects existing store version by checking list of database in store
+ * environment
+ *
+ * @param fromDir
+ * store folder
+ * @return version
+ */
+ public static int getStoreVersion(File fromDir)
+ {
+ int version = 0;
+ EnvironmentConfig envConfig = new EnvironmentConfig();
+ envConfig.setAllowCreate(false);
+ envConfig.setTransactional(false);
+ envConfig.setReadOnly(true);
+ Environment environment = null;
+ try
+ {
+
+ environment = new Environment(fromDir, envConfig);
+ List<String> databases = environment.getDatabaseNames();
+ for (String name : databases)
+ {
+ if (name.startsWith("exchangeDb"))
+ {
+ if (name.startsWith("exchangeDb_v"))
+ {
+ version = Integer.parseInt(name.substring(12));
+ }
+ else
+ {
+ version = 1;
+ }
+ break;
+ }
+ }
+ }
+ catch (Exception e)
+ {
+ _logger.error("Failure to open existing database: " + e.getMessage());
+ }
+ finally
+ {
+ if (environment != null)
+ {
+ try
+ {
+ environment.close();
+ }
+ catch (Exception e)
+ {
+ // ignoring. It should never happen.
+ }
+ }
+ }
+ return version;
+ }
+
+ private static void fatalError(String message)
+ {
+ System.out.println(message);
+ usage();
+ System.exit(1);
+ }
+
+ private static void showHelp()
+ {
+ help();
+ System.exit(0);
+ }
+
+}
Added: qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/BindingKey.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/BindingKey.java?rev=1175235&view=auto
==============================================================================
--- qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/BindingKey.java (added)
+++ qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/BindingKey.java Sat Sep 24 20:16:00 2011
@@ -0,0 +1,62 @@
+/*
+ *
+ * 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.framing.AMQShortString;
+import org.apache.qpid.framing.FieldTable;
+
+public class BindingKey extends Object
+{
+ private final AMQShortString _exchangeName;
+ private final AMQShortString _queueName;
+ private final AMQShortString _routingKey;
+ private final FieldTable _arguments;
+
+ public BindingKey(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;
+ }
+
+}
Added: qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/ContentTB.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/ContentTB.java?rev=1175235&view=auto
==============================================================================
--- qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/ContentTB.java (added)
+++ qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/ContentTB.java Sat Sep 24 20:16:00 2011
@@ -0,0 +1,52 @@
+/*
+ *
+ * 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 java.nio.ByteBuffer;
+
+import com.sleepycat.bind.tuple.TupleBinding;
+import com.sleepycat.bind.tuple.TupleInput;
+import com.sleepycat.bind.tuple.TupleOutput;
+
+public class ContentTB extends TupleBinding
+{
+ public Object entryToObject(TupleInput tupleInput)
+ {
+
+ final int size = tupleInput.readInt();
+ byte[] underlying = new byte[size];
+ tupleInput.readFast(underlying);
+ return ByteBuffer.wrap(underlying);
+ }
+
+ public void objectToEntry(Object object, TupleOutput tupleOutput)
+ {
+ ByteBuffer src = (ByteBuffer) object;
+
+ src = src.slice();
+
+ byte[] chunkData = new byte[src.limit()];
+ src.duplicate().get(chunkData);
+
+ tupleOutput.writeInt(chunkData.length);
+ tupleOutput.writeFast(chunkData);
+ }
+}
Added: qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/DatabaseVisitor.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/DatabaseVisitor.java?rev=1175235&view=auto
==============================================================================
--- qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/DatabaseVisitor.java (added)
+++ qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/DatabaseVisitor.java Sat Sep 24 20:16:00 2011
@@ -0,0 +1,44 @@
+/*
+ *
+ * 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.AMQStoreException;
+
+import com.sleepycat.je.DatabaseEntry;
+import com.sleepycat.je.DatabaseException;
+
+/** Visitor Interface so that each DatabaseEntry for a database can easily be processed. */
+public abstract class DatabaseVisitor
+{
+ protected int _count;
+
+ abstract public void visit(DatabaseEntry entry, DatabaseEntry value) throws AMQStoreException, DatabaseException;
+
+ public int getVisitedCount()
+ {
+ return _count;
+ }
+
+ public void resetVisitCount()
+ {
+ _count = 0;
+ }
+}
Added: qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/ExchangeTB.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/ExchangeTB.java?rev=1175235&view=auto
==============================================================================
--- qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/ExchangeTB.java (added)
+++ qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/ExchangeTB.java Sat Sep 24 20:16:00 2011
@@ -0,0 +1,58 @@
+/*
+ *
+ * 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 com.sleepycat.bind.tuple.TupleBinding;
+import com.sleepycat.bind.tuple.TupleInput;
+import com.sleepycat.bind.tuple.TupleOutput;
+import org.apache.log4j.Logger;
+import org.apache.qpid.framing.AMQShortString;
+import org.apache.qpid.server.store.berkeleydb.records.ExchangeRecord;
+
+public class ExchangeTB extends TupleBinding
+{
+ private static final Logger _log = Logger.getLogger(ExchangeTB.class);
+
+ public ExchangeTB()
+ {
+ }
+
+ public Object 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(Object object, TupleOutput tupleOutput)
+ {
+ ExchangeRecord exchange = (ExchangeRecord) object;
+
+ AMQShortStringEncoding.writeShortString(exchange.getNameShortString(), tupleOutput);
+ AMQShortStringEncoding.writeShortString(exchange.getType(), tupleOutput);
+
+ tupleOutput.writeBoolean(exchange.isAutoDelete());
+ }
+}
Added: qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/FieldTableEncoding.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/FieldTableEncoding.java?rev=1175235&view=auto
==============================================================================
--- qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/FieldTableEncoding.java (added)
+++ qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/FieldTableEncoding.java Sat Sep 24 20:16:00 2011
@@ -0,0 +1,74 @@
+/*
+ *
+ * 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.framing.FieldTable;
+
+import com.sleepycat.bind.tuple.TupleInput;
+import com.sleepycat.bind.tuple.TupleOutput;
+import com.sleepycat.je.DatabaseException;
+
+import java.io.ByteArrayInputStream;
+import java.io.DataInputStream;
+import java.io.IOException;
+
+public class FieldTableEncoding
+{
+ public static FieldTable readFieldTable(TupleInput tupleInput) throws DatabaseException
+ {
+ long length = tupleInput.readLong();
+ if (length <= 0)
+ {
+ return null;
+ }
+ else
+ {
+
+ byte[] data = new byte[(int)length];
+ tupleInput.readFast(data);
+
+ try
+ {
+ return new FieldTable(new DataInputStream(new ByteArrayInputStream(data)),length);
+ }
+ catch (IOException e)
+ {
+ throw new RuntimeException(e);
+ }
+
+ }
+
+ }
+
+ public static void writeFieldTable(FieldTable fieldTable, TupleOutput tupleOutput)
+ {
+
+ if (fieldTable == null)
+ {
+ tupleOutput.writeLong(0);
+ }
+ else
+ {
+ tupleOutput.writeLong(fieldTable.getEncodedSize());
+ tupleOutput.writeFast(fieldTable.getDataAsBytes());
+ }
+ }
+}
Added: qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/MessageContentKey.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/MessageContentKey.java?rev=1175235&view=auto
==============================================================================
--- qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/MessageContentKey.java (added)
+++ qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/MessageContentKey.java Sat Sep 24 20:16:00 2011
@@ -0,0 +1,42 @@
+/*
+ *
+ * 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;
+
+public class MessageContentKey
+{
+ private long _messageId;
+
+ public MessageContentKey(long messageId)
+ {
+ _messageId = messageId;
+ }
+
+
+ public long getMessageId()
+ {
+ return _messageId;
+ }
+
+ public void setMessageId(long messageId)
+ {
+ this._messageId = messageId;
+ }
+}
\ No newline at end of file
Added: qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/QueueEntryKey.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/QueueEntryKey.java?rev=1175235&view=auto
==============================================================================
--- qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/QueueEntryKey.java (added)
+++ qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/QueueEntryKey.java Sat Sep 24 20:16:00 2011
@@ -0,0 +1,49 @@
+/*
+ *
+ * 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.framing.AMQShortString;
+
+public class QueueEntryKey
+{
+ private AMQShortString _queueName;
+ private long _messageId;
+
+
+ public QueueEntryKey(AMQShortString queueName, long messageId)
+ {
+ _queueName = queueName;
+ _messageId = messageId;
+ }
+
+
+ public AMQShortString getQueueName()
+ {
+ return _queueName;
+ }
+
+
+ public long getMessageId()
+ {
+ return _messageId;
+ }
+
+}
Added: qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/keys/MessageContentKey_4.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/keys/MessageContentKey_4.java?rev=1175235&view=auto
==============================================================================
--- qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/keys/MessageContentKey_4.java (added)
+++ qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/keys/MessageContentKey_4.java Sat Sep 24 20:16:00 2011
@@ -0,0 +1,44 @@
+/*
+ *
+ * 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.keys;
+
+import org.apache.qpid.server.store.berkeleydb.MessageContentKey;
+
+public class MessageContentKey_4 extends MessageContentKey
+{
+ private int _chunkNum;
+
+ public MessageContentKey_4(long messageId, int chunkNo)
+ {
+ super(messageId);
+ _chunkNum = chunkNo;
+ }
+
+ public int getChunk()
+ {
+ return _chunkNum;
+ }
+
+ public void setChunk(int chunk)
+ {
+ this._chunkNum = chunk;
+ }
+}
\ No newline at end of file
Added: qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/keys/MessageContentKey_5.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/keys/MessageContentKey_5.java?rev=1175235&view=auto
==============================================================================
--- qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/keys/MessageContentKey_5.java (added)
+++ qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/keys/MessageContentKey_5.java Sat Sep 24 20:16:00 2011
@@ -0,0 +1,44 @@
+/*
+ *
+ * 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.keys;
+
+import org.apache.qpid.server.store.berkeleydb.MessageContentKey;
+
+public class MessageContentKey_5 extends MessageContentKey
+{
+ private int _offset;
+
+ public MessageContentKey_5(long messageId, int chunkNo)
+ {
+ super(messageId);
+ _offset = chunkNo;
+ }
+
+ public int getOffset()
+ {
+ return _offset;
+ }
+
+ public void setOffset(int chunk)
+ {
+ this._offset = chunk;
+ }
+}
\ No newline at end of file
Added: qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/records/ExchangeRecord.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/records/ExchangeRecord.java?rev=1175235&view=auto
==============================================================================
--- qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/records/ExchangeRecord.java (added)
+++ qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/records/ExchangeRecord.java Sat Sep 24 20:16:00 2011
@@ -0,0 +1,53 @@
+/*
+ *
+ * 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.records;
+
+import org.apache.qpid.framing.AMQShortString;
+
+public 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;
+ }
+
+}
Added: qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/records/QueueRecord.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/records/QueueRecord.java?rev=1175235&view=auto
==============================================================================
--- qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/records/QueueRecord.java (added)
+++ qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/records/QueueRecord.java Sat Sep 24 20:16:00 2011
@@ -0,0 +1,66 @@
+/*
+ *
+ * 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.records;
+
+import org.apache.qpid.framing.AMQShortString;
+import org.apache.qpid.framing.FieldTable;
+
+public class QueueRecord extends Object
+{
+ private final AMQShortString _queueName;
+ private final AMQShortString _owner;
+ private final FieldTable _arguments;
+ private boolean _exclusive;
+
+ public QueueRecord(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;
+ }
+
+}
Added: qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/testclient/BackupTestClient.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/testclient/BackupTestClient.java?rev=1175235&view=auto
==============================================================================
--- qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/testclient/BackupTestClient.java (added)
+++ qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/testclient/BackupTestClient.java Sat Sep 24 20:16:00 2011
@@ -0,0 +1,120 @@
+/*
+ *
+ * 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.testclient;
+
+import org.apache.log4j.Logger;
+
+import org.apache.qpid.ping.PingDurableClient;
+import org.apache.qpid.server.store.berkeleydb.BDBBackup;
+import org.apache.qpid.util.CommandLineParser;
+
+import java.util.Properties;
+
+/**
+ * BackupTestClient extends {@link PingDurableClient} with an action that takes a BDB backup when a configurable
+ * message count is reached. This enables a test user to restore this beackup, knowing how many committed but undelivered
+ * messages were in the backup, in order to check that all are re-delivered when the backup is retored.
+ *
+ * <p><table id="crc"><caption>CRC Card</caption>
+ * <tr><th> Responsibilities <th> Collaborations
+ * <tr><td> Perform BDB Backup on configurable message count.
+ * </table>
+ */
+public class BackupTestClient extends PingDurableClient
+{
+ /** Used for debugging. */
+ private static final Logger log = Logger.getLogger(BackupTestClient.class);
+
+ /** Holds the from directory to take backups from. */
+ private String fromDir;
+
+ /** Holds the to directory to store backups in. */
+ private String toDir;
+
+ /**
+ * Default constructor, passes all property overrides to the parent.
+ *
+ * @param overrides Any property overrides to apply to the defaults.
+ *
+ * @throws Exception Any underlying exception is allowed to fall through.
+ */
+ BackupTestClient(Properties overrides) throws Exception
+ {
+ super(overrides);
+ }
+
+ /**
+ * Starts the ping/wait/receive process. From and to directory locations for the BDB backups must be specified
+ * on the command line:
+ *
+ * <p/><table><caption>Command Line</caption>
+ * <tr><th> Option <th> Comment
+ * <tr><td> -fromdir <td> The path to the directory to back the bdb log file from.
+ * <tr><td> -todir <td> The path to the directory to save the backed up bdb log files to.
+ * </table>
+ *
+ * @param args The command line arguments.
+ */
+ public static void main(String[] args)
+ {
+ try
+ {
+ // Use the same command line format as BDBBackup utility, (compulsory from and to directories).
+ Properties options =
+ CommandLineParser.processCommandLine(args, new CommandLineParser(BDBBackup.COMMAND_LINE_SPEC),
+ System.getProperties());
+ BackupTestClient pingProducer = new BackupTestClient(options);
+
+ // Keep the from and to directories for backups.
+ pingProducer.fromDir = options.getProperty("fromdir");
+ pingProducer.toDir = options.getProperty("todir");
+
+ // Create a shutdown hook to terminate the ping-pong producer.
+ Runtime.getRuntime().addShutdownHook(pingProducer.getShutdownHook());
+
+ // Ensure that the ping pong producer is registered to listen for exceptions on the connection too.
+ // pingProducer.getConnection().setExceptionListener(pingProducer);
+
+ // Run the test procedure.
+ int sent = pingProducer.send();
+ pingProducer.waitForUser("Press return to begin receiving the pings.");
+ pingProducer.receive(sent);
+
+ System.exit(0);
+ }
+ catch (Exception e)
+ {
+ System.err.println(e.getMessage());
+ log.error("Top level handler caught execption.", e);
+ System.exit(1);
+ }
+ }
+
+ /**
+ * Supplies a triggered action extension, based on message count. This action takes a BDB log file backup.
+ */
+ public void takeAction()
+ {
+ BDBBackup backupUtil = new BDBBackup();
+ backupUtil.takeBackupNoLock(fromDir, toDir);
+ System.out.println("Took backup of BDB log files from directory: " + fromDir);
+ }
+}
Added: qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/BindingTuple.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/BindingTuple.java?rev=1175235&view=auto
==============================================================================
--- qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/BindingTuple.java (added)
+++ qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/BindingTuple.java Sat Sep 24 20:16:00 2011
@@ -0,0 +1,25 @@
+/*
+ *
+ * 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.tuples;
+
+public interface BindingTuple
+{
+}
Added: qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/BindingTupleBindingFactory.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/BindingTupleBindingFactory.java?rev=1175235&view=auto
==============================================================================
--- qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/BindingTupleBindingFactory.java (added)
+++ qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/BindingTupleBindingFactory.java Sat Sep 24 20:16:00 2011
@@ -0,0 +1,45 @@
+/*
+ *
+ * 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.tuples;
+
+import org.apache.qpid.server.store.berkeleydb.BindingKey;
+
+import com.sleepycat.bind.tuple.TupleBinding;
+
+public class BindingTupleBindingFactory extends TupleBindingFactory<BindingKey>
+{
+ public BindingTupleBindingFactory(int version)
+ {
+ super(version);
+ }
+
+ public TupleBinding<BindingKey> getInstance()
+ {
+ switch (_version)
+ {
+ default:
+ case 5:
+ //no change from v4
+ case 4:
+ return new BindingTuple_4();
+ }
+ }
+}
Added: qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/BindingTuple_4.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/BindingTuple_4.java?rev=1175235&view=auto
==============================================================================
--- qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/BindingTuple_4.java (added)
+++ qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/BindingTuple_4.java Sat Sep 24 20:16:00 2011
@@ -0,0 +1,76 @@
+/*
+ *
+ * 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.tuples;
+
+import org.apache.log4j.Logger;
+import org.apache.qpid.server.store.berkeleydb.AMQShortStringEncoding;
+import org.apache.qpid.server.store.berkeleydb.BindingKey;
+import org.apache.qpid.server.store.berkeleydb.FieldTableEncoding;
+import org.apache.qpid.framing.AMQShortString;
+import org.apache.qpid.framing.FieldTable;
+
+import com.sleepycat.bind.tuple.TupleBinding;
+import com.sleepycat.bind.tuple.TupleInput;
+import com.sleepycat.bind.tuple.TupleOutput;
+import com.sleepycat.je.DatabaseException;
+
+public class BindingTuple_4 extends TupleBinding<BindingKey> implements BindingTuple
+{
+ protected static final Logger _log = Logger.getLogger(BindingTuple.class);
+
+ public BindingTuple_4()
+ {
+ super();
+ }
+
+ public BindingKey entryToObject(TupleInput tupleInput)
+ {
+ AMQShortString exchangeName = AMQShortStringEncoding.readShortString(tupleInput);
+ AMQShortString queueName = AMQShortStringEncoding.readShortString(tupleInput);
+ AMQShortString routingKey = AMQShortStringEncoding.readShortString(tupleInput);
+
+ FieldTable arguments;
+
+ // Addition for Version 2 of this table
+ try
+ {
+ arguments = FieldTableEncoding.readFieldTable(tupleInput);
+ }
+ catch (DatabaseException e)
+ {
+ _log.error("Unable to create binding: " + e, e);
+ return null;
+ }
+
+ return new BindingKey(exchangeName, queueName, routingKey, arguments);
+ }
+
+ public void objectToEntry(BindingKey binding, TupleOutput tupleOutput)
+ {
+ AMQShortStringEncoding.writeShortString(binding.getExchangeName(), tupleOutput);
+ AMQShortStringEncoding.writeShortString(binding.getQueueName(), tupleOutput);
+ AMQShortStringEncoding.writeShortString(binding.getRoutingKey(), tupleOutput);
+
+ // Addition for Version 2 of this table
+ FieldTableEncoding.writeFieldTable(binding.getArguments(), tupleOutput);
+ }
+
+}
\ No newline at end of file
Added: qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/MessageContentKeyTB_4.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/MessageContentKeyTB_4.java?rev=1175235&view=auto
==============================================================================
--- qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/MessageContentKeyTB_4.java (added)
+++ qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/MessageContentKeyTB_4.java Sat Sep 24 20:16:00 2011
@@ -0,0 +1,47 @@
+/*
+ *
+ * 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.tuples;
+
+import org.apache.qpid.server.store.berkeleydb.MessageContentKey;
+import org.apache.qpid.server.store.berkeleydb.keys.MessageContentKey_4;
+
+import com.sleepycat.bind.tuple.TupleBinding;
+import com.sleepycat.bind.tuple.TupleInput;
+import com.sleepycat.bind.tuple.TupleOutput;
+
+public class MessageContentKeyTB_4 extends TupleBinding<MessageContentKey>
+{
+
+ public MessageContentKey entryToObject(TupleInput tupleInput)
+ {
+ long messageId = tupleInput.readLong();
+ int chunk = tupleInput.readInt();
+ return new MessageContentKey_4(messageId, chunk);
+ }
+
+ public void objectToEntry(MessageContentKey object, TupleOutput tupleOutput)
+ {
+ final MessageContentKey_4 mk = (MessageContentKey_4) object;
+ tupleOutput.writeLong(mk.getMessageId());
+ tupleOutput.writeInt(mk.getChunk());
+ }
+
+}
\ No newline at end of file
Added: qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/MessageContentKeyTB_5.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/MessageContentKeyTB_5.java?rev=1175235&view=auto
==============================================================================
--- qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/MessageContentKeyTB_5.java (added)
+++ qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/MessageContentKeyTB_5.java Sat Sep 24 20:16:00 2011
@@ -0,0 +1,46 @@
+/*
+ *
+ * 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.tuples;
+
+import org.apache.qpid.server.store.berkeleydb.MessageContentKey;
+import org.apache.qpid.server.store.berkeleydb.keys.MessageContentKey_5;
+
+import com.sleepycat.bind.tuple.TupleBinding;
+import com.sleepycat.bind.tuple.TupleInput;
+import com.sleepycat.bind.tuple.TupleOutput;
+
+public class MessageContentKeyTB_5 extends TupleBinding<MessageContentKey>
+{
+ public MessageContentKey entryToObject(TupleInput tupleInput)
+ {
+ long messageId = tupleInput.readLong();
+ int offset = tupleInput.readInt();
+ return new MessageContentKey_5(messageId, offset);
+ }
+
+ public void objectToEntry(MessageContentKey object, TupleOutput tupleOutput)
+ {
+ final MessageContentKey_5 mk = (MessageContentKey_5) object;
+ tupleOutput.writeLong(mk.getMessageId());
+ tupleOutput.writeInt(mk.getOffset());
+ }
+
+}
\ No newline at end of file
Added: qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/MessageContentKeyTupleBindingFactory.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/MessageContentKeyTupleBindingFactory.java?rev=1175235&view=auto
==============================================================================
--- qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/MessageContentKeyTupleBindingFactory.java (added)
+++ qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/MessageContentKeyTupleBindingFactory.java Sat Sep 24 20:16:00 2011
@@ -0,0 +1,45 @@
+/*
+ *
+ * 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.tuples;
+
+import org.apache.qpid.server.store.berkeleydb.MessageContentKey;
+
+import com.sleepycat.bind.tuple.TupleBinding;
+
+public class MessageContentKeyTupleBindingFactory extends TupleBindingFactory<MessageContentKey>
+{
+ public MessageContentKeyTupleBindingFactory(int version)
+ {
+ super(version);
+ }
+
+ public TupleBinding<MessageContentKey> getInstance()
+ {
+ switch (_version)
+ {
+ default:
+ case 5:
+ return new MessageContentKeyTB_5();
+ case 4:
+ return new MessageContentKeyTB_4();
+ }
+ }
+}
---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project: http://qpid.apache.org
Use/Interact: mailto:commits-subscribe@qpid.apache.org