path: root/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/BDBMessageStore.java

/*
 *
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 *
 */
package org.apache.qpid.server.store.berkeleydb;

import java.io.File;
import java.lang.ref.SoftReference;
import java.nio.ByteBuffer;
import java.util.ArrayList;
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 org.apache.commons.configuration.Configuration;
import org.apache.log4j.Logger;
import org.apache.qpid.AMQStoreException;
import org.apache.qpid.framing.AMQShortString;
import org.apache.qpid.framing.FieldTable;
import org.apache.qpid.server.exchange.Exchange;
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.store.ConfigurationRecoveryHandler;
import org.apache.qpid.server.store.DurableConfigurationStore;
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.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_5;
import org.apache.qpid.server.store.berkeleydb.records.ExchangeRecord;
import org.apache.qpid.server.store.berkeleydb.records.QueueRecord;
import org.apache.qpid.server.store.berkeleydb.tuples.BindingTupleBindingFactory;
import org.apache.qpid.server.store.berkeleydb.tuples.MessageContentKeyTB_5;
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 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.
 *
 * <p/><table id="crc"><caption>CRC Card</caption> <tr><th> Responsibilities <th> Collaborations <tr><td> Accept
 * transaction boundary demarcations: Begin, Commit, Abort. <tr><td> Store and remove queues. <tr><td> Store and remove
 * 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>
 */
@SuppressWarnings({"unchecked"})
public class BDBMessageStore implements MessageStore
{
    private static final Logger _log = Logger.getLogger(BDBMessageStore.class);

    static final int DATABASE_FORMAT_VERSION = 5;
    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";
    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 Database _messageContentDb;
    private Database _queueBindingsDb;
    private Database _deliveryDb;
    private Database _exchangeDb;
    private Database _queueDb;

    /* =======
     * Schema:
     * =======
     * 
     * Queue:
     * name(AMQShortString) - name(AMQShortString), owner(AMQShortString), 
     *                        arguments(FieldTable encoded as binary), exclusive (boolean)
     *
     * Exchange:
     * name(AMQShortString) - name(AMQShortString), typeName(AMQShortString), autodelete (boolean)
     *
     * Binding:
     * exchangeName(AMQShortString), queueName(AMQShortString), routingKey(AMQShortString),
     *                                            arguments (FieldTable encoded as binary) - 0 (zero)
     *
     * QueueEntry:
     * queueName(AMQShortString), messageId (long) - 0 (zero)
     *
     * Message (MetaData):
     * messageId (long) - bodySize (integer), metaData (MessageMetaData encoded as binary)
     *
     * Message (Content):
     * messageId (long), byteOffset (integer) - dataLength(integer), data(binary);
     */

    private LogSubject _logSubject;

    private final AtomicLong _messageId = new AtomicLong(0);

    private final CommitThread _commitThread = new CommitThread("Commit-Thread");

    // Factory Classes to create the TupleBinding objects that reflect the version instance of this BDBStore
    private MessageMetaDataTupleBindingFactory _metaDataTupleBindingFactory;
    private QueueTupleBindingFactory _queueTupleBindingFactory;
    private BindingTupleBindingFactory _bindingTupleBindingFactory;

    /** The data version this store should run with */
    private int _version;
    private enum State
    {
        INITIAL,
        CONFIGURING,
        CONFIGURED,
        RECOVERING,
        STARTED,
        CLOSING,
        CLOSED
    }

    private State _state = State.INITIAL;

    private TransactionConfig _transactionConfig = new TransactionConfig();

    private boolean _readOnly = false;

    private boolean _configured;

    
    public BDBMessageStore()
    {
        this(DATABASE_FORMAT_VERSION);
    }

    public BDBMessageStore(int version)
    {
        _version = version;
    }

    private void setDatabaseNames(int version)
    {
        if (version > 1)
        {
            MESSAGEMETADATADB_NAME += "_v" + version;

            MESSAGECONTENTDB_NAME += "_v" + version;

            QUEUEDB_NAME += "_v" + version;

            DELIVERYDB_NAME += "_v" + version;

            EXCHANGEDB_NAME += "_v" + version;

            QUEUEBINDINGSDB_NAME += "_v" + version;
        }
    }
 
