[rhmessaging-commits] rhmessaging commits: r3971 - in store/trunk/java/bdbstore/src: main/java/org/apache/qpid/server/store/berkeleydb/keys and 4 other directories.
rhmessaging-commits at lists.jboss.org
rhmessaging-commits at lists.jboss.org
Fri May 14 12:24:01 EDT 2010
Author: rgemmell
Date: 2010-05-14 12:23:59 -0400 (Fri, 14 May 2010)
New Revision: 3971
Added:
store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/keys/
store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/keys/MessageContentKey_1.java
store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/keys/MessageContentKey_3.java
store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/records/
store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/records/ExchangeRecord.java
store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/records/QueueRecord.java
store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/MessageContentKeyTB_1.java
store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/MessageContentKeyTB_3.java
store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/MessageContentKeyTupleBindingFactory.java
store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/MessageMetaDataTB_1.java
store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/MessageMetaDataTB_3.java
store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/MessageMetaDataTupleBindingFactory.java
store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/QueueEntryTB.java
Removed:
store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/MessageMetaDataTB.java
Modified:
store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/BDBMessageStore.java
store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/ContentTB.java
store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/ExchangeTB.java
store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/MessageContentKey.java
store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/QueueEntryKey.java
store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/BindingTuple.java
store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/BindingTupleBindingFactory.java
store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/BindingTuple_1.java
store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/BindingTuple_2.java
store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/QueueTuple.java
store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/QueueTupleBindingFactory.java
store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/QueueTuple_1.java
store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/QueueTuple_2.java
store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/TupleBindingFactory.java
store/trunk/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/BDBUpgradeTest.java
store/trunk/java/bdbstore/src/tools/java/org/apache/qpid/server/store/berkeleydb/BDBStoreUpgrade.java
Log:
Initial update to implement the new RecoveryHandler store behaviour, plus other updates required since 0-10 support was added to the trunk broker.
Modified: store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/BDBMessageStore.java
===================================================================
--- store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/BDBMessageStore.java 2010-05-13 18:28:33 UTC (rev 3970)
+++ store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/BDBMessageStore.java 2010-05-14 16:23:59 UTC (rev 3971)
@@ -20,53 +20,66 @@
*/
package org.apache.qpid.server.store.berkeleydb;
-import com.sleepycat.bind.EntryBinding;
-import com.sleepycat.bind.tuple.ByteBinding;
-import com.sleepycat.bind.tuple.TupleBinding;
+import java.io.File;
+import java.lang.ref.SoftReference;
+import java.nio.ByteBuffer;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Queue;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicReference;
-import com.sleepycat.je.*;
-
import org.apache.commons.configuration.Configuration;
-
import org.apache.log4j.Logger;
-
import org.apache.qpid.AMQException;
import org.apache.qpid.framing.AMQShortString;
import org.apache.qpid.framing.FieldTable;
-import org.apache.qpid.framing.abstraction.ContentChunk;
import org.apache.qpid.server.exchange.Exchange;
-import org.apache.qpid.server.queue.AMQMessage;
+import org.apache.qpid.server.logging.LogSubject;
+import org.apache.qpid.server.logging.actors.CurrentActor;
+import org.apache.qpid.server.logging.messages.ConfigStoreMessages;
+import org.apache.qpid.server.logging.messages.MessageStoreMessages;
+import org.apache.qpid.server.logging.messages.TransactionLogMessages;
import org.apache.qpid.server.queue.AMQQueue;
-import org.apache.qpid.server.queue.MessageHandleFactory;
-import org.apache.qpid.server.queue.MessageMetaData;
-import org.apache.qpid.server.queue.QueueRegistry;
-import org.apache.qpid.server.queue.AMQQueueFactory;
+import org.apache.qpid.server.store.ConfigurationRecoveryHandler;
import org.apache.qpid.server.store.MessageStore;
+import org.apache.qpid.server.store.MessageStoreRecoveryHandler;
+import org.apache.qpid.server.store.StorableMessageMetaData;
import org.apache.qpid.server.store.StoreContext;
-import org.apache.qpid.server.store.AbstractMessageStore;
+import org.apache.qpid.server.store.StoredMemoryMessage;
+import org.apache.qpid.server.store.StoredMessage;
+import org.apache.qpid.server.store.TransactionLogRecoveryHandler;
+import org.apache.qpid.server.store.TransactionLogResource;
+import org.apache.qpid.server.store.ConfigurationRecoveryHandler.BindingRecoveryHandler;
+import org.apache.qpid.server.store.ConfigurationRecoveryHandler.ExchangeRecoveryHandler;
+import org.apache.qpid.server.store.ConfigurationRecoveryHandler.QueueRecoveryHandler;
+import org.apache.qpid.server.store.MessageStoreRecoveryHandler.StoredMessageRecoveryHandler;
+import org.apache.qpid.server.store.TransactionLogRecoveryHandler.QueueEntryRecoveryHandler;
+import org.apache.qpid.server.store.berkeleydb.keys.MessageContentKey_3;
+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.BindingTupleBindingFactory;
+import org.apache.qpid.server.store.berkeleydb.tuples.MessageContentKeyTB_3;
+import org.apache.qpid.server.store.berkeleydb.tuples.MessageMetaDataTupleBindingFactory;
+import org.apache.qpid.server.store.berkeleydb.tuples.QueueEntryTB;
import org.apache.qpid.server.store.berkeleydb.tuples.QueueTupleBindingFactory;
-import org.apache.qpid.server.store.berkeleydb.tuples.BindingTupleBindingFactory;
-import org.apache.qpid.server.store.berkeleydb.tuples.QueueTuple;
-import org.apache.qpid.server.txn.NonTransactionalContext;
-import org.apache.qpid.server.txn.TransactionalContext;
-import org.apache.qpid.server.virtualhost.VirtualHost;
-import org.apache.qpid.server.configuration.VirtualHostConfiguration;
-import org.apache.qpid.server.logging.actors.CurrentActor;
-import org.apache.qpid.server.logging.messages.MessageStoreMessages;
-import java.io.File;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Map;
-import java.util.Queue;
-import java.util.TreeMap;
-import java.util.concurrent.ConcurrentLinkedQueue;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicLong;
-import java.util.concurrent.atomic.AtomicReference;
+import com.sleepycat.bind.EntryBinding;
+import com.sleepycat.bind.tuple.ByteBinding;
+import com.sleepycat.bind.tuple.TupleBinding;
+import com.sleepycat.je.CheckpointConfig;
+import com.sleepycat.je.Cursor;
+import com.sleepycat.je.Database;
+import com.sleepycat.je.DatabaseConfig;
+import com.sleepycat.je.DatabaseEntry;
+import com.sleepycat.je.DatabaseException;
+import com.sleepycat.je.Environment;
+import com.sleepycat.je.EnvironmentConfig;
+import com.sleepycat.je.LockMode;
+import com.sleepycat.je.OperationStatus;
+import com.sleepycat.je.TransactionConfig;
/**
* BDBMessageStore implements a persistent {@link MessageStore} using the BDB high performance log.
@@ -76,62 +89,40 @@
* exchanges. <tr><td> Store and remove messages. <tr><td> Bind and unbind queues to exchanges. <tr><td> Enqueue and
* dequeue messages to queues. <tr><td> Generate message identifiers. </table>
*/
-public class BDBMessageStore extends AbstractMessageStore
+public class BDBMessageStore implements MessageStore
{
private static final Logger _log = Logger.getLogger(BDBMessageStore.class);
- private static final int DATABASE_FORMAT_VERSION = 2;
+ private static final int DATABASE_FORMAT_VERSION = 3;
private static final String DATABASE_FORMAT_VERSION_PROPERTY = "version";
-
public static final String ENVIRONMENT_PATH_PROPERTY = "environment-path";
private Environment _environment;
private String MESSAGEMETADATADB_NAME = "messageMetaDataDb";
-
- /**
- * Maps from messageId to an AMQMessage (note we don't use serialisation but this is what it roughly corresponds
- * to)
- */
+ private String MESSAGECONTENTDB_NAME = "messageContentDb";
+ private String QUEUEBINDINGSDB_NAME = "queueBindingsDb";
+ private String DELIVERYDB_NAME = "deliveryDb";
+ private String EXCHANGEDB_NAME = "exchangeDb";
+ private String QUEUEDB_NAME = "queueDb";
private Database _messageMetaDataDb;
-
- private String MESSAGECONTENTDB_NAME = "messageContentDb";
-
private Database _messageContentDb;
-
- private String QUEUEDB_NAME = "queueDb";
-
- /** Maps from name (which uniquely identifies a queue) to an AMQQueue */
- private Database _queueDb;
-
- private String DELIVERYDB_NAME = "deliveryDb";
-
- /** Maps from a queue name to a message id. This is what stores the pending deliveries for a given queue */
+ private Database _queueBindingsDb;
private Database _deliveryDb;
-
- private String EXCHANGEDB_NAME = "exchangeDb";
private Database _exchangeDb;
+ private Database _queueDb;
- private String QUEUEBINDINGSDB_NAME = "queueBindingsDb";
- private Database _queueBindingsDb;
+ private LogSubject _logSubject;
- private VirtualHost _virtualHost;
+ private final AtomicLong _messageId = new AtomicLong(0);
- private final AtomicLong _messageId = new AtomicLong(1);
-
- private final AtomicLong _queueId = new AtomicLong(1);
-
private final CommitThread _commitThread = new CommitThread("Commit-Thread");
- private Map<AMQShortString, Long> _queueNameToIdMap = new ConcurrentHashMap<AMQShortString, Long>();
-
- // Factory Classes to create the TupleBinding objects that relfect the version instance of this BDBStore
-
+ // Factory Classes to create the TupleBinding objects that reflect the version instance of this BDBStore
+ private MessageMetaDataTupleBindingFactory _metaDataTupleBindingFactory;
private QueueTupleBindingFactory _queueTupleBindingFactory;
private BindingTupleBindingFactory _bindingTupleBindingFactory;
- Map<AMQShortString, Integer> _queueRecoveries = new TreeMap<AMQShortString, Integer>();
-
/** The data version this store should run with */
private int _version;
private enum State
@@ -149,6 +140,9 @@
private TransactionConfig _transactionConfig = new TransactionConfig();
+ private boolean _configured;
+
+
public BDBMessageStore()
{
this(DATABASE_FORMAT_VERSION);
@@ -176,23 +170,78 @@
QUEUEBINDINGSDB_NAME += "_v" + version;
}
}
+
+ public void configureConfigStore(String name,
+ ConfigurationRecoveryHandler recoveryHandler,
+ Configuration storeConfiguration,
+ LogSubject logSubject) throws Exception
+ {
+ _logSubject = logSubject;
+ CurrentActor.get().message(_logSubject, ConfigStoreMessages.CFG_1001(this.getClass().getName()));
+ if(_configured)
+ {
+ throw new Exception("ConfigStore already configured");
+ }
+
+ configure(name,storeConfiguration);
+
+ _configured = true;
+ stateTransition(State.CONFIGURING, State.CONFIGURED);
+
+ recover(recoveryHandler);
+ stateTransition(State.RECOVERING, State.STARTED);
+ }
+
+ public void configureMessageStore(String name,
+ MessageStoreRecoveryHandler recoveryHandler,
+ Configuration storeConfiguration,
+ LogSubject logSubject) throws Exception
+ {
+ CurrentActor.get().message(_logSubject, MessageStoreMessages.MST_CREATED(this.getClass().getName()));
+
+ if(!_configured)
+ {
+ throw new Exception("ConfigStore not configured");
+ }
+
+ recoverMessages(recoveryHandler);
+ }
+
+ public void configureTransactionLog(String name, TransactionLogRecoveryHandler recoveryHandler,
+ Configuration storeConfiguration, LogSubject logSubject) throws Exception
+ {
+ CurrentActor.get().message(_logSubject, TransactionLogMessages.TXN_1001(this.getClass().getName()));
+
+ if(!_configured)
+ {
+ throw new Exception("ConfigStore not configured");
+ }
+
+ recoverQueueEntries(recoveryHandler);
+
+
+ }
+
+ public org.apache.qpid.server.store.TransactionLog.Transaction newTransaction()
+ {
+ return new BDBTransaction();
+ }
+
+
/**
- * Called after instantiation in order to configure the message store. A particular implementation can define
- * whatever parameters it wants.
+ * Called after instantiation in order to configure the message store.
*
- * @param virtualHost The virtual host using by this store
- * @param base Not used
+ * @param name The name of the virtual host using this store
* @param vHostConfig The configuration for this virtualhost
+ * @return whether a new store environment was created or not (to indicate whether recovery is necessary)
*
* @throws Exception If any error occurs that means the store is unable to configure itself.
*/
- public void configure(VirtualHost virtualHost, String base, VirtualHostConfiguration vHostConfig) throws Exception
+ public boolean configure(String name, Configuration storeConfig) throws Exception
{
- super.configure(virtualHost, base, vHostConfig);
-
- Configuration config = vHostConfig.getStoreConfiguration();
- File environmentPath = new File(config.getString(ENVIRONMENT_PATH_PROPERTY, System.getProperty("QPID_WORK") + "/bdbstore/" + virtualHost.getName()));
+ File environmentPath = new File(storeConfig.getString(ENVIRONMENT_PATH_PROPERTY,
+ System.getProperty("QPID_WORK") + "/bdbstore/" + name));
if (!environmentPath.exists())
{
if (!environmentPath.mkdirs())
@@ -204,53 +253,39 @@
CurrentActor.get().message(_logSubject, MessageStoreMessages.MST_STORE_LOCATION(environmentPath.getAbsolutePath()));
- _version = config.getInt(DATABASE_FORMAT_VERSION_PROPERTY, DATABASE_FORMAT_VERSION);
+ _version = storeConfig.getInt(DATABASE_FORMAT_VERSION_PROPERTY, DATABASE_FORMAT_VERSION);
- configure(virtualHost, environmentPath, false);
+ return configure(environmentPath, false);
}
- public void configure(File environmentPath) throws AMQException, DatabaseException
+
+ /**
+ *
+ * @param environmentPath location for the store to be created in/recovered from
+ * @param readonly if true then don't allow modifications to an existing store, and don't create a new store if none exists
+ * @return whether or not a new store environment was created
+ * @throws AMQException
+ * @throws DatabaseException
+ */
+ protected boolean configure(File environmentPath, boolean readonly) throws AMQException, DatabaseException
{
- configure(null, environmentPath, false);
- }
-
- public void configure(VirtualHost virtualHost, File environmentPath, boolean readonly) throws AMQException, DatabaseException
- {
stateTransition(State.INITIAL, State.CONFIGURING);
_log.info("Configuring BDB message store");
+ createTupleBindingFactories(_version);
+
setDatabaseNames(_version);
- if (virtualHost != null)
- {
- setVirtualHost(virtualHost);
- }
-
- boolean newEnvironment = setupStore(environmentPath, readonly);
-
- // Performing recovery when we only want read access will cause all the broker objects to be recreated
- // This will/may include thread pool creations that may be duplicated when manually inspecting the state of
- // the store. Simplest solution is to prevent the initial creation of the state by blocking recovery.
- if (!readonly)
- {
- stateTransition(State.CONFIGURING, State.CONFIGURED);
-
- //If we have loaded an environment and have virtualHost configured then recover environment
- if (!newEnvironment && virtualHost != null)
- {
- // this recovers durable queues and persistent messages
- recover(virtualHost);
- }
- }
-
- //if we have a new environment there we can jump to started as there is no recovery requried..
- if (newEnvironment) // && !readonly is implied as you cant get a newEnviroment in readonly mode.
- {
- stateTransition(State.CONFIGURED, State.STARTED);
- }
-
+ return setupStore(environmentPath, readonly);
}
+
+ private void createTupleBindingFactories(int version)
+ {
+ _bindingTupleBindingFactory = new BindingTupleBindingFactory(version);
+ _queueTupleBindingFactory = new QueueTupleBindingFactory(version);
+ _metaDataTupleBindingFactory = new MessageMetaDataTupleBindingFactory(version);
+ }
/**
* Move the store state from CONFIGURING to STARTED.
@@ -288,7 +323,8 @@
{
int versionIndex = s.indexOf("_v");
- // DB is v1 if _version is not v1 then error
+ // lack of _v index suggests DB is v1
+ // so if _version is not v1 then error
if (versionIndex == -1)
{
if (_version != 1)
@@ -315,12 +351,6 @@
}
}
- private void createTupleBindingFactories(int version)
- {
- _queueTupleBindingFactory = new QueueTupleBindingFactory(version, _virtualHost);
- _bindingTupleBindingFactory = new BindingTupleBindingFactory(version, _virtualHost);
- }
-
private synchronized void stateTransition(State requiredState, State newState) throws AMQException
{
if (_state != requiredState)
@@ -344,16 +374,15 @@
{
_log.info("BDB message store using environment path " + environmentPath.getAbsolutePath());
EnvironmentConfig envConfig = new EnvironmentConfig();
- // This is what allos the creation of the store if it does not already exist.
+ // This is what allows the creation of the store if it does not already exist.
envConfig.setAllowCreate(false);
envConfig.setTransactional(true);
envConfig.setConfigParam("je.lock.nLockTables", "7");
- // Restore 500,000 default timeout.
+ // Restore 500,000 default timeout.
//envConfig.setLockTimeout(15000);
// Added to help diagnosis of Deadlock issue
- //
// http://www.oracle.com/technology/products/berkeley-db/faq/je_faq.html#23
if (Boolean.getBoolean("qpid.bdb.lock.debug"))
{
@@ -467,8 +496,8 @@
closeEnvironment();
_state = State.CLOSED;
-
- super.close();
+
+ CurrentActor.get().message(_logSubject,MessageStoreMessages.MST_CLOSED());
}
private void closeEnvironment() throws DatabaseException
@@ -483,43 +512,275 @@
}
}
+
+ public void recover(ConfigurationRecoveryHandler recoveryHandler) throws AMQException
+ {
+ stateTransition(State.CONFIGURED, State.RECOVERING);
+
+ CurrentActor.get().message(_logSubject,MessageStoreMessages.MST_RECOVERY_START());
+
+ try
+ {
+ QueueRecoveryHandler qrh = recoveryHandler.begin(this);
+ loadQueues(qrh);
+
+ ExchangeRecoveryHandler erh = qrh.completeQueueRecovery();
+ loadExchanges(erh);
+
+ BindingRecoveryHandler brh = erh.completeExchangeRecovery();
+ recoverBindings(brh);
+
+ brh.completeBindingRecovery();
+ }
+ catch (DatabaseException e)
+ {
+ throw new AMQException("Error recovering persistent state: " + e, e);
+ }
+
+ }
+
+ private void loadQueues(QueueRecoveryHandler qrh) throws DatabaseException
+ {
+ Cursor cursor = null;
+
+ try
+ {
+ cursor = _queueDb.openCursor(null, null);
+ DatabaseEntry key = new DatabaseEntry();
+ DatabaseEntry value = new DatabaseEntry();
+ TupleBinding binding = _queueTupleBindingFactory.getInstance();
+ while (cursor.getNext(key, value, LockMode.RMW) == OperationStatus.SUCCESS)
+ {
+ QueueRecord queueRecord = (QueueRecord) binding.entryToObject(value);
+
+ String queueName = queueRecord.getNameShortString() == null ? null :
+ queueRecord.getNameShortString().asString();
+ String owner = queueRecord.getOwner() == null ? null :
+ queueRecord.getOwner().asString();
+ FieldTable arguments = queueRecord.getArguments();
+
+ qrh.queue(queueName, owner, arguments);
+ }
+
+ }
+ finally
+ {
+ if (cursor != null)
+ {
+ cursor.close();
+ }
+ }
+ }
+
+
+ private void loadExchanges(ExchangeRecoveryHandler erh) throws AMQException, DatabaseException
+ {
+ Cursor cursor = null;
+
+ try
+ {
+ cursor = _exchangeDb.openCursor(null, null);
+ DatabaseEntry key = new DatabaseEntry();
+ DatabaseEntry value = new DatabaseEntry();
+ TupleBinding binding = new ExchangeTB();
+
+ while (cursor.getNext(key, value, LockMode.RMW) == OperationStatus.SUCCESS)
+ {
+ ExchangeRecord exchangeRec = (ExchangeRecord) binding.entryToObject(value);
+
+ String exchangeName = exchangeRec.getNameShortString() == null ? null :
+ exchangeRec.getNameShortString().asString();
+ String type = exchangeRec.getType() == null ? null :
+ exchangeRec.getType().asString();
+ boolean autoDelete = exchangeRec.isAutoDelete();
+
+ erh.exchange(exchangeName, type, autoDelete);
+ }
+ }
+ finally
+ {
+ if (cursor != null)
+ {
+ cursor.close();
+ }
+ }
+
+ }
+
+ private void recoverBindings(BindingRecoveryHandler brh) throws DatabaseException
+ {
+ Cursor cursor = null;
+ try
+ {
+ cursor = _queueBindingsDb.openCursor(null, null);
+ DatabaseEntry key = new DatabaseEntry();
+ DatabaseEntry value = new DatabaseEntry();
+ TupleBinding binding = _bindingTupleBindingFactory.getInstance();
+
+ while (cursor.getNext(key, value, LockMode.RMW) == OperationStatus.SUCCESS)
+ {
+ //yes, this is retrieving all the useful information from the key only.
+ //For table compatibility it shall currently be left as is
+ BindingKey bindingRecord = (BindingKey) binding.entryToObject(key);
+
+ String exchangeName = bindingRecord.getExchangeName() == null ? null :
+ bindingRecord.getExchangeName().asString();
+ String queueName = bindingRecord.getQueueName() == null ? null :
+ bindingRecord.getQueueName().asString();
+ String routingKey = bindingRecord.getRoutingKey() == null ? null :
+ bindingRecord.getRoutingKey().asString();
+ ByteBuffer argumentsBB = (bindingRecord.getArguments() == null ? null :
+ java.nio.ByteBuffer.wrap(bindingRecord.getArguments().getDataAsBytes()));
+
+ brh.binding(exchangeName, queueName, routingKey, argumentsBB);
+ }
+ }
+ finally
+ {
+ if (cursor != null)
+ {
+ cursor.close();
+ }
+ }
+
+ }
+
+ private void recoverMessages(MessageStoreRecoveryHandler msrh) throws DatabaseException
+ {
+ StoredMessageRecoveryHandler mrh = msrh.begin();
+
+ Cursor cursor = null;
+ try
+ {
+ cursor = _messageMetaDataDb.openCursor(null, null);
+ DatabaseEntry key = new DatabaseEntry();
+ EntryBinding keyBinding = TupleBinding.getPrimitiveBinding(Long.class);;
+
+ DatabaseEntry value = new DatabaseEntry();
+ EntryBinding valueBinding = _metaDataTupleBindingFactory.getInstance();
+
+ long maxId = 0;
+
+ while (cursor.getNext(key, value, LockMode.RMW) == OperationStatus.SUCCESS)
+ {
+ long messageId = (Long) keyBinding.entryToObject(key);
+ StorableMessageMetaData metaData = (StorableMessageMetaData) valueBinding.entryToObject(value);
+
+ StoredBDBMessage message = new StoredBDBMessage(messageId, metaData, false);
+ mrh.message(message);
+
+ maxId = Math.max(maxId, messageId);
+ }
+
+ _messageId.set(maxId);
+ }
+ catch (DatabaseException e)
+ {
+ _log.error("Database Error: " + e, e);
+ throw e;
+ }
+ finally
+ {
+ if (cursor != null)
+ {
+ cursor.close();
+ }
+ }
+ }
+
+ private void recoverQueueEntries(TransactionLogRecoveryHandler recoveryHandler)
+ throws DatabaseException, AMQException
+ {
+ QueueEntryRecoveryHandler qerh = recoveryHandler.begin(this);
+
+ Cursor cursor = null;
+ try
+ {
+ cursor = _deliveryDb.openCursor(null, null);
+ DatabaseEntry key = new DatabaseEntry();
+ EntryBinding keyBinding = new QueueEntryTB();
+
+ DatabaseEntry value = new DatabaseEntry();
+ EntryBinding valueBinding = TupleBinding.getPrimitiveBinding(Long.class);
+
+ while (cursor.getNext(key, value, LockMode.RMW) == OperationStatus.SUCCESS)
+ {
+
+ QueueEntryKey dd = (QueueEntryKey) keyBinding.entryToObject(key);
+
+ AMQShortString queueName = dd.getQueueName();
+ long messageId = dd.getMessageId();
+
+ qerh.queueEntry(queueName.asString(),messageId);
+
+// if (_log.isDebugEnabled())
+// {
+// _log.debug("On recovery, delivering Message ID:" + message.getMessageId() + " to " + queue.getName());
+// }
+
+ }
+ }
+ catch (DatabaseException e)
+ {
+ _log.error("Database Error: " + e, e);
+ throw e;
+ }
+ finally
+ {
+ if (cursor != null)
+ {
+ cursor.close();
+ }
+ }
+
+// if (_log.isInfoEnabled())
+// {
+// _log.info("Recovered message counts: " + _queueRecoveries);
+// }
+//
+// for(Map.Entry<AMQShortString,Integer> entry : _queueRecoveries.entrySet())
+// {
+// CurrentActor.get().message(_logSubject, MessageStoreMessages.MST_RECOVERED(entry.getValue(), String.valueOf(entry.getKey())));
+//
+// CurrentActor.get().message(_logSubject, MessageStoreMessages.MST_RECOVERY_COMPLETE(String.valueOf(entry.getKey()), true));
+// }
+ }
+
/**
- * Removes the specified message from the store in the given transactional store context.
+ * Removes the specified message from the store.
*
- * @param context The transactional context to remove the message in.
* @param messageId Identifies the message to remove.
*
* @throws AMQException If the operation fails for any reason.
+ * @throws DatabaseException
*/
- public void removeMessage(StoreContext context, Long messageId) throws AMQException
+ public void removeMessage(Long messageId) throws AMQException
{
// _log.debug("public void removeMessage(StoreContext context = " + context + ", Long messageId = " + messageId
// + "): called");
- boolean localTx = getOrCreateTransaction(context);
- Transaction tx = (Transaction) context.getPayload();
+ com.sleepycat.je.Transaction tx = null;
+
+ Cursor cursor = null;
+ try
+ {
+ tx = _environment.beginTransaction(null, null);
+
+ //remove the message meta data from the store
+ DatabaseEntry key = new DatabaseEntry();
+ EntryBinding metaKeyBindingTuple = TupleBinding.getPrimitiveBinding(Long.class);
+ metaKeyBindingTuple.objectToEntry(messageId, key);
- DatabaseEntry key = new DatabaseEntry();
- EntryBinding keyBinding = TupleBinding.getPrimitiveBinding(Long.class);
- keyBinding.objectToEntry(messageId, key);
+ if (_log.isDebugEnabled())
+ {
+ _log.debug("Message Id: " + messageId + " Removing");
+ }
- if (_log.isDebugEnabled())
- {
- _log.debug("Message Id: " + messageId + " Removing");
- }
-
- // first we need to look up the header to get the chunk count
- MessageMetaData mmd = getMessageMetaData(context, messageId);
- try
- {
+
OperationStatus status = _messageMetaDataDb.delete(tx, key);
if (status == OperationStatus.NOTFOUND)
{
- if (localTx)
- {
- tx.abort();
- context.setPayload(null);
- }
+ tx.abort();
throw new AMQException("Message metadata not found for message id " + messageId);
}
@@ -529,56 +790,73 @@
_log.debug("Deleted metadata for message " + messageId);
}
- DatabaseEntry contentKey = new DatabaseEntry();
- TupleBinding contentKeyBinding = new MessageContentKey.TupleBinding();
- for (int i = 0; i < mmd.getContentChunkCount(); i++)
+ //now remove the content data from the store if there is any.
+
+ DatabaseEntry contentKeyEntry = new DatabaseEntry();
+ MessageContentKey_3 mck = new MessageContentKey_3(messageId,0);
+
+ TupleBinding contentKeyTupleBinding = new MessageContentKeyTB_3();
+ contentKeyTupleBinding.objectToEntry(mck, contentKeyEntry);
+
+ //Use a partial record for the value to prevent retrieving the
+ //data itself as we only need the key to identify what to remove.
+ DatabaseEntry value = new DatabaseEntry();
+ value.setPartial(0, 0, true);
+
+ cursor = _messageContentDb.openCursor(tx, null);
+
+ status = cursor.getSearchKeyRange(contentKeyEntry, value, LockMode.RMW);
+ while (status == OperationStatus.SUCCESS)
{
- MessageContentKey mck = new MessageContentKey(messageId, i);
- contentKeyBinding.objectToEntry(mck, contentKey);
- status = _messageContentDb.get(tx, contentKey, new DatabaseEntry(), LockMode.RMW);
- if (status == OperationStatus.NOTFOUND)
+ mck = (MessageContentKey_3) contentKeyTupleBinding.entryToObject(contentKeyEntry);
+
+ if(mck.getMessageId() != messageId)
{
- if (localTx)
- {
- tx.abort();
- context.setPayload(null);
- }
-
- throw new AMQException("Content chunk " + i + " not found for message " + messageId);
+ //we have exhausted all chunks for this message id, break
+ break;
}
-
- status = _messageContentDb.delete(tx, contentKey);
- if (status == OperationStatus.NOTFOUND)
+ else
{
- if (localTx)
+ status = cursor.delete();
+
+ if(status == OperationStatus.NOTFOUND)
{
+ cursor.close();
+ cursor = null;
+
tx.abort();
- context.setPayload(null);
+ throw new AMQException("Content chunk offset" + mck.getOffset() + " not found for message " + messageId);
}
-
- throw new AMQException("Content chunk " + i + " not found for message " + messageId);
+
+ if (_log.isDebugEnabled())
+ {
+ _log.debug("Deleted content chunk offset " + mck.getOffset() + " for message " + messageId);
+ }
}
-
- if (_log.isDebugEnabled())
- {
- _log.debug("Deleted content chunk " + i + " for message " + messageId);
- }
+
+ status = cursor.getNext(contentKeyEntry, value, LockMode.RMW);
}
- if (localTx)
- {
- commit(tx);
- context.setPayload(null);
- }
+ cursor.close();
+ cursor = null;
+
+ commit(tx);
}
catch (DatabaseException e)
{
- if ((tx != null) && localTx)
+ e.printStackTrace();
+
+ if (tx != null)
{
try
{
+ if(cursor != null)
+ {
+ cursor.close();
+ cursor = null;
+ }
+
tx.abort();
- context.setPayload(null);
}
catch (DatabaseException e1)
{
@@ -586,8 +864,23 @@
}
}
- throw new AMQException("Error writing AMQMessage with id " + messageId + " to database: " + e, e);
+ throw new AMQException("Error removing message with id " + messageId + " from database: " + e, e);
}
+ finally
+ {
+ if(cursor != null)
+ {
+ try
+ {
+ cursor.close();
+ }
+ catch (DatabaseException e)
+ {
+ //TODO
+ throw new RuntimeException(e);
+ }
+ }
+ }
}
/**
@@ -601,12 +894,17 @@
{
if (_state != State.RECOVERING)
{
+ ExchangeRecord exchangeRec = new ExchangeRecord(exchange.getNameShortString(),
+ exchange.getTypeShortString(), exchange.isAutoDelete());
+
DatabaseEntry key = new DatabaseEntry();
EntryBinding keyBinding = new AMQShortStringTB();
- keyBinding.objectToEntry(exchange.getName(), key);
+ keyBinding.objectToEntry(exchange.getNameShortString(), key);
+
DatabaseEntry value = new DatabaseEntry();
- TupleBinding exchangeBinding = new ExchangeTB(_virtualHost);
- exchangeBinding.objectToEntry(exchange, value);
+ TupleBinding exchangeBinding = new ExchangeTB();
+ exchangeBinding.objectToEntry(exchangeRec, value);
+
try
{
_exchangeDb.put(null, key, value);
@@ -630,7 +928,7 @@
{
DatabaseEntry key = new DatabaseEntry();
EntryBinding keyBinding = new AMQShortStringTB();
- keyBinding.objectToEntry(exchange.getName(), key);
+ keyBinding.objectToEntry(exchange.getNameShortString(), key);
try
{
OperationStatus status = _exchangeDb.delete(null, key);
@@ -645,117 +943,9 @@
}
}
- private void recoverExchanges() throws AMQException, DatabaseException
- {
- for (Exchange exchange : loadExchanges())
- {
- recoverExchange(exchange);
- }
- }
- private void recoverExchange(Exchange exchange) throws AMQException, DatabaseException
- {
- _log.info("Recovering durable exchange " + exchange.getName() + " of type " + exchange.getType() + "...");
- QueueRegistry queueRegistry = _virtualHost.getQueueRegistry();
- for (BindingKey binding : loadQueueBindings(exchange))
- {
- AMQQueue queue = queueRegistry.getQueue(binding.getQueueName());
- if (queue == null)
- {
- _log.error("Unknown queue: " + binding.getQueueName() + " cannot be bound to exchange: "
- + exchange.getName());
- }
- else
- {
- _log.info("Restoring binding: (Exchange: " + binding.getExchangeName() + ", Queue: " + binding
- .getQueueName() + ", Routing Key: " + binding.getRoutingKey() + ", Arguments: " + binding.getArguments()
- + ")");
-
- queue.bind(exchange, binding.getRoutingKey(), binding.getArguments());
- }
- }
- }
-
- private List<BindingKey> loadQueueBindings(Exchange exchange) throws DatabaseException
- {
-
- Cursor cursor = null;
- List<BindingKey> queueBindings = new ArrayList<BindingKey>();
- try
- {
- cursor = _queueBindingsDb.openCursor(null, null);
- DatabaseEntry key = new DatabaseEntry();
- DatabaseEntry value = new DatabaseEntry();
-
- BindingKey queueBinding =
- new BindingKey(exchange.getName(), null, null, null);
-
- EntryBinding<BindingKey> keyBinding = _bindingTupleBindingFactory.getInstance();
- keyBinding.objectToEntry(queueBinding, key);
-
- OperationStatus opStatus = cursor.getSearchKeyRange(key, value, LockMode.RMW);
-
- TupleBinding binding = _bindingTupleBindingFactory.getInstance();
- while (opStatus == OperationStatus.SUCCESS)
- {
- queueBinding = (BindingKey) binding.entryToObject(key);
- if (queueBinding.getExchangeName().equals(exchange.getName()))
- {
- queueBindings.add(queueBinding);
- opStatus = cursor.getNext(key, value, LockMode.RMW);
- }
- else
- {
- break;
- }
- }
-
- return queueBindings;
- }
- finally
- {
- if (cursor != null)
- {
- cursor.close();
- }
- }
-
- }
-
- private List<Exchange> loadExchanges() throws AMQException, DatabaseException
- {
- Cursor cursor = null;
- List<Exchange> exchanges = new ArrayList<Exchange>();
- try
- {
- cursor = _exchangeDb.openCursor(null, null);
- DatabaseEntry key = new DatabaseEntry();
- DatabaseEntry value = new DatabaseEntry();
- TupleBinding binding = new ExchangeTB(_virtualHost);
- while (cursor.getNext(key, value, LockMode.RMW) == OperationStatus.SUCCESS)
- {
- Exchange exchange = (Exchange) binding.entryToObject(value);
-
- _virtualHost.getExchangeRegistry().registerExchange(exchange);
-
- exchanges.add(exchange);
- _log.info("Registering exchange " + exchange.getName());
- }
-
- return exchanges;
- }
- finally
- {
- if (cursor != null)
- {
- cursor.close();
- }
- }
-
- }
-
/**
* Binds the specified queue to an exchange with a routing key.
*
@@ -773,13 +963,20 @@
if (_state != State.RECOVERING)
{
+ BindingKey bindingRecord = new BindingKey(exchange.getNameShortString(),
+ queue.getNameShortString(), routingKey, args);
+
DatabaseEntry key = new DatabaseEntry();
EntryBinding keyBinding = _bindingTupleBindingFactory.getInstance();
- keyBinding.objectToEntry(new BindingKey(exchange.getName(), queue.getName(), routingKey, args), key);
+
+ keyBinding.objectToEntry(bindingRecord, key);
+ //yes, this is writing out 0 as a value and putting all the
+ //useful info into the key, don't ask me why. For table
+ //compatibility it shall currently be left as is
DatabaseEntry value = new DatabaseEntry();
ByteBinding.byteToEntry((byte) 0, value);
-
+
try
{
_queueBindingsDb.put(null, key, value);
@@ -807,7 +1004,7 @@
{
DatabaseEntry key = new DatabaseEntry();
EntryBinding keyBinding = _bindingTupleBindingFactory.getInstance();
- keyBinding.objectToEntry(new BindingKey(exchange.getName(), queue.getName(), routingKey, args), key);
+ keyBinding.objectToEntry(new BindingKey(exchange.getNameShortString(), queue.getNameShortString(), routingKey, args), key);
try
{
@@ -844,20 +1041,17 @@
if (_state != State.RECOVERING)
{
- long queueId = _queueId.getAndIncrement();
- _queueNameToIdMap.put(queue.getName(), queueId);
-
+ QueueRecord queueRecord= new QueueRecord(queue.getNameShortString(),
+ queue.getOwner(), arguments);
+
DatabaseEntry key = new DatabaseEntry();
-
EntryBinding keyBinding = new AMQShortStringTB();
- keyBinding.objectToEntry(queue.getName(), key);
+ keyBinding.objectToEntry(queue.getNameShortString(), key);
DatabaseEntry value = new DatabaseEntry();
TupleBinding queueBinding = _queueTupleBindingFactory.getInstance();
- ((QueueTuple) queueBinding).setArguments(arguments);
-
- queueBinding.objectToEntry(queue, value);
+ queueBinding.objectToEntry(queueRecord, value);
try
{
_queueDb.put(null, key, value);
@@ -878,12 +1072,10 @@
*/
public void removeQueue(final AMQQueue queue) throws AMQException
{
- AMQShortString name = queue.getName();
+ AMQShortString name = queue.getNameShortString();
_log.debug("public void removeQueue(AMQShortString name = " + name + "): called");
- Long queueId = _queueNameToIdMap.remove(name);
-
DatabaseEntry key = new DatabaseEntry();
EntryBinding keyBinding = new AMQShortStringTB();
keyBinding.objectToEntry(name, key);
@@ -902,34 +1094,6 @@
}
/**
- * Really for testing purposes.
- *
- * @param name
- *
- * @return
- *
- * @throws AMQException
- */
- AMQQueue getQueue(AMQShortString name) throws AMQException
- {
- DatabaseEntry key = new DatabaseEntry();
- EntryBinding keyBinding = new AMQShortStringTB();
- keyBinding.objectToEntry(name, key);
- DatabaseEntry value = new DatabaseEntry();
- try
- {
- _queueDb.get(null, key, value, LockMode.RMW);
- TupleBinding binding = _queueTupleBindingFactory.getInstance();
-
- return (AMQQueue) binding.entryToObject(value);
- }
- catch (DatabaseException e)
- {
- throw new AMQException("Error getting queue with name " + name + " from database: " + e, e);
- }
- }
-
- /**
* Places a message onto a specified queue, in a given transactional context.
*
* @param context The transactional context for the operation.
@@ -938,15 +1102,16 @@
*
* @throws AMQException If the operation fails for any reason.
*/
- public void enqueueMessage(StoreContext context, AMQQueue queue, Long messageId) throws AMQException
+ public void enqueueMessage(StoreContext context, final TransactionLogResource queue, Long messageId) throws AMQException
{
// _log.debug("public void enqueueMessage(StoreContext context = " + context + ", AMQShortString name = " + name
// + ", Long messageId): called");
- AMQShortString name = queue.getName();
- Transaction tx = (Transaction) context.getPayload();
+ AMQShortString name = new AMQShortString(queue.getResourceName());
+ com.sleepycat.je.Transaction tx = (com.sleepycat.je.Transaction) context.getPayload();
+
DatabaseEntry key = new DatabaseEntry();
- EntryBinding keyBinding = new QueueEntryKey.TupleBinding();
+ EntryBinding keyBinding = new QueueEntryTB();
QueueEntryKey dd = new QueueEntryKey(name, messageId);
keyBinding.objectToEntry(dd, key);
DatabaseEntry value = new DatabaseEntry();
@@ -968,28 +1133,6 @@
}
}
- private boolean getOrCreateTransaction(StoreContext context) throws AMQException
- {
-
- Transaction tx = (Transaction) context.getPayload();
- if (tx == null)
- {
- try
- {
- tx = _environment.beginTransaction(null, null);
- context.setPayload(tx);
-
- return true;
- }
- catch (DatabaseException e)
- {
- throw new AMQException("Error beginning transaction: " + e, e);
- }
- }
-
- return false;
- }
-
/**
* Extracts a message from a specified queue, in a given transactional context.
*
@@ -999,14 +1142,14 @@
*
* @throws AMQException If the operation fails for any reason, or if the specified message does not exist.
*/
- public void dequeueMessage(StoreContext context, final AMQQueue queue, Long messageId) throws AMQException
+ public void dequeueMessage(StoreContext context, final TransactionLogResource queue, Long messageId) throws AMQException
{
- AMQShortString name = queue.getName();
- boolean isLocal = getOrCreateTransaction(context);
- Transaction tx = (Transaction) context.getPayload();
+ AMQShortString name = new AMQShortString(queue.getResourceName());
+ com.sleepycat.je.Transaction tx = (com.sleepycat.je.Transaction) context.getPayload();
+
DatabaseEntry key = new DatabaseEntry();
- EntryBinding keyBinding = new QueueEntryKey.TupleBinding();
+ EntryBinding keyBinding = new QueueEntryTB();
QueueEntryKey dd = new QueueEntryKey(name, messageId);
keyBinding.objectToEntry(dd, key);
@@ -1034,82 +1177,18 @@
_log.debug("Removed message " + messageId + ", " + name + " from delivery db");
}
-
- if (isLocal)
- {
- commit(tx);
- context.setPayload(null);
- }
}
catch (DatabaseException e)
{
_log.error("Failed to dequeue message " + messageId + ": " + e, e);
_log.error(tx);
- if (isLocal)
- {
- try
- {
- tx.abort();
- context.setPayload(null);
- }
- catch (DatabaseException e1)
- {
- throw new AMQException("Error rolling back transaction: " + e1, e1);
- }
- }
throw new AMQException("Error accessing database while dequeuing message: " + e, e);
}
}
- private boolean isLocalTransaction(StoreContext context)
- {
- return context.getPayload() == null;
- }
-
/**
- * Begins a transactional context.
- *
- * @param context The transactional context to begin.
- *
- * @throws AMQException If the operation fails for any reason.
- */
- public void beginTran(StoreContext context) throws AMQException
- {
- // _log.debug("public void beginTran(StoreContext context = " + context + "): called");
-
- if (context.getPayload() != null)
- {
- throw new AMQException("Fatal internal error: transactional context is not empty at beginTran: "
- + context.getPayload());
- }
- else
- {
- try
- {
- context.setPayload(_environment.beginTransaction(null, _transactionConfig ));
- }
- catch (DatabaseException e)
- {
- throw new AMQException("Error starting transaction: " + e, e);
- }
- }
- }
-
- /**
- * Tests a transactional context to see if it has been begun but not yet committed or aborted.
- *
- * @param context The transactional context to test.
- *
- * @return <tt>true</tt> if the transactional context is live, <tt>false</tt> otherwise.
- */
- public boolean inTran(StoreContext context)
- {
- return context.getPayload() != null;
- }
-
- /**
* Commits all operations performed within a given transactional context.
*
* @param context The transactional context to commit all operations for.
@@ -1120,7 +1199,7 @@
{
// _log.debug("public void commitTran(StoreContext context = " + context + "): called");
- Transaction tx = (Transaction) context.getPayload();
+ com.sleepycat.je.Transaction tx = (com.sleepycat.je.Transaction) context.getPayload();
if (tx == null)
{
@@ -1146,6 +1225,13 @@
}
}
+ public StoreFuture commitTranAsync(StoreContext context) throws AMQException
+ {
+ //TODO - Actually create an async commit implementation.
+ commitTran(context);
+ return IMMEDIATE_FUTURE;
+ }
+
/**
* Abandons all operations performed within a given transactional context.
*
@@ -1155,7 +1241,7 @@
*/
public void abortTran(StoreContext context) throws AMQException
{
- Transaction tx = (Transaction) context.getPayload();
+ com.sleepycat.je.Transaction tx = (com.sleepycat.je.Transaction) context.getPayload();
if (_log.isDebugEnabled())
{
@@ -1194,7 +1280,7 @@
QueueEntryKey dd = new QueueEntryKey(queueName, 0);
- EntryBinding keyBinding = new QueueEntryKey.TupleBinding();
+ EntryBinding keyBinding = new QueueEntryTB();
keyBinding.objectToEntry(dd, key);
DatabaseEntry value = new DatabaseEntry();
@@ -1204,10 +1290,10 @@
OperationStatus status = cursor.getSearchKeyRange(key, value, LockMode.DEFAULT);
dd = (QueueEntryKey) keyBinding.entryToObject(key);
- while ((status == OperationStatus.SUCCESS) && dd.queueName.equals(queueName))
+ while ((status == OperationStatus.SUCCESS) && dd.getQueueName().equals(queueName))
{
- messageIds.add(dd.messageId);
+ messageIds.add(dd.getMessageId());
status = cursor.getNext(key, value, LockMode.DEFAULT);
if (status == OperationStatus.SUCCESS)
{
@@ -1237,59 +1323,6 @@
}
}
- public void recover(VirtualHost virtualHost) throws AMQException
- {
- stateTransition(State.CONFIGURED, State.RECOVERING);
-
- _log.info("Recovering persistent state...");
- CurrentActor.get().message(_logSubject,MessageStoreMessages.MST_RECOVERY_START(null, false));
-
- StoreContext context = new StoreContext();
- try
- {
- beginTran(context);
- Map<AMQShortString, AMQQueue> queues = loadQueues();
-
- for (AMQQueue q : queues.values())
- {
- CurrentActor.get().message(_logSubject,MessageStoreMessages.MST_RECOVERY_START(String.valueOf(q.getName()), true));
- //Record that we have a queue for recovery
- _queueRecoveries.put(q.getName(), 0);
-
- q.configure(virtualHost.getConfiguration().getQueueConfiguration(q.getName().asString()));
-
- }
-
- recoverExchanges();
-
- deliverMessages(context, queues);
- _log.info("Persistent state recovered successfully");
- CurrentActor.get().message(_logSubject, MessageStoreMessages.MST_RECOVERY_COMPLETE(null, false));
-
- commitTran(context);
-
- }
- catch (DatabaseException e)
- {
- abortTran(context);
-
- throw new AMQException("Error recovering persistent state: " + e, e);
- }
- catch (AMQException amqe)
- {
- abortTran(context);
- throw new AMQException("Error recovering persistent state: " + amqe, amqe);
- }
- catch (Throwable ioobe)
- {
- abortTran(context);
- throw new AMQException("Invalid database format. Please use upgrade tool for store in Virtualhost:'"
- + _virtualHost.getName() + "'", ioobe);
- }
-
- stateTransition(State.RECOVERING, State.STARTED);
- }
-
/**
* Return a valid, currently unused message id.
*
@@ -1297,7 +1330,7 @@
*/
public Long getNewMessageId()
{
- return _messageId.getAndIncrement();
+ return _messageId.incrementAndGet();
}
/**
@@ -1305,20 +1338,21 @@
*
* @param context The transactional context for the operation.
* @param messageId The message to store the data for.
- * @param index The index of the data chunk.
+ * @param offset The offset of the data chunk in the message.
* @param contentBody The content of the data chunk.
* @param lastContentBody Flag to indicate that this is the last such chunk for the message.
*
* @throws AMQException If the operation fails for any reason, or if the specified message does not exist.
*/
- public void storeContentBodyChunk(StoreContext context, Long messageId, int index, ContentChunk contentBody,
- boolean lastContentBody) throws AMQException
+ protected void addContent(StoreContext context, Long messageId, int offset,
+ ByteBuffer contentBody) throws AMQException
{
- Transaction tx = (Transaction) context.getPayload();
+ com.sleepycat.je.Transaction tx = (com.sleepycat.je.Transaction) context.getPayload();
+
DatabaseEntry key = new DatabaseEntry();
- TupleBinding keyBinding = new MessageContentKey.TupleBinding();
- keyBinding.objectToEntry(new MessageContentKey(messageId, index), key);
+ TupleBinding keyBinding = new MessageContentKeyTB_3();
+ keyBinding.objectToEntry(new MessageContentKey_3(messageId, offset), key);
DatabaseEntry value = new DatabaseEntry();
TupleBinding messageBinding = new ContentTB();
messageBinding.objectToEntry(contentBody, value);
@@ -1327,13 +1361,13 @@
OperationStatus status = _messageContentDb.put(tx, key, value);
if (status != OperationStatus.SUCCESS)
{
- throw new AMQException("Error adding content chunk " + index + " for message id " + messageId + ": "
+ throw new AMQException("Error adding content chunk offset" + offset + " for message id " + messageId + ": "
+ status);
}
if (_log.isDebugEnabled())
{
- _log.debug("Storing content chunk " + index + " for message " + messageId + "[Transaction" + tx + "]");
+ _log.debug("Storing content chunk offset" + offset + " for message " + messageId + "[Transaction" + tx + "]");
}
}
catch (DatabaseException e)
@@ -1351,7 +1385,7 @@
*
* @throws AMQException If the operation fails for any reason, or if the specified message does not exist.
*/
- public void storeMessageMetaData(StoreContext context, Long messageId, MessageMetaData messageMetaData)
+ private void storeMetaData(StoreContext context, Long messageId, StorableMessageMetaData messageMetaData)
throws AMQException
{
if (_log.isDebugEnabled())
@@ -1359,14 +1393,15 @@
_log.debug("public void storeMessageMetaData(StoreContext context = " + context + ", Long messageId = "
+ messageId + ", MessageMetaData messageMetaData = " + messageMetaData + "): called");
}
- //This call breaking tests - not sure where the txn it creates should be committed ??
- //getOrCreateTransaction(context);
- Transaction tx = (Transaction) context.getPayload();
+
+ com.sleepycat.je.Transaction tx = (com.sleepycat.je.Transaction) context.getPayload();
+
DatabaseEntry key = new DatabaseEntry();
EntryBinding keyBinding = TupleBinding.getPrimitiveBinding(Long.class);
keyBinding.objectToEntry(messageId, key);
DatabaseEntry value = new DatabaseEntry();
- TupleBinding messageBinding = new MessageMetaDataTB();
+
+ TupleBinding messageBinding = _metaDataTupleBindingFactory.getInstance();
messageBinding.objectToEntry(messageMetaData, value);
try
{
@@ -1392,11 +1427,11 @@
*
* @throws AMQException If the operation fails for any reason, or if the specified message does not exist.
*/
- public MessageMetaData getMessageMetaData(StoreContext context, Long messageId) throws AMQException
+ public StorableMessageMetaData getMessageMetaData(Long messageId) throws AMQException
{
if (_log.isDebugEnabled())
{
- _log.debug("public MessageMetaData getMessageMetaData(StoreContext context = " + context + ", Long messageId = "
+ _log.debug("public MessageMetaData getMessageMetaData(Long messageId = "
+ messageId + "): called");
}
@@ -1404,7 +1439,7 @@
EntryBinding keyBinding = TupleBinding.getPrimitiveBinding(Long.class);
keyBinding.objectToEntry(messageId, key);
DatabaseEntry value = new DatabaseEntry();
- TupleBinding messageBinding = new MessageMetaDataTB();
+ TupleBinding messageBinding = _metaDataTupleBindingFactory.getInstance();
try
{
@@ -1414,7 +1449,7 @@
throw new AMQException("Metadata not found for message with id " + messageId);
}
- MessageMetaData mdd = (MessageMetaData) messageBinding.entryToObject(value);
+ StorableMessageMetaData mdd = (StorableMessageMetaData) messageBinding.entryToObject(value);
return mdd;
}
@@ -1426,47 +1461,110 @@
}
/**
- * Retrieves a chunk of message data.
+ * Fills the provided ByteBuffer with as much content for the specified message as possible, starting
+ * from the specified offset in the message.
*
- * @param context The transactional context for the operation.
- * @param messageId The message to get the data chunk for.
- * @param index The offset index of the data chunk within the message.
+ * @param messageId The message to get the data for.
+ * @param offset The offset of the data within the message.
+ * @param dst The destination of the content read back
*
- * @return A chunk of message data.
+ * @return The number of bytes inserted into the destination
*
* @throws AMQException If the operation fails for any reason, or if the specified message does not exist.
*/
- public ContentChunk getContentBodyChunk(StoreContext context, Long messageId, int index) throws AMQException
+ public int getContent(Long messageId, int offset, ByteBuffer dst) throws AMQException
{
-
- DatabaseEntry key = new DatabaseEntry();
- TupleBinding keyBinding = new MessageContentKey.TupleBinding();
- keyBinding.objectToEntry(new MessageContentKey(messageId, index), key);
+ final int limit = offset + dst.remaining();
+
+ DatabaseEntry contentKeyEntry = new DatabaseEntry();
+
+ //TODO: if requested offset is non zero, use partial record (key-only) search to
+ //locate the first record key to prevent reading in data we dont need
+
+ //Start from 0 offset and search for the starting chunk.
+ MessageContentKey_3 mck = new MessageContentKey_3(messageId, 0);
+ TupleBinding contentKeyTupleBinding = new MessageContentKeyTB_3();
+ contentKeyTupleBinding.objectToEntry(mck, contentKeyEntry);
DatabaseEntry value = new DatabaseEntry();
- TupleBinding messageBinding = new ContentTB();
+ TupleBinding contentTupleBinding = new ContentTB();
+
if (_log.isDebugEnabled())
{
- _log.debug("Message Id: " + messageId + " Getting content body chunk: " + index);
+ _log.debug("Message Id: " + messageId + " Getting content body from offset: " + offset);
}
+ int written = 0;
+ int seenSoFar = 0;
+
+ Cursor cursor = null;
try
{
- OperationStatus status = _messageContentDb.get(null, key, value, LockMode.READ_UNCOMMITTED);
- if (status != OperationStatus.SUCCESS)
+ cursor = _messageContentDb.openCursor(null, null);
+
+ OperationStatus status = cursor.getSearchKeyRange(contentKeyEntry, value, LockMode.RMW);
+ while (status == OperationStatus.SUCCESS)
{
+ mck = (MessageContentKey_3) contentKeyTupleBinding.entryToObject(contentKeyEntry);
+ long id = mck.getMessageId();
+
+ if(id != messageId)
+ {
+ //we have exhausted all chunks for this message id, break
+ break;
+ }
+
+ int offsetInMessage = mck.getOffset();
+ ByteBuffer buf = (ByteBuffer) contentTupleBinding.entryToObject(value);
+
+ final int size = (int) buf.limit();
+
+ seenSoFar += size;
+
+ //TODO: can change this guard if we start recording the last byte in the chunk record
+ if(seenSoFar >= offset)
+ {
+ byte[] dataAsBytes = buf.array();
- throw new AMQException("Content chunk " + index + " not found for message with id " + messageId);
+ int posInArray = offset + written - offsetInMessage;
+ int count = size - posInArray;
+ if(count > dst.remaining())
+ {
+ count = dst.remaining();
+ }
+ dst.put(dataAsBytes,posInArray,count);
+ written+=count;
+
+ if(dst.remaining() == 0)
+ {
+ break;
+ }
+ }
+
+ status = cursor.getNext(contentKeyEntry, value, LockMode.RMW);
}
- ContentChunk cb = (ContentChunk) messageBinding.entryToObject(value);
-
- return cb;
+ return written;
}
catch (DatabaseException e)
{
throw new AMQException("Error writing AMQMessage with id " + messageId + " to database: " + e, e);
}
+ finally
+ {
+ if(cursor != null)
+ {
+ try
+ {
+ cursor.close();
+ }
+ catch (DatabaseException e)
+ {
+ // TODO
+ throw new RuntimeException(e);
+ }
+ }
+ }
}
public boolean isPersistent()
@@ -1474,47 +1572,27 @@
return true;
}
- Map<AMQShortString, AMQQueue> loadQueues() throws DatabaseException, AMQException
+ public <T extends StorableMessageMetaData> StoredMessage<T> addMessage(T metaData)
{
- Cursor cursor = null;
- Map<AMQShortString, AMQQueue> queues = new HashMap<AMQShortString, AMQQueue>();
- try
+ if(metaData.isPersistent())
{
- cursor = _queueDb.openCursor(null, null);
- DatabaseEntry key = new DatabaseEntry();
- DatabaseEntry value = new DatabaseEntry();
- TupleBinding binding = _queueTupleBindingFactory.getInstance();
- while (cursor.getNext(key, value, LockMode.RMW) == OperationStatus.SUCCESS)
- {
- AMQQueue queue = (AMQQueue) binding.entryToObject(value);
- if (queue != null)
- {
- _virtualHost.getQueueRegistry().registerQueue(queue);
- queues.put(queue.getName(), queue);
- _log.info("Recovering queue " + queue.getName() + " with owner:"
- + ((queue.getOwner() == null) ? " <<null>> " : (" \"" + queue.getOwner() + "\" ")));
- }
- }
-
- return queues;
+ return new StoredBDBMessage(getNewMessageId(), metaData);
}
- finally
+ else
{
- if (cursor != null)
- {
- cursor.close();
- }
+ return new StoredMemoryMessage(getNewMessageId(), metaData);
}
}
- //public getters for the TupleBindingFactories
- public QueueTupleBindingFactory getQueueTupleBindingFactory()
+ //protected getters for the TupleBindingFactories
+
+ protected QueueTupleBindingFactory getQueueTupleBindingFactory()
{
return _queueTupleBindingFactory;
}
- public BindingTupleBindingFactory getBindingTupleBindingFactory()
+ protected BindingTupleBindingFactory getBindingTupleBindingFactory()
{
return _bindingTupleBindingFactory;
}
@@ -1612,162 +1690,8 @@
}
}
- private static final class ProcessAction
+ private void commit(com.sleepycat.je.Transaction tx) throws DatabaseException
{
- private final AMQQueue _queue;
- private final StoreContext _context;
- private final AMQMessage _message;
-
- public ProcessAction(AMQQueue queue, StoreContext context, AMQMessage message)
- {
- _queue = queue;
- _context = context;
- _message = message;
- }
-
- public void process() throws AMQException
- {
- _queue.enqueue(_context, _message);
- }
-
- }
-
- private void deliverMessages(final StoreContext context, Map<AMQShortString, AMQQueue> queues)
- throws DatabaseException, AMQException
- {
- Map<Long, AMQMessage> msgMap = new HashMap<Long, AMQMessage>();
- List<ProcessAction> actions = new ArrayList<ProcessAction>();
-
-
- Cursor cursor = null;
- try
- {
- Transaction tx = (Transaction) context.getPayload();
- cursor = _deliveryDb.openCursor(tx, null);
- DatabaseEntry key = new DatabaseEntry();
- EntryBinding keyBinding = new QueueEntryKey.TupleBinding();
-
- DatabaseEntry value = new DatabaseEntry();
- EntryBinding valueBinding = TupleBinding.getPrimitiveBinding(Long.class);
- MessageHandleFactory messageHandleFactory = new MessageHandleFactory();
- long maxId = 1;
-
- TransactionalContext txnContext = new NonTransactionalContext(this, new StoreContext(), null, null);
-
- while (cursor.getNext(key, value, LockMode.RMW) == OperationStatus.SUCCESS)
- {
-
- QueueEntryKey dd = (QueueEntryKey) keyBinding.entryToObject(key);
-
- AMQShortString queueName = dd.queueName;
-
- AMQQueue queue = queues.get(queueName);
-
- // If the matching queue was not already found in the store, check in case a queue
- // with the same name exists in the virtualhost, otherwise we will create a duplicate
- // queue and generate a JMX InstanceAlreadyExistsException, preventing startup.
- if (queue == null)
- {
- queue = _virtualHost.getQueueRegistry().getQueue(queueName);
- if (queue != null)
- {
- queues.put(queueName, queue);
- }
- }
-
- if (queue == null)
- {
- queue = AMQQueueFactory.createAMQQueueImpl(queueName, false, null, false, _virtualHost, null);
- _virtualHost.getQueueRegistry().registerQueue(queue);
- queues.put(queueName, queue);
- }
-
- long messageId = dd.messageId;
- maxId = Math.max(maxId, messageId);
- AMQMessage message = msgMap.get(messageId);
-
- if (message != null)
- {
- message.incrementReference();
- }
- else
- {
- message = new AMQMessage(messageId, this, messageHandleFactory, txnContext);
- msgMap.put(messageId, message);
- }
-
- if (_log.isDebugEnabled())
- {
- _log.debug("On recovery, delivering Message ID:" + message.getMessageId() + " to " + queue.getName());
- }
-
- Integer count = _queueRecoveries.get(queueName);
- if (count == null)
- {
- count = 0;
- }
-
- _queueRecoveries.put(queueName, ++count);
-
- actions.add(new ProcessAction(queue, context, message));
-
- }
-
- for (ProcessAction action : actions)
- {
- action.process();
- }
-
- _messageId.set(maxId + 1);
- }
- catch (DatabaseException e)
- {
- _log.error("Database Error: " + e, e);
- throw e;
- }
- catch (AMQException e)
- {
- _log.error("Store Error: " + e, e);
- throw e;
- }
- finally
- {
- if (cursor != null)
- {
- cursor.close();
- }
- }
-
- if (_log.isInfoEnabled())
- {
- _log.info("Recovered message counts: " + _queueRecoveries);
- }
-
- for(Map.Entry<AMQShortString,Integer> entry : _queueRecoveries.entrySet())
- {
- CurrentActor.get().message(_logSubject, MessageStoreMessages.MST_RECOVERED(entry.getValue(), String.valueOf(entry.getKey())));
-
- CurrentActor.get().message(_logSubject, MessageStoreMessages.MST_RECOVERY_COMPLETE(String.valueOf(entry.getKey()), true));
- }
-
- //Free the memory
- _queueRecoveries = null;
- }
-
- QueueRegistry getQueueRegistry()
- {
- return _virtualHost.getQueueRegistry();
- }
-
- void setVirtualHost(VirtualHost virtualHost)
- {
- _virtualHost = virtualHost;
-
- createTupleBindingFactories(_version);
- }
-
- void commit(Transaction tx) throws DatabaseException
- {
// _log.debug("void commit(Transaction tx = " + tx + "): called");
tx.commitNoSync();
@@ -1787,11 +1711,11 @@
// private static final Logger _log = Logger.getLogger(Commit.class);
private final CommitThread _commitThread;
- private final Transaction _tx;
+ private final com.sleepycat.je.Transaction _tx;
private DatabaseException _databaseException;
private boolean _complete;
- public Commit(CommitThread commitThread, Transaction tx)
+ public Commit(CommitThread commitThread, com.sleepycat.je.Transaction tx)
{
// _log.debug("public Commit(CommitThread commitThread = " + commitThread + ", Transaction tx = " + tx
// + "): called");
@@ -1973,5 +1897,184 @@
}
}
}
+
+
+ private class StoredBDBMessage implements StoredMessage
+ {
+ private final long _messageId;
+ private volatile SoftReference<StorableMessageMetaData> _metaDataRef;
+ private StoreContext _ctx;
+ private com.sleepycat.je.Transaction _txn;
+
+ StoredBDBMessage(long messageId, StorableMessageMetaData metaData)
+ {
+ this(messageId, metaData, true);
+ }
+
+
+ StoredBDBMessage(long messageId,
+ StorableMessageMetaData metaData, boolean persist)
+ {
+ try
+ {
+ _messageId = messageId;
+
+ _metaDataRef = new SoftReference<StorableMessageMetaData>(metaData);
+ if(persist)
+ {
+ _ctx = new StoreContext();
+ _txn = _environment.beginTransaction(null, null);
+ _ctx.setPayload(_txn);
+ storeMetaData(_ctx, messageId, metaData);
+ }
+ }
+ catch (DatabaseException e)
+ {
+ //TODO
+ throw new RuntimeException(e);
+ }
+ catch (AMQException e)
+ {
+ //TODO
+ throw new RuntimeException(e);
+ }
+
+ }
+
+ public StorableMessageMetaData getMetaData()
+ {
+ StorableMessageMetaData metaData = _metaDataRef.get();
+ if(metaData == null)
+ {
+ try
+ {
+ metaData = BDBMessageStore.this.getMessageMetaData(_messageId);
+ }
+ catch (AMQException e)
+ {
+ //TODO
+ throw new RuntimeException(e);
+ }
+ _metaDataRef = new SoftReference<StorableMessageMetaData>(metaData);
+ }
+
+ return metaData;
+ }
+
+ public long getMessageNumber()
+ {
+ return _messageId;
+ }
+
+ public void addContent(int offsetInMessage, java.nio.ByteBuffer src)
+ {
+ try
+ {
+ BDBMessageStore.this.addContent(_ctx, _messageId, offsetInMessage, src);
+ }
+ catch (AMQException e)
+ {
+ //TODO
+ throw new RuntimeException(e);
+ }
+ }
+
+ public int getContent(int offsetInMessage, java.nio.ByteBuffer dst)
+ {
+ try
+ {
+ return BDBMessageStore.this.getContent(_messageId, offsetInMessage, dst);
+ }
+ catch (AMQException e)
+ {
+ // TODO
+ throw new RuntimeException(e);
+ }
+ }
+
+ public StoreFuture flushToStore()
+ {
+ try
+ {
+ if(_ctx != null)
+ {
+ BDBMessageStore.this.commitTran(_ctx);
+ }
+ }
+ catch (AMQException e)
+ {
+ //TODO
+ throw new RuntimeException(e);
+ }
+ finally
+ {
+ _txn = null;
+ _ctx = null;
+ }
+ return IMMEDIATE_FUTURE;
+ }
+
+ public void remove()
+ {
+ flushToStore();
+ try
+ {
+ BDBMessageStore.this.removeMessage(_messageId);
+ }
+ catch (AMQException e)
+ {
+ // TODO
+ throw new RuntimeException(e);
+ }
+ }
+ }
+
+ private class BDBTransaction implements Transaction
+ {
+ private com.sleepycat.je.Transaction _txn;
+ private StoreContext _ctx;
+
+ private BDBTransaction()
+ {
+ _ctx = new StoreContext();
+ try
+ {
+ _txn = _environment.beginTransaction(null, null);
+ }
+ catch (DatabaseException e)
+ {
+ //TODO
+ throw new RuntimeException(e);
+ }
+ _ctx.setPayload(_txn);
+ }
+
+ public void enqueueMessage(TransactionLogResource queue, Long messageId) throws AMQException
+ {
+ BDBMessageStore.this.enqueueMessage(_ctx, queue, messageId);
+ }
+
+ public void dequeueMessage(TransactionLogResource queue, Long messageId) throws AMQException
+ {
+ BDBMessageStore.this.dequeueMessage(_ctx, queue, messageId);
+
+ }
+
+ public void commitTran() throws AMQException
+ {
+ BDBMessageStore.this.commitTran(_ctx);
+ }
+
+ public StoreFuture commitTranAsync() throws AMQException
+ {
+ return BDBMessageStore.this.commitTranAsync(_ctx);
+ }
+
+ public void abortTran() throws AMQException
+ {
+ BDBMessageStore.this.abortTran(_ctx);
+ }
+ }
+
}
Modified: store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/ContentTB.java
===================================================================
--- store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/ContentTB.java 2010-05-13 18:28:33 UTC (rev 3970)
+++ store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/ContentTB.java 2010-05-14 16:23:59 UTC (rev 3971)
@@ -20,14 +20,12 @@
*/
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;
-import org.apache.qpid.framing.ContentBody;
-import org.apache.qpid.framing.abstraction.ContentChunk;
-import org.apache.mina.common.ByteBuffer;
-
/**
* @author Robert Greig (robert.j.greig at jpmorgan.com)
*/
@@ -39,39 +37,19 @@
final int size = tupleInput.readInt();
byte[] underlying = new byte[size];
tupleInput.readFast(underlying);
- final ByteBuffer data = ByteBuffer.wrap(underlying);
- ContentChunk cb = new ContentChunk()
- {
-
- public int getSize()
- {
- return size;
- }
-
- public ByteBuffer getData()
- {
- return data;
- }
-
- public void reduceToFit()
- {
-
- }
- };
- return cb;
+ return ByteBuffer.wrap(underlying);
}
public void objectToEntry(Object object, TupleOutput tupleOutput)
{
- ContentChunk cb = (ContentChunk) object;
- final int size = cb.getSize();
- byte[] underlying = new byte[size];
+ ByteBuffer src = (ByteBuffer) object;
+
+ src = src.slice();
- ByteBuffer buf = cb.getData();
+ byte[] chunkData = new byte[src.limit()];
+ src.duplicate().get(chunkData);
- buf.duplicate().rewind().get(underlying);
-
- tupleOutput.writeInt(size);
- tupleOutput.writeFast(underlying);
+ tupleOutput.writeInt(chunkData.length);
+ tupleOutput.writeFast(chunkData);
}
}
Modified: store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/ExchangeTB.java
===================================================================
--- store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/ExchangeTB.java 2010-05-13 18:28:33 UTC (rev 3970)
+++ store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/ExchangeTB.java 2010-05-14 16:23:59 UTC (rev 3971)
@@ -4,20 +4,15 @@
import com.sleepycat.bind.tuple.TupleInput;
import com.sleepycat.bind.tuple.TupleOutput;
import org.apache.log4j.Logger;
-import org.apache.qpid.AMQException;
import org.apache.qpid.framing.AMQShortString;
-import org.apache.qpid.server.exchange.Exchange;
-import org.apache.qpid.server.virtualhost.VirtualHost;
+import org.apache.qpid.server.store.berkeleydb.records.ExchangeRecord;
public class ExchangeTB extends TupleBinding
{
private static final Logger _log = Logger.getLogger(ExchangeTB.class);
- private final VirtualHost _virtualHost;
-
- public ExchangeTB(VirtualHost virtualHost)
+ public ExchangeTB()
{
- _virtualHost = virtualHost;
}
public Object entryToObject(TupleInput tupleInput)
@@ -27,43 +22,17 @@
AMQShortString typeName = AMQShortStringEncoding.readShortString(tupleInput);
boolean autoDelete = tupleInput.readBoolean();
-
- try
- {
- Exchange exchange;
- Exchange existing = _virtualHost.getExchangeRegistry().getExchange(name);
-
- if (existing != null)
- {
- _log.info("Exchange :" + existing + ": already exists in configured broker.");
- exchange = existing;
- }
- else
- {
- exchange = _virtualHost.getExchangeFactory().createExchange(name, typeName, true, autoDelete, 0);
-
- _virtualHost.getExchangeRegistry().registerExchange(exchange);
-
- _log.info("Registering new durable exchange from BDB:" + exchange);
- }
-
- return exchange;
- }
- catch (AMQException e)
- {
- _log.error("Unable to create exchange: " + e, e);
- return null;
- }
+
+ return new ExchangeRecord(name, typeName, autoDelete);
}
public void objectToEntry(Object object, TupleOutput tupleOutput)
{
- Exchange exchange = (Exchange) object;
+ ExchangeRecord exchange = (ExchangeRecord) object;
- AMQShortStringEncoding.writeShortString(exchange.getName(), tupleOutput);
+ AMQShortStringEncoding.writeShortString(exchange.getNameShortString(), tupleOutput);
AMQShortStringEncoding.writeShortString(exchange.getType(), tupleOutput);
tupleOutput.writeBoolean(exchange.isAutoDelete());
-
}
}
Modified: store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/MessageContentKey.java
===================================================================
--- store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/MessageContentKey.java 2010-05-13 18:28:33 UTC (rev 3970)
+++ store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/MessageContentKey.java 2010-05-14 16:23:59 UTC (rev 3971)
@@ -1,223 +1,42 @@
/*
*
- * Copyright (c) 2006 The Apache Software Foundation
+ * 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
*
- * Licensed 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
*
- * 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.
*
- * 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.TupleInput;
-import com.sleepycat.bind.tuple.TupleOutput;
-
-import java.util.Comparator;
-import java.io.Serializable;
-
-/**
- * @author Apache Software Foundation
- */
public class MessageContentKey
{
- public long messageId;
- public int chunk;
-
- public MessageContentKey()
+ private long _messageId;
+
+ public MessageContentKey(long messageId)
{
- }
-
- public MessageContentKey(byte[] payload)
+ _messageId = messageId;
+ }
+
+
+ public long getMessageId()
{
- final TupleInput ti = new TupleInput(payload);
- messageId = ti.readLong();
- chunk = ti.readInt();
+ return _messageId;
}
- public static class TupleBinding extends com.sleepycat.bind.tuple.TupleBinding
+ public void setMessageId(long messageId)
{
- public Object entryToObject(TupleInput tupleInput)
- {
- final MessageContentKey mk = new MessageContentKey();
- mk.messageId = tupleInput.readLong();
- mk.chunk = tupleInput.readInt();
- return mk;
- }
-
- public void objectToEntry(Object object, TupleOutput tupleOutput)
- {
- final MessageContentKey mk = (MessageContentKey) object;
- tupleOutput.writeLong(mk.messageId);
- tupleOutput.writeInt(mk.chunk);
- }
-
-
+ this._messageId = messageId;
}
-
- public static void writeModifiedLong(TupleOutput t, long l)
- {
- int ln = (int) (0x0F & l);
- int hn = (int) (0xF0 & l);
- t.writeByte(((ln <<4) | (hn>>4)) ^ 0x55);
- t.writeByte((int) (0xFF & (l >> 8)));
- t.writeByte((int) (0xFF & (l >> 16)));
- t.writeByte((int) (0xFF & (l >> 24)));
- t.writeByte((int) (0xFF & (l >> 32)));
- t.writeByte((int) (0xFF & (l >> 40)));
- t.writeByte((int) (0xFF & (l >> 48)));
- t.writeByte((int) (0xFF & (l >> 56)));
-
- }
-
- public static long readModifiedLong(TupleInput t)
- {
- byte b0 = (byte) (0xFF & (t.readByte() ^ 0x55));
- long l = (0x0f & (b0 >> 4)) | ((0xf0 & (b0 << 4)));
-
- l |= ((long) (0xFF & t.readByte())) << 8;
- l |= ((long) (0xFF & t.readByte())) << 16;
- l |= ((long) (0xFF & t.readByte())) << 24;
- l |= ((long) (0xFF & t.readByte())) << 32;
- l |= ((long) (0xFF & t.readByte())) << 40;
- l |= ((long) (0xFF & t.readByte())) << 48;
- l |= ((long) (0xFF & t.readByte())) << 56;
-
- return l;
- }
-
-
- public static class ContentKeyComparator implements Comparator, Serializable
- {
-
- public int compare(Object o1, Object o2)
- {
- byte[] b1 = (byte[]) o1;
- byte[] b2 = (byte[]) o2;
-
- MessageContentKey ck1 = new MessageContentKey(b1);
- MessageContentKey ck2 = new MessageContentKey(b2);
- if (ck1.messageId == ck2.messageId)
- {
- // reminder of Comparator return value:
- // return a negative integer if the first item is "less" than the second, 0 if equal
- return ck1.chunk - ck2.chunk;
- }
- else
- {
- return (int) (ck1.messageId - ck2.messageId);
- }
- }
- }
-
- public MessageContentKey(long messageId, int chunk)
- {
- this.chunk = chunk;
- this.messageId = messageId;
- }
-
-
-
- private static final char[] HEX = { '0','1','2','3','4','5','6','7','8','9','A','B','C','D','E','F'};
-
- private static StringBuilder appendHex(StringBuilder str, byte b)
- {
- str.append(HEX[0xF & (b >> 4)]);
- str.append(HEX[0xF & b]);
- return str;
- }
-
- private static StringBuilder appendHex(StringBuilder str, byte[] b)
- {
- for(int i = 0; i < b.length; i++)
- {
- appendHex(str,b[i]);
- }
- return str;
- }
-
- private static StringBuilder appendHex(StringBuilder str, int i)
- {
- appendHex(str,(byte)(0xFF & (i >> 24)));
- appendHex(str,(byte)(0xFF & (i >> 16)));
- appendHex(str,(byte)(0xFF & (i >> 8)));
- appendHex(str,(byte)(0xFF & i));
- return str;
- }
-
-
- private static StringBuilder appendHex(StringBuilder str, long l)
- {
- appendHex(str,(byte)(0xFF & (l >> 56)));
- appendHex(str,(byte)(0xFF & (l >> 48)));
- appendHex(str,(byte)(0xFF & (l >> 40)));
- appendHex(str,(byte)(0xFF & (l >> 32)));
- appendHex(str,(byte)(0xFF & (l >> 24)));
- appendHex(str,(byte)(0xFF & (l >> 16)));
- appendHex(str,(byte)(0xFF & (l >> 8)));
- appendHex(str,(byte)(0xFF & l));
- return str;
- }
-
-
-
-
- private static byte[] convertLong(long l)
- {
- byte[] b = new byte[8];
- int ln = (int) (0x0F & l);
- int hn = (int) (0xF0 & l);
- b[0] = (byte)(((ln <<4) | (hn>>4)) ^ 0x55);
- b[1] = ((byte)(0xFF & (l >> 8)));
- b[2] = ((byte)(0xFF & (l >> 16)));
- b[3] = ((byte)(0xFF & (l >> 24)));
- b[4] = ((byte)(0xFF & (l >> 32)));
- b[5] = ((byte)(0xFF & (l >> 40)));
- b[6] = ((byte)(0xFF & (l >> 48)));
- b[7] = ((byte)(0xFF & (l >> 56)));
-
- return b;
-
- }
-
-
- private static long readModifiedLong(byte[] b)
- {
- byte b0 = (byte) (b[0] ^ 0x55);
- long l = (0x0f & (b0 >> 4)) | ((0xf0 & (b0 << 4)));
- l |= ((long) b[1]) << 8;
- l |= ((long) b[2]) << 16;
- l |= ((long) b[3]) << 24;
- l |= ((long) b[4]) << 32;
- l |= ((long) b[5]) << 40;
- l |= ((long) b[6]) << 48;
- l |= ((long) b[7]) << 56;
-
-
- return l;
- }
-
-
- public static void main(String[] args)
- {
- StringBuilder s = new StringBuilder();
-
-
-
- for(long i = 1000; i < 1010; i++)
- {
- byte[] b = convertLong(i);
- System.out.println(appendHex(new StringBuilder(),b));
- System.out.println(readModifiedLong(b));
-
- }
-
- }
-}
+}
\ No newline at end of file
Deleted: store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/MessageMetaDataTB.java
===================================================================
--- store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/MessageMetaDataTB.java 2010-05-13 18:28:33 UTC (rev 3970)
+++ store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/MessageMetaDataTB.java 2010-05-14 16:23:59 UTC (rev 3971)
@@ -1,142 +0,0 @@
-/*
- *
- * Copyright (c) 2006 The Apache Software Foundation
- *
- * Licensed 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.mina.common.ByteBuffer;
-import org.apache.qpid.AMQException;
-import org.apache.qpid.framing.*;
-import org.apache.qpid.framing.abstraction.MessagePublishInfo;
-import org.apache.qpid.server.queue.MessageMetaData;
-
-/**
- * Handles the mapping to and from message meta data
- */
-public class MessageMetaDataTB extends TupleBinding
-{
- private static final Logger _log = Logger.getLogger(MessageMetaDataTB.class);
-
- public Object entryToObject(TupleInput tupleInput)
- {
- try
- {
- final MessagePublishInfo publishBody = readMessagePublishInfo(tupleInput);
- final ContentHeaderBody contentHeaderBody = readContentHeaderBody(tupleInput);
- final int contentChunkCount = tupleInput.readInt();
- return new MessageMetaData(publishBody, contentHeaderBody, contentChunkCount);
- }
- catch (Exception e)
- {
- _log.error("Error converting entry to object: " + e, e);
- // annoyingly just have to return null since we cannot throw
- return null;
- }
- }
-
- public void objectToEntry(Object object, TupleOutput tupleOutput)
- {
- MessageMetaData message = (MessageMetaData) object;
- try
- {
- writeMessagePublishInfo(message.getMessagePublishInfo(), tupleOutput);
- }
- catch (AMQException e)
- {
- // can't do anything else since the BDB interface precludes throwing any exceptions
- // in practice we should never get an exception
- throw new RuntimeException("Error converting object to entry: " + e, e);
- }
- writeContentHeader(message.getContentHeaderBody(), tupleOutput);
- tupleOutput.writeInt(message.getContentChunkCount());
- }
-
- private MessagePublishInfo readMessagePublishInfo(TupleInput tupleInput)
- {
-
- final AMQShortString exchange = AMQShortStringEncoding.readShortString(tupleInput);
- final AMQShortString routingKey = AMQShortStringEncoding.readShortString(tupleInput);
- final boolean mandatory = tupleInput.readBoolean();
- final boolean immediate = tupleInput.readBoolean();
-
- return new MessagePublishInfo()
- {
-
- public AMQShortString getExchange()
- {
- return exchange;
- }
-
- public void setExchange(AMQShortString exchange)
- {
-
- }
-
- public boolean isImmediate()
- {
- return immediate;
- }
-
- public boolean isMandatory()
- {
- return mandatory;
- }
-
- public AMQShortString getRoutingKey()
- {
- return routingKey;
- }
- } ;
-
- }
-
-
- private ContentHeaderBody readContentHeaderBody(TupleInput tupleInput) throws AMQFrameDecodingException, AMQProtocolVersionException
- {
- int bodySize = tupleInput.readInt();
- byte[] underlying = new byte[bodySize];
- tupleInput.readFast(underlying);
- ByteBuffer buf = ByteBuffer.wrap(underlying);
-
- return ContentHeaderBody.createFromBuffer(buf, bodySize);
- }
-
- private void writeMessagePublishInfo(MessagePublishInfo publishBody, TupleOutput tupleOutput) throws AMQException
- {
-
-
- AMQShortStringEncoding.writeShortString(publishBody.getExchange(), tupleOutput);
- AMQShortStringEncoding.writeShortString(publishBody.getRoutingKey(), tupleOutput);
- tupleOutput.writeBoolean(publishBody.isMandatory());
- tupleOutput.writeBoolean(publishBody.isImmediate());
-
- }
-
- private void writeContentHeader(ContentHeaderBody headerBody, TupleOutput tupleOutput)
- {
- // write out the content header body
- final int bodySize = headerBody.getSize();
- byte[] underlying = new byte[bodySize];
- ByteBuffer buf = ByteBuffer.wrap(underlying);
- headerBody.writePayload(buf);
- tupleOutput.writeInt(bodySize);
- tupleOutput.writeFast(underlying);
- }
-}
Modified: store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/QueueEntryKey.java
===================================================================
--- store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/QueueEntryKey.java 2010-05-13 18:28:33 UTC (rev 3970)
+++ store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/QueueEntryKey.java 2010-05-14 16:23:59 UTC (rev 3971)
@@ -1,8 +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;
-import com.sleepycat.bind.tuple.TupleInput;
-import com.sleepycat.bind.tuple.TupleOutput;
-
import org.apache.qpid.framing.AMQShortString;
/**
@@ -10,55 +27,26 @@
*/
public class QueueEntryKey
{
- public AMQShortString queueName;
- public long messageId;
+ private AMQShortString _queueName;
+ private long _messageId;
- public QueueEntryKey()
+ public QueueEntryKey(AMQShortString queueName, long messageId)
{
+ _queueName = queueName;
+ _messageId = messageId;
}
- public QueueEntryKey(byte[] payload)
+ public AMQShortString getQueueName()
{
- final TupleInput ti = new TupleInput(payload);
-
- queueName = AMQShortStringEncoding.readShortString(ti);
-
- messageId = ti.readLong();
-
+ return _queueName;
}
- public static class TupleBinding extends com.sleepycat.bind.tuple.TupleBinding
- {
- public Object entryToObject(TupleInput tupleInput)
- {
- final QueueEntryKey mk = new QueueEntryKey();
-
- mk.queueName = AMQShortStringEncoding.readShortString(tupleInput);
- mk.messageId = tupleInput.readLong();
-
- return mk;
- }
-
- public void objectToEntry(Object object, TupleOutput tupleOutput)
- {
- final QueueEntryKey mk = (QueueEntryKey) object;
-
- AMQShortStringEncoding.writeShortString(mk.queueName,tupleOutput);
- tupleOutput.writeLong(mk.messageId);
-
- }
-
-
- }
-
- public QueueEntryKey(AMQShortString queueName, long messageId)
+ public long getMessageId()
{
- this.queueName = queueName;
- this.messageId = messageId;
+ return _messageId;
}
-
}
Copied: store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/keys/MessageContentKey_1.java (from rev 3969, store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/BindingTuple.java)
===================================================================
--- store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/keys/MessageContentKey_1.java (rev 0)
+++ store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/keys/MessageContentKey_1.java 2010-05-14 16:23:59 UTC (rev 3971)
@@ -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_1 extends MessageContentKey
+{
+ private int _chunkNum;
+
+ public MessageContentKey_1(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
Copied: store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/keys/MessageContentKey_3.java (from rev 3969, store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/BindingTuple.java)
===================================================================
--- store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/keys/MessageContentKey_3.java (rev 0)
+++ store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/keys/MessageContentKey_3.java 2010-05-14 16:23:59 UTC (rev 3971)
@@ -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_3 extends MessageContentKey
+{
+ private int _offset;
+
+ public MessageContentKey_3(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
Copied: store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/records/ExchangeRecord.java (from rev 3969, store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/BindingTupleBindingFactory.java)
===================================================================
--- store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/records/ExchangeRecord.java (rev 0)
+++ store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/records/ExchangeRecord.java 2010-05-14 16:23:59 UTC (rev 3971)
@@ -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;
+ }
+
+}
Copied: store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/records/QueueRecord.java (from rev 3969, store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/BindingTupleBindingFactory.java)
===================================================================
--- store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/records/QueueRecord.java (rev 0)
+++ store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/records/QueueRecord.java 2010-05-14 16:23:59 UTC (rev 3971)
@@ -0,0 +1,54 @@
+/*
+ *
+ * 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;
+
+ public QueueRecord(AMQShortString queueName, AMQShortString owner, FieldTable arguments)
+ {
+ _queueName = queueName;
+ _owner = owner;
+ _arguments = arguments;
+ }
+
+ public AMQShortString getNameShortString()
+ {
+ return _queueName;
+ }
+
+ public AMQShortString getOwner()
+ {
+ return _owner;
+ }
+
+ public FieldTable getArguments()
+ {
+ return _arguments;
+ }
+
+}
Modified: store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/BindingTuple.java
===================================================================
--- store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/BindingTuple.java 2010-05-13 18:28:33 UTC (rev 3970)
+++ store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/BindingTuple.java 2010-05-14 16:23:59 UTC (rev 3971)
@@ -20,8 +20,6 @@
*/
package org.apache.qpid.server.store.berkeleydb.tuples;
-import org.apache.qpid.framing.FieldTable;
-
public interface BindingTuple
{
}
Modified: store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/BindingTupleBindingFactory.java
===================================================================
--- store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/BindingTupleBindingFactory.java 2010-05-13 18:28:33 UTC (rev 3970)
+++ store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/BindingTupleBindingFactory.java 2010-05-14 16:23:59 UTC (rev 3971)
@@ -21,13 +21,14 @@
package org.apache.qpid.server.store.berkeleydb.tuples;
import com.sleepycat.bind.tuple.TupleBinding;
-import org.apache.qpid.server.virtualhost.VirtualHost;
+import org.apache.qpid.server.store.ConfigurationRecoveryHandler.BindingRecoveryHandler;
+
public class BindingTupleBindingFactory extends TupleBindingFactory
{
- public BindingTupleBindingFactory(int version, VirtualHost virtualhost)
+ public BindingTupleBindingFactory(int version)
{
- super(version, virtualhost);
+ super(version);
}
public TupleBinding getInstance()
@@ -35,10 +36,12 @@
switch (_version)
{
default:
+ case 3:
+ //no change from v2
case 2:
- return new BindingTuple_2(_virtualhost);
+ return new BindingTuple_2();
case 1:
- return new BindingTuple_1(_virtualhost);
+ return new BindingTuple_1();
}
}
}
Modified: store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/BindingTuple_1.java
===================================================================
--- store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/BindingTuple_1.java 2010-05-13 18:28:33 UTC (rev 3970)
+++ store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/BindingTuple_1.java 2010-05-14 16:23:59 UTC (rev 3971)
@@ -1,11 +1,8 @@
package org.apache.qpid.server.store.berkeleydb.tuples;
-import org.apache.qpid.server.virtualhost.VirtualHost;
import org.apache.qpid.server.store.berkeleydb.AMQShortStringEncoding;
-import org.apache.qpid.server.store.berkeleydb.FieldTableEncoding;
import org.apache.qpid.server.store.berkeleydb.BindingKey;
import org.apache.qpid.framing.AMQShortString;
-import org.apache.qpid.framing.FieldTable;
import com.sleepycat.bind.tuple.TupleBinding;
import com.sleepycat.bind.tuple.TupleInput;
@@ -16,15 +13,8 @@
{
protected static final Logger _log = Logger.getLogger(BindingTuple.class);
- protected VirtualHost _virtualhost;
-
- public BindingTuple_1(VirtualHost virtualHost)
+ public BindingTuple_1()
{
- if (virtualHost == null)
- {
- throw new NullPointerException("Virtualhost cannot be null");
- }
- _virtualhost = virtualHost;
}
public Object entryToObject(TupleInput tupleInput)
@@ -33,7 +23,7 @@
AMQShortString queueName = AMQShortStringEncoding.readShortString(tupleInput);
AMQShortString routingKey = AMQShortStringEncoding.readShortString(tupleInput);
- return createNewBindingKey(exchangeName, queueName, routingKey);
+ return new BindingKey(exchangeName, queueName, routingKey, null);
}
public void objectToEntry(Object object, TupleOutput tupleOutput)
@@ -45,16 +35,4 @@
AMQShortStringEncoding.writeShortString(binding.getRoutingKey(), tupleOutput);
}
- private Object createNewBindingKey(AMQShortString exchangeName, AMQShortString queueName, AMQShortString routingKey)
- {
- return createNewBindingKey(exchangeName, queueName, routingKey, null);
- }
-
- // Addition for Version 2 of this table
- protected Object createNewBindingKey(AMQShortString exchangeName, AMQShortString queueName,
- AMQShortString routingKey, FieldTable arguments)
- {
- return new BindingKey(exchangeName, queueName, routingKey, arguments);
- }
-
}
Modified: store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/BindingTuple_2.java
===================================================================
--- store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/BindingTuple_2.java 2010-05-13 18:28:33 UTC (rev 3970)
+++ store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/BindingTuple_2.java 2010-05-14 16:23:59 UTC (rev 3971)
@@ -1,24 +1,21 @@
package org.apache.qpid.server.store.berkeleydb.tuples;
-import org.apache.qpid.server.virtualhost.VirtualHost;
import org.apache.qpid.server.store.berkeleydb.AMQShortStringEncoding;
-import org.apache.qpid.server.store.berkeleydb.FieldTableEncoding;
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;
-import org.apache.log4j.Logger;
public class BindingTuple_2 extends BindingTuple_1 implements BindingTuple
{
- public BindingTuple_2(VirtualHost virtualHost)
+ public BindingTuple_2()
{
- super(virtualHost);
+ super();
}
public Object entryToObject(TupleInput tupleInput)
@@ -28,7 +25,7 @@
AMQShortString routingKey = AMQShortStringEncoding.readShortString(tupleInput);
FieldTable arguments;
-
+
// Addition for Version 2 of this table
try
{
@@ -40,7 +37,7 @@
return null;
}
- return createNewBindingKey(exchangeName, queueName, routingKey, arguments);
+ return new BindingKey(exchangeName, queueName, routingKey, arguments);
}
public void objectToEntry(Object object, TupleOutput tupleOutput)
Added: store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/MessageContentKeyTB_1.java
===================================================================
--- store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/MessageContentKeyTB_1.java (rev 0)
+++ store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/MessageContentKeyTB_1.java 2010-05-14 16:23:59 UTC (rev 3971)
@@ -0,0 +1,43 @@
+/*
+ *
+ * Copyright (c) 2006 The Apache Software Foundation
+ *
+ * Licensed 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.keys.MessageContentKey_1;
+
+import com.sleepycat.bind.tuple.TupleInput;
+import com.sleepycat.bind.tuple.TupleOutput;
+
+public class MessageContentKeyTB_1 extends com.sleepycat.bind.tuple.TupleBinding
+{
+
+ public Object entryToObject(TupleInput tupleInput)
+ {
+ long messageId = tupleInput.readLong();
+ int chunk = tupleInput.readInt();
+
+ return new MessageContentKey_1(messageId, chunk);
+ }
+
+ public void objectToEntry(Object object, TupleOutput tupleOutput)
+ {
+ final MessageContentKey_1 mk = (MessageContentKey_1) object;
+ tupleOutput.writeLong(mk.getMessageId());
+ tupleOutput.writeInt(mk.getChunk());
+ }
+
+}
\ No newline at end of file
Added: store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/MessageContentKeyTB_3.java
===================================================================
--- store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/MessageContentKeyTB_3.java (rev 0)
+++ store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/MessageContentKeyTB_3.java 2010-05-14 16:23:59 UTC (rev 3971)
@@ -0,0 +1,43 @@
+/*
+ *
+ * Copyright (c) 2006 The Apache Software Foundation
+ *
+ * Licensed 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.keys.MessageContentKey_3;
+
+import com.sleepycat.bind.tuple.TupleInput;
+import com.sleepycat.bind.tuple.TupleOutput;
+
+public class MessageContentKeyTB_3 extends com.sleepycat.bind.tuple.TupleBinding
+{
+
+ public Object entryToObject(TupleInput tupleInput)
+ {
+ long messageId = tupleInput.readLong();
+ int offset = tupleInput.readInt();
+
+ return new MessageContentKey_3(messageId, offset);
+ }
+
+ public void objectToEntry(Object object, TupleOutput tupleOutput)
+ {
+ final MessageContentKey_3 mk = (MessageContentKey_3) object;
+ tupleOutput.writeLong(mk.getMessageId());
+ tupleOutput.writeInt(mk.getOffset());
+ }
+
+}
\ No newline at end of file
Copied: store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/MessageContentKeyTupleBindingFactory.java (from rev 3969, store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/QueueTupleBindingFactory.java)
===================================================================
--- store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/MessageContentKeyTupleBindingFactory.java (rev 0)
+++ store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/MessageContentKeyTupleBindingFactory.java 2010-05-14 16:23:59 UTC (rev 3971)
@@ -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 com.sleepycat.bind.tuple.TupleBinding;
+
+public class MessageContentKeyTupleBindingFactory extends TupleBindingFactory
+{
+ public MessageContentKeyTupleBindingFactory(int version)
+ {
+ super(version);
+ }
+
+ public TupleBinding getInstance()
+ {
+ switch (_version)
+ {
+ default:
+ case 3:
+ return new MessageContentKeyTB_3();
+ case 2:
+ //no change from v1
+ case 1:
+ return new MessageContentKeyTB_1();
+ }
+ }
+}
Copied: store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/MessageMetaDataTB_1.java (from rev 3969, store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/MessageMetaDataTB.java)
===================================================================
--- store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/MessageMetaDataTB_1.java (rev 0)
+++ store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/MessageMetaDataTB_1.java 2010-05-14 16:23:59 UTC (rev 3971)
@@ -0,0 +1,152 @@
+/*
+ *
+ * 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 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.mina.common.ByteBuffer;
+import org.apache.qpid.AMQException;
+import org.apache.qpid.framing.*;
+import org.apache.qpid.framing.abstraction.MessagePublishInfo;
+import org.apache.qpid.server.message.MessageMetaData;
+import org.apache.qpid.server.store.MessageStoreRecoveryHandler.StoredMessageRecoveryHandler;
+import org.apache.qpid.server.store.berkeleydb.AMQShortStringEncoding;
+
+/**
+ * Handles the mapping to and from 0-8/0-9 message meta data
+ */
+public class MessageMetaDataTB_1 extends TupleBinding
+{
+ private static final Logger _log = Logger.getLogger(MessageMetaDataTB_1.class);
+
+ public MessageMetaDataTB_1()
+ {
+ }
+
+ public Object entryToObject(TupleInput tupleInput)
+ {
+ try
+ {
+ final MessagePublishInfo publishBody = readMessagePublishInfo(tupleInput);
+ final ContentHeaderBody contentHeaderBody = readContentHeaderBody(tupleInput);
+ final int contentChunkCount = tupleInput.readInt();
+
+ return new MessageMetaData(publishBody, contentHeaderBody, contentChunkCount);
+ }
+ catch (Exception e)
+ {
+ _log.error("Error converting entry to object: " + e, e);
+ // annoyingly just have to return null since we cannot throw
+ return null;
+ }
+ }
+
+ public void objectToEntry(Object object, TupleOutput tupleOutput)
+ {
+ MessageMetaData message = (MessageMetaData) object;
+ try
+ {
+ writeMessagePublishInfo(message.getMessagePublishInfo(), tupleOutput);
+ }
+ catch (AMQException e)
+ {
+ // can't do anything else since the BDB interface precludes throwing any exceptions
+ // in practice we should never get an exception
+ throw new RuntimeException("Error converting object to entry: " + e, e);
+ }
+ writeContentHeader(message.getContentHeaderBody(), tupleOutput);
+ tupleOutput.writeInt(message.getContentChunkCount());
+ }
+
+ private MessagePublishInfo readMessagePublishInfo(TupleInput tupleInput)
+ {
+
+ final AMQShortString exchange = AMQShortStringEncoding.readShortString(tupleInput);
+ final AMQShortString routingKey = AMQShortStringEncoding.readShortString(tupleInput);
+ final boolean mandatory = tupleInput.readBoolean();
+ final boolean immediate = tupleInput.readBoolean();
+
+ return new MessagePublishInfo()
+ {
+
+ public AMQShortString getExchange()
+ {
+ return exchange;
+ }
+
+ public void setExchange(AMQShortString exchange)
+ {
+
+ }
+
+ public boolean isImmediate()
+ {
+ return immediate;
+ }
+
+ public boolean isMandatory()
+ {
+ return mandatory;
+ }
+
+ public AMQShortString getRoutingKey()
+ {
+ return routingKey;
+ }
+ } ;
+
+ }
+
+
+ private ContentHeaderBody readContentHeaderBody(TupleInput tupleInput) throws AMQFrameDecodingException, AMQProtocolVersionException
+ {
+ int bodySize = tupleInput.readInt();
+ byte[] underlying = new byte[bodySize];
+ tupleInput.readFast(underlying);
+ ByteBuffer buf = ByteBuffer.wrap(underlying);
+
+ return ContentHeaderBody.createFromBuffer(buf, bodySize);
+ }
+
+ private void writeMessagePublishInfo(MessagePublishInfo publishBody, TupleOutput tupleOutput) throws AMQException
+ {
+
+
+ AMQShortStringEncoding.writeShortString(publishBody.getExchange(), tupleOutput);
+ AMQShortStringEncoding.writeShortString(publishBody.getRoutingKey(), tupleOutput);
+ tupleOutput.writeBoolean(publishBody.isMandatory());
+ tupleOutput.writeBoolean(publishBody.isImmediate());
+
+ }
+
+ private void writeContentHeader(ContentHeaderBody headerBody, TupleOutput tupleOutput)
+ {
+ // write out the content header body
+ final int bodySize = headerBody.getSize();
+ byte[] underlying = new byte[bodySize];
+ ByteBuffer buf = ByteBuffer.wrap(underlying);
+ headerBody.writePayload(buf);
+ tupleOutput.writeInt(bodySize);
+ tupleOutput.writeFast(underlying);
+ }
+}
Added: store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/MessageMetaDataTB_3.java
===================================================================
--- store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/MessageMetaDataTB_3.java (rev 0)
+++ store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/MessageMetaDataTB_3.java 2010-05-14 16:23:59 UTC (rev 3971)
@@ -0,0 +1,77 @@
+/*
+ *
+ * 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 com.sleepycat.bind.tuple.TupleInput;
+import com.sleepycat.bind.tuple.TupleOutput;
+import org.apache.log4j.Logger;
+import org.apache.qpid.server.store.MessageMetaDataType;
+import org.apache.qpid.server.store.StorableMessageMetaData;
+
+/**
+ * Handles the mapping to and from message meta data
+ */
+public class MessageMetaDataTB_3 extends MessageMetaDataTB_1
+{
+ private static final Logger _log = Logger.getLogger(MessageMetaDataTB_3.class);
+
+ @Override
+ public Object entryToObject(TupleInput tupleInput)
+ {
+ try
+ {
+ final int bodySize = tupleInput.readInt();
+ byte[] dataAsBytes = new byte[bodySize];
+ tupleInput.readFast(dataAsBytes);
+
+ java.nio.ByteBuffer buf = java.nio.ByteBuffer.wrap(dataAsBytes);
+ buf.position(1);
+ buf = buf.slice();
+ MessageMetaDataType type = MessageMetaDataType.values()[dataAsBytes[0]];
+ StorableMessageMetaData metaData = type.getFactory().createMetaData(buf);
+
+ return metaData;
+ }
+ catch (Exception e)
+ {
+ _log.error("Error converting entry to object: " + e, e);
+ // annoyingly just have to return null since we cannot throw
+ return null;
+ }
+ }
+
+ @Override
+ public void objectToEntry(Object object, TupleOutput tupleOutput)
+ {
+ StorableMessageMetaData metaData = (StorableMessageMetaData) object;
+
+ final int bodySize = 1 + metaData.getStorableSize();
+ byte[] underlying = new byte[bodySize];
+ underlying[0] = (byte) metaData.getType().ordinal();
+ java.nio.ByteBuffer buf = java.nio.ByteBuffer.wrap(underlying);
+ buf.position(1);
+ buf = buf.slice();
+
+ metaData.writeToBuffer(0, buf);
+ tupleOutput.writeInt(bodySize);
+ tupleOutput.writeFast(underlying);
+ }
+}
Copied: store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/MessageMetaDataTupleBindingFactory.java (from rev 3969, store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/QueueTupleBindingFactory.java)
===================================================================
--- store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/MessageMetaDataTupleBindingFactory.java (rev 0)
+++ store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/MessageMetaDataTupleBindingFactory.java 2010-05-14 16:23:59 UTC (rev 3971)
@@ -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 com.sleepycat.bind.tuple.TupleBinding;
+
+public class MessageMetaDataTupleBindingFactory extends TupleBindingFactory
+{
+ public MessageMetaDataTupleBindingFactory(int version)
+ {
+ super(version);
+ }
+
+ public TupleBinding getInstance()
+ {
+ switch (_version)
+ {
+ default:
+ case 3:
+ return new MessageMetaDataTB_3();
+ case 2:
+ //no change from v1
+ case 1:
+ return new MessageMetaDataTB_1();
+ }
+ }
+}
Copied: store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/QueueEntryTB.java (from rev 3969, store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/BindingTupleBindingFactory.java)
===================================================================
--- store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/QueueEntryTB.java (rev 0)
+++ store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/QueueEntryTB.java 2010-05-14 16:23:59 UTC (rev 3971)
@@ -0,0 +1,48 @@
+/*
+ *
+ * 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.framing.AMQShortString;
+import org.apache.qpid.server.store.berkeleydb.AMQShortStringEncoding;
+import org.apache.qpid.server.store.berkeleydb.QueueEntryKey;
+
+import com.sleepycat.bind.tuple.TupleBinding;
+import com.sleepycat.bind.tuple.TupleInput;
+import com.sleepycat.bind.tuple.TupleOutput;
+
+public class QueueEntryTB extends TupleBinding
+{
+ public Object entryToObject(TupleInput tupleInput)
+ {
+ AMQShortString queueName = AMQShortStringEncoding.readShortString(tupleInput);
+ Long messageId = tupleInput.readLong();
+
+ return new QueueEntryKey(queueName, messageId);
+ }
+
+ public void objectToEntry(Object object, TupleOutput tupleOutput)
+ {
+ final QueueEntryKey mk = (QueueEntryKey) object;
+
+ AMQShortStringEncoding.writeShortString(mk.getQueueName(),tupleOutput);
+ tupleOutput.writeLong(mk.getMessageId());
+ }
+}
\ No newline at end of file
Modified: store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/QueueTuple.java
===================================================================
--- store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/QueueTuple.java 2010-05-13 18:28:33 UTC (rev 3970)
+++ store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/QueueTuple.java 2010-05-14 16:23:59 UTC (rev 3971)
@@ -20,10 +20,6 @@
*/
package org.apache.qpid.server.store.berkeleydb.tuples;
-import org.apache.qpid.framing.FieldTable;
-
public interface QueueTuple
{
- // Addition for Version 2
- public void setArguments(FieldTable arguments);
}
Modified: store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/QueueTupleBindingFactory.java
===================================================================
--- store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/QueueTupleBindingFactory.java 2010-05-13 18:28:33 UTC (rev 3970)
+++ store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/QueueTupleBindingFactory.java 2010-05-14 16:23:59 UTC (rev 3971)
@@ -20,14 +20,14 @@
*/
package org.apache.qpid.server.store.berkeleydb.tuples;
-import org.apache.qpid.server.virtualhost.VirtualHost;
import com.sleepycat.bind.tuple.TupleBinding;
public class QueueTupleBindingFactory extends TupleBindingFactory
{
- public QueueTupleBindingFactory(int version, VirtualHost virtualHost)
+
+ public QueueTupleBindingFactory(int version)
{
- super(version,virtualHost);
+ super(version);
}
public TupleBinding getInstance()
@@ -35,10 +35,12 @@
switch (_version)
{
default:
+ case 3:
+ //no change from v2
case 2:
- return new QueueTuple_2(_virtualhost);
+ return new QueueTuple_2();
case 1:
- return new QueueTuple_1(_virtualhost);
+ return new QueueTuple_1();
}
}
}
Modified: store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/QueueTuple_1.java
===================================================================
--- store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/QueueTuple_1.java 2010-05-13 18:28:33 UTC (rev 3970)
+++ store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/QueueTuple_1.java 2010-05-14 16:23:59 UTC (rev 3971)
@@ -20,28 +20,15 @@
import com.sleepycat.bind.tuple.TupleBinding;
import com.sleepycat.bind.tuple.TupleInput;
import com.sleepycat.bind.tuple.TupleOutput;
-import org.apache.log4j.Logger;
-import org.apache.qpid.AMQException;
import org.apache.qpid.framing.AMQShortString;
-import org.apache.qpid.framing.FieldTable;
-import org.apache.qpid.server.queue.AMQQueue;
-import org.apache.qpid.server.queue.AMQQueueFactory;
import org.apache.qpid.server.store.berkeleydb.AMQShortStringEncoding;
-import org.apache.qpid.server.virtualhost.VirtualHost;
+import org.apache.qpid.server.store.berkeleydb.records.QueueRecord;
public class QueueTuple_1 extends TupleBinding implements QueueTuple
{
- protected static final Logger _logger = Logger.getLogger(QueueTuple.class);
- protected final VirtualHost _virtualHost;
-
- public QueueTuple_1(VirtualHost virtualHost)
+ public QueueTuple_1()
{
- if (virtualHost == null)
- {
- throw new NullPointerException("Virtualhost cannot be null");
- }
- _virtualHost = virtualHost;
}
public Object entryToObject(TupleInput tupleInput)
@@ -49,58 +36,15 @@
AMQShortString name = AMQShortStringEncoding.readShortString(tupleInput);
AMQShortString owner = AMQShortStringEncoding.readShortString(tupleInput);
- return createNewQueue(name, owner);
+ return new QueueRecord(name, owner, null);
}
public void objectToEntry(Object object, TupleOutput tupleOutput)
{
- AMQQueue queue = (AMQQueue) object;
+ QueueRecord queue = (QueueRecord) object;
- AMQShortStringEncoding.writeShortString(queue.getName(), tupleOutput);
+ AMQShortStringEncoding.writeShortString(queue.getNameShortString(), tupleOutput);
AMQShortStringEncoding.writeShortString(queue.getOwner(), tupleOutput);
+ //Version1 tuple can't store the queue arguments.
}
-
- // Addition for Version 2 of this table
- public void setArguments(FieldTable arguments)
- {
- //no-op
- }
-
- protected Object createNewQueue(AMQShortString name, AMQShortString owner)
- {
- return createNewQueue(name, owner, null);
- }
-
- // Addition for Version 2 of this table
- protected Object createNewQueue(AMQShortString name, AMQShortString owner, FieldTable arguments)
- {
- try
- {
-
- AMQQueue queue;
- AMQQueue existing = _virtualHost.getQueueRegistry().getQueue(name);
-
- if (existing != null)
- {
- _logger.info("Queue :" + existing + ": already exists in configured broker.");
-
- queue = existing;
- }
- else
- {
- // Retrieve the existing Queue object
- queue = AMQQueueFactory.createAMQQueueImpl(name, true, owner, false, _virtualHost, arguments);
- _virtualHost.getQueueRegistry().registerQueue(queue);
- _logger.info("Recovering queue " + queue.getName() + " with owner:"
- + ((queue.getOwner() == null) ? " <<null>> " : (" \"" + queue.getOwner() + "\" ")));
- }
-
- return queue;
- }
- catch (AMQException e)
- {
- _logger.error("Unable to create queue: " + e, e);
- return null;
- }
- }
}
Modified: store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/QueueTuple_2.java
===================================================================
--- store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/QueueTuple_2.java 2010-05-13 18:28:33 UTC (rev 3970)
+++ store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/QueueTuple_2.java 2010-05-14 16:23:59 UTC (rev 3971)
@@ -20,22 +20,23 @@
import com.sleepycat.bind.tuple.TupleInput;
import com.sleepycat.bind.tuple.TupleOutput;
import com.sleepycat.je.DatabaseException;
-import org.apache.qpid.server.queue.AMQQueue;
-import org.apache.qpid.server.queue.AMQQueueFactory;
-import org.apache.qpid.server.virtualhost.VirtualHost;
+
+import org.apache.log4j.Logger;
import org.apache.qpid.server.store.berkeleydb.AMQShortStringEncoding;
import org.apache.qpid.server.store.berkeleydb.FieldTableEncoding;
-import org.apache.qpid.AMQException;
+import org.apache.qpid.server.store.berkeleydb.records.QueueRecord;
import org.apache.qpid.framing.AMQShortString;
import org.apache.qpid.framing.FieldTable;
public class QueueTuple_2 extends QueueTuple_1
{
+ protected static final Logger _logger = Logger.getLogger(QueueTuple_2.class);
+
protected FieldTable _arguments;
- public QueueTuple_2(VirtualHost virtualHost)
+ public QueueTuple_2()
{
- super(virtualHost);
+ super();
}
public Object entryToObject(TupleInput tupleInput)
@@ -44,10 +45,10 @@
{
AMQShortString name = AMQShortStringEncoding.readShortString(tupleInput);
AMQShortString owner = AMQShortStringEncoding.readShortString(tupleInput);
- // Addition for Version 2 of this table
+ // Addition for Version 2 of this table, read the queue arguments
FieldTable arguments = FieldTableEncoding.readFieldTable(tupleInput);
- return createNewQueue(name, owner, arguments);
+ return new QueueRecord(name, owner, arguments);
}
catch (DatabaseException e)
{
@@ -59,17 +60,11 @@
public void objectToEntry(Object object, TupleOutput tupleOutput)
{
- AMQQueue queue = (AMQQueue) object;
+ QueueRecord queue = (QueueRecord) object;
- AMQShortStringEncoding.writeShortString(queue.getName(), tupleOutput);
+ AMQShortStringEncoding.writeShortString(queue.getNameShortString(), tupleOutput);
AMQShortStringEncoding.writeShortString(queue.getOwner(), tupleOutput);
- // Addition for Version 2 of this table
- FieldTableEncoding.writeFieldTable(_arguments, tupleOutput);
+ // Addition for Version 2 of this table, store the queue arguments
+ FieldTableEncoding.writeFieldTable(queue.getArguments(), tupleOutput);
}
-
- // Addition for Version 2 of this table
- public void setArguments(FieldTable arguments)
- {
- _arguments = arguments;
- }
}
Modified: store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/TupleBindingFactory.java
===================================================================
--- store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/TupleBindingFactory.java 2010-05-13 18:28:33 UTC (rev 3970)
+++ store/trunk/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/TupleBindingFactory.java 2010-05-14 16:23:59 UTC (rev 3971)
@@ -27,17 +27,9 @@
{
protected int _version;
- protected VirtualHost _virtualhost;
-
- public TupleBindingFactory(int version, VirtualHost virtualhost)
+ public TupleBindingFactory(int version)
{
- if (virtualhost == null)
- {
- throw new NullPointerException("Virtualhost cannot be null");
- }
-
_version = version;
- _virtualhost = virtualhost;
}
public abstract TupleBinding getInstance();
Modified: store/trunk/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/BDBUpgradeTest.java
===================================================================
--- store/trunk/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/BDBUpgradeTest.java 2010-05-13 18:28:33 UTC (rev 3970)
+++ store/trunk/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/BDBUpgradeTest.java 2010-05-14 16:23:59 UTC (rev 3971)
@@ -25,6 +25,7 @@
import org.apache.qpid.server.store.berkeleydb.utils.Publisher;
import org.apache.qpid.server.store.berkeleydb.utils.DurableSubscriber;
import org.apache.qpid.server.virtualhost.VirtualHost;
+import org.apache.qpid.server.virtualhost.VirtualHostImpl;
import org.apache.qpid.server.configuration.VirtualHostConfiguration;
import org.apache.qpid.client.transport.TransportConnection;
import org.apache.qpid.util.FileUtils;
@@ -200,7 +201,7 @@
testVirtualhost.setProperty("store.version", version);
ApplicationRegistry.getInstance(port).getVirtualHostRegistry().
- registerVirtualHost(new VirtualHost(new VirtualHostConfiguration("bdbtest",testVirtualhost)));
+ registerVirtualHost(new VirtualHostImpl(new VirtualHostConfiguration("bdbtest",testVirtualhost), null /* TODO */));
TransportConnection.createVMBroker(port);
Modified: store/trunk/java/bdbstore/src/tools/java/org/apache/qpid/server/store/berkeleydb/BDBStoreUpgrade.java
===================================================================
--- store/trunk/java/bdbstore/src/tools/java/org/apache/qpid/server/store/berkeleydb/BDBStoreUpgrade.java 2010-05-13 18:28:33 UTC (rev 3970)
+++ store/trunk/java/bdbstore/src/tools/java/org/apache/qpid/server/store/berkeleydb/BDBStoreUpgrade.java 2010-05-14 16:23:59 UTC (rev 3971)
@@ -25,6 +25,7 @@
import org.apache.qpid.server.store.MemoryMessageStore;
import org.apache.qpid.server.queue.AMQQueue;
import org.apache.qpid.server.virtualhost.VirtualHost;
+import org.apache.qpid.server.virtualhost.VirtualHostImpl;
import org.apache.qpid.server.registry.ApplicationRegistry;
import org.apache.qpid.server.configuration.VirtualHostConfiguration;
import org.apache.qpid.server.logging.actors.CurrentActor;
@@ -341,12 +342,12 @@
CurrentActor.set(new BrokerActor(new NullRootMessageLogger()));
// Note the name of the Vhosts is not important, the store doesnot record the name of the vhost.
- _newVirtualHost = new VirtualHost(new VirtualHostConfiguration("Upgraded", new PropertiesConfiguration()), new MemoryMessageStore());
- _oldVirtualHost = new VirtualHost(new VirtualHostConfiguration("Old", new PropertiesConfiguration()), new MemoryMessageStore());
+ _newVirtualHost = new VirtualHostImpl(new VirtualHostConfiguration("Upgraded", new PropertiesConfiguration()), new MemoryMessageStore());
+ _oldVirtualHost = new VirtualHostImpl(new VirtualHostConfiguration("Old", new PropertiesConfiguration()), new MemoryMessageStore());
//Create a new messageStore
_newMessageStore = new BDBMessageStore();
- _newMessageStore.configure(_newVirtualHost, toDir, false);
+ //TODO _newMessageStore.configure(_newVirtualHost, toDir, false);
try
{
@@ -356,7 +357,7 @@
default:
case 1:
_oldMessageStore = new BDBMessageStore(1);
- _oldMessageStore.configure(_oldVirtualHost, fromDir, true);
+ //TODO _oldMessageStore.configure(_oldVirtualHost, fromDir, true);
_oldMessageStore.start();
upgradeFromVersion_1();
break;
More information about the rhmessaging-commits
mailing list