    public void configureConfigStore(String name, 
                                     ConfigurationRecoveryHandler recoveryHandler, 
                                     Configuration storeConfiguration,
                                     LogSubject logSubject) throws Exception
    {
        _logSubject = logSubject;
        CurrentActor.get().message(_logSubject, ConfigStoreMessages.CREATED(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.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.CREATED(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.
     *
     * @param name The name of the virtual host using this store
     * @return whether a new store environment was created or not (to indicate whether recovery is necessary)
     *
     * @throws Exception If any error occurs that means the store is unable to configure itself.
     */
    public boolean configure(String name, Configuration storeConfig) throws Exception
    {
        File environmentPath = new File(storeConfig.getString(ENVIRONMENT_PATH_PROPERTY, 
                                System.getProperty("QPID_WORK") + "/bdbstore/" + name));        
        if (!environmentPath.exists())
        {
            if (!environmentPath.mkdirs())
            {
                throw new IllegalArgumentException("Environment path " + environmentPath + " could not be read or created. "
                                                   + "Ensure the path is correct and that the permissions are correct.");
            }
        }

        CurrentActor.get().message(_logSubject, MessageStoreMessages.STORE_LOCATION(environmentPath.getAbsolutePath()));

        _version = storeConfig.getInt(DATABASE_FORMAT_VERSION_PROPERTY, DATABASE_FORMAT_VERSION);

        return configure(environmentPath, false);
    }

    /**
     * @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 AMQStoreException
     * @throws DatabaseException
     */
    protected boolean configure(File environmentPath, boolean readonly) throws AMQStoreException, DatabaseException
    {
        _readOnly = readonly;
        stateTransition(State.INITIAL, State.CONFIGURING);

        _log.info("Configuring BDB message store");

        createTupleBindingFactories(_version);
        
        setDatabaseNames(_version);

        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.
     *
     * This is required if you do not want to perform recovery of the store data
     *
     * @throws AMQStoreException if the store is not in the correct state
     */
    public void start() throws AMQStoreException
    {
        stateTransition(State.CONFIGURING, State.STARTED);
    }

    private boolean setupStore(File storePath, boolean readonly) throws DatabaseException, AMQStoreException
    {
        checkState(State.CONFIGURING);

        boolean newEnvironment = createEnvironment(storePath, readonly);

        verifyVersionByTables();

        openDatabases(readonly);

        if (!readonly)
        {
            _commitThread.start();
        }

        return newEnvironment;
    }

    private void verifyVersionByTables() throws DatabaseException
    {
        for (String s : _environment.getDatabaseNames())
        {
            int versionIndex = s.indexOf("_v");

            // lack of _v index suggests DB is v1
            // so if _version is not v1 then error
            if (versionIndex == -1)
            {
                if (_version != 1)
                {
                    closeEnvironment();
                    throw new IllegalArgumentException("Error: Unable to load BDBStore as version " + _version
                                            + ". Store on disk contains version 1 data.");
                }
                else // DB is v1 and _version is v1
                {
                    continue;
                }
            }

            // Otherwise Check Versions
            int version = Integer.parseInt(s.substring(versionIndex + 2));

            if (version != _version)
            {
                closeEnvironment();
                throw new IllegalArgumentException("Error: Unable to load BDBStore as version " + _version
                                            + ". Store on disk contains version " + version + " data.");
            }
        }
    }

    private synchronized void stateTransition(State requiredState, State newState) throws AMQStoreException
    {
        if (_state != requiredState)
        {
            throw new AMQStoreException("Cannot transition to the state: " + newState + "; need to be in state: " + requiredState
                                   + "; currently in state: " + _state);
        }

        _state = newState;
    }

    private void checkState(State requiredState) throws AMQStoreException
    {
        if (_state != requiredState)
        {
            throw new AMQStoreException("Unexpected state: " + _state + "; required state: " + requiredState);
        }
    }

    private boolean createEnvironment(File environmentPath, boolean readonly) throws DatabaseException
    {
        _log.info("BDB message store using environment path " + environmentPath.getAbsolutePath());
        EnvironmentConfig envConfig = new EnvironmentConfig();
        // This is what allows the creation of the store if it does not already exist.
        envConfig.setAllowCreate(true);
        envConfig.setTransactional(true);
        envConfig.setConfigParam("je.lock.nLockTables", "7");

        // 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"))
        {
            envConfig.setConfigParam("je.txn.deadlockStackTrace", "true");
            envConfig.setConfigParam("je.txn.dumpLocks", "true");
        }
        
        // Set transaction mode
        _transactionConfig.setReadCommitted(true);
        
        //This prevents background threads running which will potentially update the store.
        envConfig.setReadOnly(readonly);
        try
        {
            _environment = new Environment(environmentPath, envConfig);
            return false;
        }
        catch (DatabaseException de)
        {
            if (de.getMessage().contains("Environment.setAllowCreate is false"))
            {
                //Allow the creation this time
                envConfig.setAllowCreate(true);
                if (_environment != null )
                {
                    _environment.cleanLog();
                    _environment.close();
                }
                _environment = new Environment(environmentPath, envConfig);

                return true;
            }
            else
            {
                throw de;
            }
        }
    }

    private void openDatabases(boolean readonly) throws DatabaseException
    {
        DatabaseConfig dbConfig = new DatabaseConfig();
        dbConfig.setTransactional(true);
        dbConfig.setAllowCreate(true);

        //This is required if we are wanting read only access.
        dbConfig.setReadOnly(readonly);

        _messageMetaDataDb = _environment.openDatabase(null, MESSAGEMETADATADB_NAME, dbConfig);
        _queueDb = _environment.openDatabase(null, QUEUEDB_NAME, dbConfig);
        _exchangeDb = _environment.openDatabase(null, EXCHANGEDB_NAME, dbConfig);
        _queueBindingsDb = _environment.openDatabase(null, QUEUEBINDINGSDB_NAME, dbConfig);
        _messageContentDb = _environment.openDatabase(null, MESSAGECONTENTDB_NAME, dbConfig);
        _deliveryDb = _environment.openDatabase(null, DELIVERYDB_NAME, dbConfig);

    }

    /**
     * Called to close and cleanup any resources used by the message store.
     *
     * @throws Exception If the close fails.
     */
    public void close() throws Exception
    {
        if (_state != State.STARTED)
        {
            return;
        }

        _state = State.CLOSING;

        _commitThread.close();
        _commitThread.join();

        if (_messageMetaDataDb != null)
        {
            _log.info("Closing message metadata database");
            _messageMetaDataDb.close();
        }

        if (_messageContentDb != null)
        {
            _log.info("Closing message content database");
            _messageContentDb.close();
        }

        if (_exchangeDb != null)
        {
            _log.info("Closing exchange database");
            _exchangeDb.close();
        }

        if (_queueBindingsDb != null)
        {
            _log.info("Closing bindings database");
            _queueBindingsDb.close();
        }

        if (_queueDb != null)
        {
            _log.info("Closing queue database");
            _queueDb.close();
        }

        if (_deliveryDb != null)
        {
            _log.info("Close delivery database");
            _deliveryDb.close();
        }

        closeEnvironment();

        _state = State.CLOSED;
        
        CurrentActor.get().message(_logSubject,MessageStoreMessages.CLOSED());
    }

    private void closeEnvironment() throws DatabaseException
    {
        if (_environment != null)
        {
            if(!_readOnly)
            {
                // Clean the log before closing. This makes sure it doesn't contain
                // redundant data. Closing without doing this means the cleaner may not
                // get a chance to finish.
                _environment.cleanLog();
            }
            _environment.close();
        }
    }

    
    public void recover(ConfigurationRecoveryHandler recoveryHandler) throws AMQStoreException
    {
        stateTransition(State.CONFIGURED, State.RECOVERING);

        CurrentActor.get().message(_logSubject,MessageStoreMessages.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 AMQStoreException("Error recovering persistent state: " + e.getMessage(), 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();
                boolean exclusive = queueRecord.isExclusive();
                
                FieldTable arguments = queueRecord.getArguments();

                qrh.queue(queueName, owner, exclusive, arguments);
            }

        }
        finally
        {
            if (cursor != null)
            {
                cursor.close();
            }
        }
    }
    
    
    private void loadExchanges(ExchangeRecoveryHandler erh) throws 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.getMessage(), e);
            throw e;
        }
        finally
        {
            if (cursor != null)
            {
                cursor.close();
            }
        }
    }
    
    private void recoverQueueEntries(TransactionLogRecoveryHandler recoveryHandler) 
    throws DatabaseException
    {
        QueueEntryRecoveryHandler qerh = recoveryHandler.begin(this);

        ArrayList<QueueEntryKey> entries = new ArrayList<QueueEntryKey>();
        
        Cursor cursor = null;
        try
        {
            cursor = _deliveryDb.openCursor(null, null);
            DatabaseEntry key = new DatabaseEntry();
            EntryBinding keyBinding = new QueueEntryTB();

            DatabaseEntry value = new DatabaseEntry();

            while (cursor.getNext(key, value, LockMode.RMW) == OperationStatus.SUCCESS)
            {
                QueueEntryKey qek = (QueueEntryKey) keyBinding.entryToObject(key);

                entries.add(qek);
            }

            try
            {
                cursor.close();
            }
            finally
            {
                cursor = null;
            }
            
            for(QueueEntryKey entry : entries)
            {
                AMQShortString queueName = entry.getQueueName();
                long messageId = entry.getMessageId();
                
                qerh.queueEntry(queueName.asString(),messageId);
            }
        }
        catch (DatabaseException e)
        {
            _log.error("Database Error: " + e.getMessage(), e);
            throw e;
        }
        finally
        {
            if (cursor != null)
            {
                cursor.close();
            }
        }

        qerh.completeQueueEntryRecovery();
    }

    /**
     * Removes the specified message from the store.
     *
     * @param messageId Identifies the message to remove.
     *
     * @throws AMQInternalException If the operation fails for any reason.
     */
    public void removeMessage(Long messageId) throws AMQStoreException
    {
        // _log.debug("public void removeMessage(Long messageId = " + messageId): called");

        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);

            if (_log.isDebugEnabled())
            {
                _log.debug("Removing message id " + messageId);
            }

            
            OperationStatus status = _messageMetaDataDb.delete(tx, key);
            if (status == OperationStatus.NOTFOUND)
            {
                tx.abort();

                throw new AMQStoreException("Message metadata not found for message id " + messageId);
            }

            if (_log.isDebugEnabled())
            {
                _log.debug("Deleted metadata for message " + messageId);
            }

            //now remove the content data from the store if there is any.

            DatabaseEntry contentKeyEntry = new DatabaseEntry();
            MessageContentKey_5 mck = new MessageContentKey_5(messageId,0);

            TupleBinding<MessageContentKey> contentKeyTupleBinding = new MessageContentKeyTB_5();
            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)
            {
                mck = (MessageContentKey_5) contentKeyTupleBinding.entryToObject(contentKeyEntry);
                
                if(mck.getMessageId() != messageId)
                {
                    //we have exhausted all chunks for this message id, break
                    break;
                }
                else
                {
                    status = cursor.delete();
                    
                    if(status == OperationStatus.NOTFOUND)
                    {
                        cursor.close();
                        cursor = null;
                        
                        tx.abort();
                        throw new AMQStoreException("Content chunk offset" + mck.getOffset() + " not found for message " + messageId);
                    }
                    
                    if (_log.isDebugEnabled())
                    {
                        _log.debug("Deleted content chunk offset " + mck.getOffset() + " for message " + messageId);
                    }
                }
                
                status = cursor.getNext(contentKeyEntry, value, LockMode.RMW);
            }

            cursor.close();
            cursor = null;
            
            commit(tx, true);
        }
        catch (DatabaseException e)
        {
            e.printStackTrace();
            
            if (tx != null)
            {
                try
                {
                    if(cursor != null)
                    {
                        cursor.close();
                        cursor = null;
                    }
                    
                    tx.abort();
                }
                catch (DatabaseException e1)
                {
                    throw new AMQStoreException("Error aborting transaction " + e1, e1);
                }
            }

            throw new AMQStoreException("Error removing message with id " + messageId + " from database: " + e.getMessage(), e);
        }
        finally
        {
            if(cursor != null)
            {
                try
                {
                    cursor.close();
                }
                catch (DatabaseException e)
                {
                    throw new AMQStoreException("Error closing database connection: " + e.getMessage(), e);
                }
            }
        }
    }

    /**
     * @see DurableConfigurationStore#createExchange(Exchange)
     */
    public void createExchange(Exchange exchange) throws AMQStoreException
    {
        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.getNameShortString(), key);

            DatabaseEntry value = new DatabaseEntry();
            TupleBinding exchangeBinding = new ExchangeTB();
            exchangeBinding.objectToEntry(exchangeRec, value);

            try
            {
                _exchangeDb.put(null, key, value);
            }
            catch (DatabaseException e)
            {
                throw new AMQStoreException("Error writing Exchange with name " + exchange.getName() + " to database: " + e.getMessage(), e);
            }
        }
    }

    /**
     * @see DurableConfigurationStore#removeExchange(Exchange)
     */
    public void removeExchange(Exchange exchange) throws AMQStoreException
    {
        DatabaseEntry key = new DatabaseEntry();
        EntryBinding keyBinding = new AMQShortStringTB();
        keyBinding.objectToEntry(exchange.getNameShortString(), key);
        try
        {
            OperationStatus status = _exchangeDb.delete(null, key);
            if (status == OperationStatus.NOTFOUND)
            {
                throw new AMQStoreException("Exchange " + exchange.getName() + " not found");
            }
        }
        catch (DatabaseException e)
        {
            throw new AMQStoreException("Error writing deleting with name " + exchange.getName() + " from database: " + e.getMessage(), e);
        }
    }




    /**
     * @see DurableConfigurationStore#bindQueue(Exchange, AMQShortString, AMQQueue, FieldTable)
     */
    public void bindQueue(Exchange exchange, AMQShortString routingKey, AMQQueue queue, FieldTable args) throws AMQStoreException
    {
        // _log.debug("public void bindQueue(Exchange exchange = " + exchange + ", AMQShortString routingKey = " + routingKey
        // + ", AMQQueue queue = " + queue + ", FieldTable args = " + args + "): called");

        if (_state != State.RECOVERING)
        {
            BindingKey bindingRecord = new BindingKey(exchange.getNameShortString(), 
                                                queue.getNameShortString(), routingKey, args);

            DatabaseEntry key = new DatabaseEntry();
            EntryBinding keyBinding = _bindingTupleBindingFactory.getInstance();
            
            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);
            }
            catch (DatabaseException e)
            {
                throw new AMQStoreException("Error writing binding for AMQQueue with name " + queue.getName() + " to exchange "
                                       + exchange.getName() + " to database: " + e.getMessage(), e);
            }
        }
    }

    /**
     * @see DurableConfigurationStore#unbindQueue(Exchange, AMQShortString, AMQQueue, FieldTable)
     */
    public void unbindQueue(Exchange exchange, AMQShortString routingKey, AMQQueue queue, FieldTable args)
            throws AMQStoreException
    {
        DatabaseEntry key = new DatabaseEntry();
        EntryBinding keyBinding = _bindingTupleBindingFactory.getInstance();
        keyBinding.objectToEntry(new BindingKey(exchange.getNameShortString(), queue.getNameShortString(), routingKey, args), key);

        try
        {
            OperationStatus status = _queueBindingsDb.delete(null, key);
            if (status == OperationStatus.NOTFOUND)
            {
                throw new AMQStoreException("Queue binding for queue with name " + queue.getName() + " to exchange "
                                       + exchange.getName() + "  not found");
            }
        }
        catch (DatabaseException e)
        {
            throw new AMQStoreException("Error deleting queue binding for queue with name " + queue.getName() + " to exchange "
                                   + exchange.getName() + " from database: " + e.getMessage(), e);
        }
    }

    /**
     * @see DurableConfigurationStore#createQueue(AMQQueue)
     */
    public void createQueue(AMQQueue queue) throws AMQStoreException
    {
        createQueue(queue, null);
    }

    /**
     * @see DurableConfigurationStore#createQueue(AMQQueue, FieldTable)
     */
    public void createQueue(AMQQueue queue, FieldTable arguments) throws AMQStoreException
    {
        if (_log.isDebugEnabled())
        {
            _log.debug("public void createQueue(AMQQueue queue(" + queue.getName() + ") = " + queue + "): called");
        }
        
        QueueRecord queueRecord= new QueueRecord(queue.getNameShortString(), 
                                                queue.getOwner(), queue.isExclusive(), arguments);
        
        createQueue(queueRecord);
    }

    /**
     * Makes the specified queue persistent. 
     * 
     * Only intended for direct use during store upgrades.
     *
     * @param queueRecord     Details of the queue to store.
     *
     * @throws AMQStoreException If the operation fails for any reason.
     */
    protected void createQueue(QueueRecord queueRecord) throws AMQStoreException
    {
        if (_state != State.RECOVERING)
        {
            DatabaseEntry key = new DatabaseEntry();
            EntryBinding keyBinding = new AMQShortStringTB();
            keyBinding.objectToEntry(queueRecord.getNameShortString(), key);

            DatabaseEntry value = new DatabaseEntry();
            TupleBinding queueBinding = _queueTupleBindingFactory.getInstance();

            queueBinding.objectToEntry(queueRecord, value);
            try
            {
                _queueDb.put(null, key, value);
            }
            catch (DatabaseException e)
            {
                throw new AMQStoreException("Error writing AMQQueue with name " + queueRecord.getNameShortString().asString()
                        + " to database: " + e.getMessage(), e);
            }
        }
    }

    /**
     * Updates the specified queue in the persistent store, IF it is already present. If the queue
     * is not present in the store, it will not be added.
     * 
     * NOTE: Currently only updates the exclusivity.
     *
     * @param queue The queue to update the entry for.
     * @throws AMQStoreException If the operation fails for any reason.
     */
    public void updateQueue(final AMQQueue queue) throws AMQStoreException
    {
        if (_log.isDebugEnabled())
        {
            _log.debug("Updating queue: " + queue.getName());
        }

        try
        {
            DatabaseEntry key = new DatabaseEntry();
            EntryBinding keyBinding = new AMQShortStringTB();
            keyBinding.objectToEntry(queue.getNameShortString(), key);

            DatabaseEntry value = new DatabaseEntry();
            DatabaseEntry newValue = new DatabaseEntry();
            TupleBinding queueBinding = _queueTupleBindingFactory.getInstance();

            OperationStatus status = _queueDb.get(null, key, value, LockMode.DEFAULT);
            if(status == OperationStatus.SUCCESS)
            {
                //read the existing record and apply the new exclusivity setting 
                QueueRecord queueRecord = (QueueRecord) queueBinding.entryToObject(value);
                queueRecord.setExclusive(queue.isExclusive());
                
                //write the updated entry to the store
                queueBinding.objectToEntry(queueRecord, newValue);
                
                _queueDb.put(null, key, newValue);
            }
            else if(status != OperationStatus.NOTFOUND)
            {
                throw new AMQStoreException("Error updating queue details within the store: " + status);
            }
        }
        catch (DatabaseException e)
        {
            throw new AMQStoreException("Error updating queue details within the store: " + e,e);
        }
    }

    /**
     * Removes the specified queue from the persistent store.
     *
     * @param queue The queue to remove.
     *
     * @throws AMQStoreException If the operation fails for any reason.
     */
    public void removeQueue(final AMQQueue queue) throws AMQStoreException
    {
        AMQShortString name = queue.getNameShortString();

        if (_log.isDebugEnabled())
        {
            _log.debug("public void removeQueue(AMQShortString name = " + name + "): called");
        }
            
        DatabaseEntry key = new DatabaseEntry();
        EntryBinding keyBinding = new AMQShortStringTB();
        keyBinding.objectToEntry(name, key);
        try
        {
            OperationStatus status = _queueDb.delete(null, key);
            if (status == OperationStatus.NOTFOUND)
            {
                throw new AMQStoreException("Queue " + name + " not found");
            }
        }
        catch (DatabaseException e)
        {
            throw new AMQStoreException("Error writing deleting with name " + name + " from database: " + e.getMessage(), e);
        }
    }

    /**
     * Places a message onto a specified queue, in a given transaction.
     *
     * @param tx   The transaction for the operation.
     * @param queue     The the queue to place the message on.
     * @param messageId The message to enqueue.
     *
     * @throws AMQStoreException If the operation fails for any reason.
     */
    public void enqueueMessage(final com.sleepycat.je.Transaction tx, final TransactionLogResource queue, Long messageId) throws AMQStoreException
    {
        // _log.debug("public void enqueueMessage(Transaction tx = " + tx + ", AMQShortString name = " + name + ", Long messageId): called");

        AMQShortString name = new AMQShortString(queue.getResourceName());
        
        DatabaseEntry key = new DatabaseEntry();
        EntryBinding keyBinding = new QueueEntryTB();
        QueueEntryKey dd = new QueueEntryKey(name, messageId);
        keyBinding.objectToEntry(dd, key);
        DatabaseEntry value = new DatabaseEntry();
        ByteBinding.byteToEntry((byte) 0, value);

        try
        {
            if (_log.isDebugEnabled())
            {
                _log.debug("Enqueuing message " + messageId + " on queue " + name + " [Transaction" + tx + "]");
            }
            _deliveryDb.put(tx, key, value);
        }
        catch (DatabaseException e)
        {
            _log.error("Failed to enqueue: " + e.getMessage(), e);
            throw new AMQStoreException("Error writing enqueued message with id " + messageId + " for queue " + name
                                   + " to database", e);
        }
    }

    /**
     * Extracts a message from a specified queue, in a given transaction.
     *
     * @param tx   The transaction for the operation.
     * @param queue     The name queue to take the message from.
     * @param messageId The message to dequeue.
     *
     * @throws AMQStoreException If the operation fails for any reason, or if the specified message does not exist.
     */
    public void dequeueMessage(final com.sleepycat.je.Transaction tx, final TransactionLogResource queue, Long messageId) throws AMQStoreException
    {
        AMQShortString name = new AMQShortString(queue.getResourceName());

        DatabaseEntry key = new DatabaseEntry();
        EntryBinding keyBinding = new QueueEntryTB();
        QueueEntryKey dd = new QueueEntryKey(name, messageId);

        keyBinding.objectToEntry(dd, key);

        if (_log.isDebugEnabled())
        {
            _log.debug("Dequeue message id " + messageId);
        }
        
        try
        {

            OperationStatus status = _deliveryDb.delete(tx, key);
            if (status == OperationStatus.NOTFOUND)
            {
                throw new AMQStoreException("Unable to find message with id " + messageId + " on queue " + name);
            } 
            else if (status != OperationStatus.SUCCESS)
            {
                throw new AMQStoreException("Unable to remove message with id " + messageId + " on queue " + name);
            }

            if (_log.isDebugEnabled())
            {
                _log.debug("Removed message " + messageId + ", " + name + " from delivery db");

            }
        }
        catch (DatabaseException e)
        {

            _log.error("Failed to dequeue message " + messageId + ": " + e.getMessage(), e);
            _log.error(tx);

            throw new AMQStoreException("Error accessing database while dequeuing message: " + e.getMessage(), e);
        }
    }

    /**
     * Commits all operations performed within a given transaction.
     *
     * @param tx The transaction to commit all operations for.
     *
     * @throws AMQStoreException If the operation fails for any reason.
     */
    private StoreFuture commitTranImpl(final com.sleepycat.je.Transaction tx, boolean syncCommit) throws AMQStoreException
    {
        //if (_log.isDebugEnabled())
        //{
        //    _log.debug("public void commitTranImpl() called with (Transaction=" + tx + ", syncCommit= "+ syncCommit + ")");
        //}
        
        if (tx == null)
        {
            throw new AMQStoreException("Fatal internal error: transactional is null at commitTran");
        }
        
        StoreFuture result;
        try
        {
            result = commit(tx, syncCommit);

            if (_log.isDebugEnabled())
            {
                _log.debug("commitTranImpl completed for [Transaction:" + tx + "]");
            }
        }
        catch (DatabaseException e)
        {
            throw new AMQStoreException("Error commit tx: " + e.getMessage(), e);
        }
        
        return result;
    }

    /**
     * Abandons all operations performed within a given transaction.
     *
     * @param tx The transaction to abandon.
     *
     * @throws AMQStoreException If the operation fails for any reason.
     */
    public void abortTran(final com.sleepycat.je.Transaction tx) throws AMQStoreException
    {
        if (_log.isDebugEnabled())
        {
            _log.debug("abortTran called for [Transaction:" + tx + "]");
        }

        try
        {
            tx.abort();
        }
        catch (DatabaseException e)
        {
            throw new AMQStoreException("Error aborting transaction: " + e.getMessage(), e);
        }
    }

    /**
     * Primarily for testing purposes.
     *
     * @param queueName
     *
     * @return a list of message ids for messages enqueued for a particular queue
     */
    List<Long> getEnqueuedMessages(AMQShortString queueName) throws AMQStoreException
    {
        Cursor cursor = null;
        try
        {
            cursor = _deliveryDb.openCursor(null, null);

            DatabaseEntry key = new DatabaseEntry();

            QueueEntryKey dd = new QueueEntryKey(queueName, 0);

            EntryBinding keyBinding = new QueueEntryTB();
            keyBinding.objectToEntry(dd, key);

            DatabaseEntry value = new DatabaseEntry();

            LinkedList<Long> messageIds = new LinkedList<Long>();

            OperationStatus status = cursor.getSearchKeyRange(key, value, LockMode.DEFAULT);
            dd = (QueueEntryKey) keyBinding.entryToObject(key);

            while ((status == OperationStatus.SUCCESS) && dd.getQueueName().equals(queueName))
            {

                messageIds.add(dd.getMessageId());
                status = cursor.getNext(key, value, LockMode.DEFAULT);
                if (status == OperationStatus.SUCCESS)
                {
                    dd = (QueueEntryKey) keyBinding.entryToObject(key);
                }
            }

            return messageIds;
        }
        catch (DatabaseException e)
        {
            throw new AMQStoreException("Database error: " + e.getMessage(), e);
        }
        finally
        {
            if (cursor != null)
            {
                try
                {
                    cursor.close();
                }
                catch (DatabaseException e)
                {
                    throw new AMQStoreException("Error closing cursor: " + e.getMessage(), e);
                }
            }
        }
    }

    /**
     * Return a valid, currently unused message id.
     *
     * @return A fresh message id.
     */
    public Long getNewMessageId()
    {
        return _messageId.incrementAndGet();
    }

    /**
     * Stores a chunk of message data.
     *
     * @param tx         The transaction for the operation.
     * @param messageId       The message to store the data for.
     * @param offset          The offset of the data chunk in the message.
     * @param contentBody     The content of the data chunk.
     *
     * @throws AMQStoreException If the operation fails for any reason, or if the specified message does not exist.
     */
    protected void addContent(final com.sleepycat.je.Transaction tx, Long messageId, int offset, 
                                      ByteBuffer contentBody) throws AMQStoreException
    {
        DatabaseEntry key = new DatabaseEntry();
        TupleBinding<MessageContentKey> keyBinding = new MessageContentKeyTB_5();
        keyBinding.objectToEntry(new MessageContentKey_5(messageId, offset), key);
        DatabaseEntry value = new DatabaseEntry();
        TupleBinding<ByteBuffer> messageBinding = new ContentTB();
        messageBinding.objectToEntry(contentBody, value);
        try
        {
            OperationStatus status = _messageContentDb.put(tx, key, value);
            if (status != OperationStatus.SUCCESS)
            {
                throw new AMQStoreException("Error adding content chunk offset" + offset + " for message id " + messageId + ": "
                                       + status);
            }

            if (_log.isDebugEnabled())
            {
                _log.debug("Storing content chunk offset" + offset + " for message " + messageId + "[Transaction" + tx + "]");
            }
        }
        catch (DatabaseException e)
        {
            throw new AMQStoreException("Error writing AMQMessage with id " + messageId + " to database: " + e.getMessage(), e);
        }
    }

    /**
     * Stores message meta-data.
     *
     * @param tx         The transaction for the operation.
     * @param messageId       The message to store the data for.
     * @param messageMetaData The message meta data to store.
     *
     * @throws AMQStoreException If the operation fails for any reason, or if the specified message does not exist.
     */
    private void storeMetaData(final com.sleepycat.je.Transaction tx, Long messageId, StorableMessageMetaData messageMetaData)
            throws AMQStoreException
    {
        if (_log.isDebugEnabled())
        {
            _log.debug("public void storeMetaData(Txn tx = " + tx + ", Long messageId = "
                       + messageId + ", MessageMetaData messageMetaData = " + messageMetaData + "): called");
        }

        DatabaseEntry key = new DatabaseEntry();
        EntryBinding keyBinding = TupleBinding.getPrimitiveBinding(Long.class);
        keyBinding.objectToEntry(messageId, key);
        DatabaseEntry value = new DatabaseEntry();
        
        TupleBinding messageBinding = _metaDataTupleBindingFactory.getInstance();
        messageBinding.objectToEntry(messageMetaData, value);
        try
        {
            _messageMetaDataDb.put(tx, key, value);
            if (_log.isDebugEnabled())
            {
                _log.debug("Storing message metadata for message id " + messageId + "[Transaction" + tx + "]");
            }
        }
        catch (DatabaseException e)
        {
            throw new AMQStoreException("Error writing message metadata with id " + messageId + " to database: " + e.getMessage(), e);
        }
    }

    /**
     * Retrieves message meta-data.
     *
     * @param messageId The message to get the meta-data for.
     *
     * @return The message meta data.
     *
     * @throws AMQStoreException If the operation fails for any reason, or if the specified message does not exist.
     */
    public StorableMessageMetaData getMessageMetaData(Long messageId) throws AMQStoreException
    {
        if (_log.isDebugEnabled())
        {
            _log.debug("public MessageMetaData getMessageMetaData(Long messageId = "
                       + messageId + "): called");
        }

        DatabaseEntry key = new DatabaseEntry();
        EntryBinding keyBinding = TupleBinding.getPrimitiveBinding(Long.class);
        keyBinding.objectToEntry(messageId, key);
        DatabaseEntry value = new DatabaseEntry();
        TupleBinding messageBinding = _metaDataTupleBindingFactory.getInstance();

        try
        {
            OperationStatus status = _messageMetaDataDb.get(null, key, value, LockMode.READ_UNCOMMITTED);
            if (status != OperationStatus.SUCCESS)
            {
                throw new AMQStoreException("Metadata not found for message with id " + messageId);
            }

            StorableMessageMetaData mdd = (StorableMessageMetaData) messageBinding.entryToObject(value);

            return mdd;
        }
        catch (DatabaseException e)
        {
            throw new AMQStoreException("Error reading message metadata for message with id " + messageId + ": " + e.getMessage(), e);
        }
    }

    /**
     * Fills the provided ByteBuffer with as much content for the specified message as possible, starting
     * from the specified offset in 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 The number of bytes inserted into the destination
     *
     * @throws AMQStoreException If the operation fails for any reason, or if the specified message does not exist.
     */
    public int getContent(Long messageId, int offset, ByteBuffer dst) throws AMQStoreException
    {    
        DatabaseEntry contentKeyEntry = new DatabaseEntry();
        
        //Start from 0 offset and search for the starting chunk. 
        MessageContentKey_5 mck = new MessageContentKey_5(messageId, 0);
        TupleBinding<MessageContentKey> contentKeyTupleBinding = new MessageContentKeyTB_5();
        contentKeyTupleBinding.objectToEntry(mck, contentKeyEntry);
        DatabaseEntry value = new DatabaseEntry();
        TupleBinding<ByteBuffer> contentTupleBinding = new ContentTB();
        
        if (_log.isDebugEnabled())
        {
            _log.debug("Message Id: " + messageId + " Getting content body from offset: " + offset);
        }

        int written = 0;
        int seenSoFar = 0;
        
        Cursor cursor = null;
        try
        {
            cursor = _messageContentDb.openCursor(null, null);
            
            OperationStatus status = cursor.getSearchKeyRange(contentKeyEntry, value, LockMode.READ_UNCOMMITTED);

            while (status == OperationStatus.SUCCESS)
            {
                mck = (MessageContentKey_5) 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;
                
                if(seenSoFar >= offset)
                {
                    byte[] dataAsBytes = buf.array();

                    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);
            }

            return written;
        }
        catch (DatabaseException e)
        {
            throw new AMQStoreException("Error writing AMQMessage with id " + messageId + " to database: " + e.getMessage(), e);
        }
        finally
        {
            if(cursor != null)
            {
                try
                {
                    cursor.close();
                }
                catch (DatabaseException e)
                {
		            throw new AMQStoreException("Error writing AMQMessage with id " + messageId + " to database: " + e.getMessage(), e);
                }
            }
        }
    }

    public boolean isPersistent()
    {
        return true;
    }

    public <T extends StorableMessageMetaData> StoredMessage<T> addMessage(T metaData)
    {
        if(metaData.isPersistent())
        {
            return new StoredBDBMessage(getNewMessageId(), metaData);
        }
        else
        {
            return new StoredMemoryMessage(getNewMessageId(), metaData);
        }
    }


    //protected getters for the TupleBindingFactories

    protected QueueTupleBindingFactory getQueueTupleBindingFactory()
    {
        return _queueTupleBindingFactory;
    }

    protected BindingTupleBindingFactory getBindingTupleBindingFactory()
    {
        return _bindingTupleBindingFactory;
    }
    
    protected MessageMetaDataTupleBindingFactory getMetaDataTupleBindingFactory()
    {
        return _metaDataTupleBindingFactory;
    }

    //Package getters for the various databases used by the Store

    Database getMetaDataDb()
    {
        return _messageMetaDataDb;
    }

    Database getContentDb()
    {
        return _messageContentDb;
    }

    Database getQueuesDb()
    {
        return _queueDb;
    }

    Database getDeliveryDb()
    {
        return _deliveryDb;
    }

    Database getExchangesDb()
    {
        return _exchangeDb;
    }

    Database getBindingsDb()
    {
        return _queueBindingsDb;
    }

    void visitMetaDataDb(DatabaseVisitor visitor) throws DatabaseException, AMQStoreException
    {
        visitDatabase(_messageMetaDataDb, visitor);
    }

    void visitContentDb(DatabaseVisitor visitor) throws DatabaseException, AMQStoreException
    {
        visitDatabase(_messageContentDb, visitor);
    }

    void visitQueues(DatabaseVisitor visitor) throws DatabaseException, AMQStoreException
    {
        visitDatabase(_queueDb, visitor);
    }

    void visitDelivery(DatabaseVisitor visitor) throws DatabaseException, AMQStoreException
    {
        visitDatabase(_deliveryDb, visitor);
    }

    void visitExchanges(DatabaseVisitor visitor) throws DatabaseException, AMQStoreException
    {
        visitDatabase(_exchangeDb, visitor);
    }

    void visitBindings(DatabaseVisitor visitor) throws DatabaseException, AMQStoreException
    {
        visitDatabase(_queueBindingsDb, visitor);
    }

    /**
     * Generic visitDatabase allows iteration through the specified database.
     *
     * @param database The database to visit
     * @param visitor  The visitor to give each entry to.
     *
     * @throws DatabaseException If there is a problem with the Database structure
     * @throws AMQStoreException If there is a problem with the Database contents
     */
    void visitDatabase(Database database, DatabaseVisitor visitor) throws DatabaseException, AMQStoreException
    {
        Cursor cursor = database.openCursor(null, null);

        try
        {
            DatabaseEntry key = new DatabaseEntry();
            DatabaseEntry value = new DatabaseEntry();
            while (cursor.getNext(key, value, LockMode.RMW) == OperationStatus.SUCCESS)
            {
                visitor.visit(key, value);
            }
        }
        finally
        {
            if (cursor != null)
            {
                cursor.close();
            }
        }
    }

    private StoreFuture commit(com.sleepycat.je.Transaction tx, boolean syncCommit) throws DatabaseException
    {
        // _log.debug("void commit(Transaction tx = " + tx + ", sync = " + syncCommit + "): called");

        tx.commitNoSync();

        BDBCommitFuture commitFuture = new BDBCommitFuture(_commitThread, tx, syncCommit);
        commitFuture.commit();
        
        return commitFuture;
    }

    public void startCommitThread()
    {
        _commitThread.start();
    }

    private static final class BDBCommitFuture implements StoreFuture
    {
        // private static final Logger _log = Logger.getLogger(BDBCommitFuture.class);

        private final CommitThread _commitThread;
        private final com.sleepycat.je.Transaction _tx;
        private DatabaseException _databaseException;
        private boolean _complete;
        private boolean _syncCommit;

        public BDBCommitFuture(CommitThread commitThread, com.sleepycat.je.Transaction tx, boolean syncCommit)
        {
            // _log.debug("public Commit(CommitThread commitThread = " + commitThread + ", Transaction tx = " + tx
            // + "): called");

            _commitThread = commitThread;
            _tx = tx;
            _syncCommit = syncCommit;
        }

        public synchronized void complete()
        {
            if (_log.isDebugEnabled())
            {
                _log.debug("public synchronized void complete(): called (Transaction = " + _tx + ")");
            }

            _complete = true;

            notifyAll();
        }

        public synchronized void abort(DatabaseException databaseException)
        {
            // _log.debug("public synchronized void abort(DatabaseException databaseException = " + databaseException
            // + "): called");

            _complete = true;
            _databaseException = databaseException;

            notifyAll();
        }

        public void commit() throws DatabaseException
        {
            //_log.debug("public void commit(): called");

            _commitThread.addJob(this);
            
            if(!_syncCommit)
            {
                _log.debug("CommitAsync was requested, returning immediately.");
                return;
            }
            
            synchronized (BDBCommitFuture.this)
            {
                while (!_complete)
                {
                    try
                    {
                        wait(250);
                    }
                    catch (InterruptedException e)
                    {
                        // _log.error("Unexpected thread interruption: " + e, e);
                        throw new RuntimeException(e);
                    }
                }

                // _log.debug("Commit completed, _databaseException = " + _databaseException);

                if (_databaseException != null)
                {
                    throw _databaseException;
                }
            }
        }

        public synchronized boolean isComplete()
        {
            return _complete;
        }

        public void waitForCompletion()
        {
            while (!isComplete())
            {
                try
                {
                    wait(250);
                }
                catch (InterruptedException e)
                {
                    //TODO Should we ignore, or throw a 'StoreException'?
                    throw new RuntimeException(e);
                }
            }
        }
    }

    /**
     * Implements a thread which batches and commits a queue of {@link BDBCommitFuture} operations. The commit operations
     * themselves are responsible for adding themselves to the queue and waiting for the commit to happen before
     * continuing, but it is the responsibility of this thread to tell the commit operations when they have been
     * completed by calling back on their {@link BDBCommitFuture#complete()} and {@link BDBCommitFuture#abort} methods.
     *
     * <p/><table id="crc"><caption>CRC Card</caption> <tr><th> Responsibilities <th> Collarations </table>
     */
    private class CommitThread extends Thread
    {
        // private final Logger _log = Logger.getLogger(CommitThread.class);

        private final AtomicBoolean _stopped = new AtomicBoolean(false);
        private final AtomicReference<Queue<BDBCommitFuture>> _jobQueue = new AtomicReference<Queue<BDBCommitFuture>>(new ConcurrentLinkedQueue<BDBCommitFuture>());
        private final CheckpointConfig _config = new CheckpointConfig();
        private final Object _lock = new Object();

        public CommitThread(String name)
        {
            super(name);
            _config.setForce(true);

        }

        public void run()
        {
            while (!_stopped.get())
            {
                synchronized (_lock)
                {
                    while (!_stopped.get() && !hasJobs())
                    {
                        try
                        {
                            // RHM-7 Periodically wake up and check, just in case we 
                            // missed a notification. Don't want to lock the broker hard.
                            _lock.wait(250);
                        }
                        catch (InterruptedException e)
                        {
                            // _log.info(getName() + " interrupted. ");
                        }
                    }
                }
                processJobs();
            }
        }

        private void processJobs()
        {
            // _log.debug("private void processJobs(): called");

            // we replace the old queue atomically with a new one and this avoids any need to
            // copy elements out of the queue
            Queue<BDBCommitFuture> jobs = _jobQueue.getAndSet(new ConcurrentLinkedQueue<BDBCommitFuture>());

            try
            {
                // _environment.checkpoint(_config);
                _environment.sync();

                for (BDBCommitFuture commit : jobs)
                {
                    commit.complete();
                }
            }
            catch (DatabaseException e)
            {
                for (BDBCommitFuture commit : jobs)
                {
                    commit.abort(e);
                }
            }

        }

        private boolean hasJobs()
        {
            return !_jobQueue.get().isEmpty();
        }

        public void addJob(BDBCommitFuture commit)
        {
            synchronized (_lock)
            {
                _jobQueue.get().add(commit);
                _lock.notifyAll();
            }
        }

        public void close()
        {
            synchronized (_lock)
            {
                _stopped.set(true);
                _lock.notifyAll();
            }
        }
    }
    
    
    private class StoredBDBMessage implements StoredMessage
    {

        private final long _messageId;
        private volatile SoftReference<StorableMessageMetaData> _metaDataRef;
        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)
                {
                    _txn = _environment.beginTransaction(null, null);
                    storeMetaData(_txn, messageId, metaData);
                }
            }
            catch (DatabaseException e)
            {
                throw new RuntimeException(e);
            }
            catch (AMQStoreException e)
            {
                throw new RuntimeException(e);
            }

        }

        public StorableMessageMetaData getMetaData()
        {
            StorableMessageMetaData metaData = _metaDataRef.get();
            if(metaData == null)
            {
                try
                {
                    metaData = BDBMessageStore.this.getMessageMetaData(_messageId);
                }
                catch (AMQStoreException e)
                {
                    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(_txn, _messageId, offsetInMessage, src);
            }
            catch (AMQStoreException e)
            {
                throw new RuntimeException(e);
            }
        }

        public int getContent(int offsetInMessage, java.nio.ByteBuffer dst)
        {
            try
            {
                return BDBMessageStore.this.getContent(_messageId, offsetInMessage, dst);
            }
            catch (AMQStoreException e)
            {
                throw new RuntimeException(e);
            }
        }

        public StoreFuture flushToStore()
        {
            try
            {
                if(_txn != null)
                {
                    //if(_log.isDebugEnabled())
                    //{
                    //   _log.debug("Flushing message " + _messageId + " to store");
                    //}
                    BDBMessageStore.this.commitTranImpl(_txn, true);
                }
            }
            catch (AMQStoreException e)
            {
                throw new RuntimeException(e);
            }
            finally
            {
                _txn = null;
            }
            return IMMEDIATE_FUTURE;
        }

        public void remove()
        {
            flushToStore();
            try
            {
                BDBMessageStore.this.removeMessage(_messageId);
            }
            catch (AMQStoreException e)
            {
                throw new RuntimeException(e);
            }
        }
    }

    private class BDBTransaction implements Transaction
    {
        private com.sleepycat.je.Transaction _txn;

        private BDBTransaction()
        {
            try
            {
                _txn = _environment.beginTransaction(null, null);
            }
            catch (DatabaseException e)
            {
                throw new RuntimeException(e);
            }
        }

        public void enqueueMessage(TransactionLogResource queue, Long messageId) throws AMQStoreException
        {
            BDBMessageStore.this.enqueueMessage(_txn, queue, messageId);
        }

        public void dequeueMessage(TransactionLogResource queue, Long messageId) throws AMQStoreException
        {
            BDBMessageStore.this.dequeueMessage(_txn, queue, messageId);

        }

        public void commitTran() throws AMQStoreException
        {
            BDBMessageStore.this.commitTranImpl(_txn, true);
        }

        public StoreFuture commitTranAsync() throws AMQStoreException
        {
            return BDBMessageStore.this.commitTranImpl(_txn, false);
        }

        public void abortTran() throws AMQStoreException
        {
            BDBMessageStore.this.abortTran(_txn);
        }
    }

}