/**
 * 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.activemq.store.kahadb;

import org.apache.activemq.broker.BrokerService;
import org.apache.activemq.broker.ConnectionContext;
import org.apache.activemq.broker.LockableServiceSupport;
import org.apache.activemq.broker.Locker;
import org.apache.activemq.broker.jmx.AnnotatedMBean;
import org.apache.activemq.broker.jmx.PersistenceAdapterView;
import org.apache.activemq.broker.scheduler.JobSchedulerStore;
import org.apache.activemq.command.ActiveMQDestination;
import org.apache.activemq.command.ActiveMQQueue;
import org.apache.activemq.command.ActiveMQTopic;
import org.apache.activemq.command.LocalTransactionId;
import org.apache.activemq.command.ProducerId;
import org.apache.activemq.command.TransactionId;
import org.apache.activemq.command.XATransactionId;
import org.apache.activemq.protobuf.Buffer;
import org.apache.activemq.store.JournaledStore;
import org.apache.activemq.store.MessageStore;
import org.apache.activemq.store.NoLocalSubscriptionAware;
import org.apache.activemq.store.PersistenceAdapter;
import org.apache.activemq.store.PersistenceAdapterStatistics;
import org.apache.activemq.store.SharedFileLocker;
import org.apache.activemq.store.TopicMessageStore;
import org.apache.activemq.store.TransactionIdTransformer;
import org.apache.activemq.store.TransactionIdTransformerAware;
import org.apache.activemq.store.TransactionStore;
import org.apache.activemq.store.kahadb.data.KahaLocalTransactionId;
import org.apache.activemq.store.kahadb.data.KahaTransactionInfo;
import org.apache.activemq.store.kahadb.data.KahaXATransactionId;
import org.apache.activemq.store.kahadb.disk.journal.DataFileFactory;
import org.apache.activemq.store.kahadb.disk.journal.Journal.JournalDiskSyncStrategy;
import org.apache.activemq.usage.SystemUsage;
import org.apache.activemq.util.ServiceStopper;

import javax.management.ObjectName;
import java.io.File;
import java.io.IOException;
import java.util.Set;
import java.util.concurrent.Callable;

import static org.apache.activemq.broker.jmx.BrokerMBeanSupport.createPersistenceAdapterName;

/**
 * An implementation of {@link PersistenceAdapter} designed for use with
 * KahaDB - Embedded Lightweight Non-Relational Database
 *
 * @org.apache.xbean.XBean element="kahaDB"
 *
 */
public class KahaDBPersistenceAdapter extends LockableServiceSupport implements PersistenceAdapter,
    JournaledStore, TransactionIdTransformerAware, NoLocalSubscriptionAware {

    private final KahaDBStore letter = createStore();

    protected KahaDBStore createStore() {
        return new KahaDBStore();
    }

    /**
     * @param context
     * @throws IOException
     * @see org.apache.activemq.store.PersistenceAdapter#beginTransaction(org.apache.activemq.broker.ConnectionContext)
     */
    @Override
    public void beginTransaction(ConnectionContext context) throws IOException {
        this.letter.beginTransaction(context);
    }

    /**
     * @param cleanup
     * @throws IOException
     * @see org.apache.activemq.store.PersistenceAdapter#checkpoint(boolean)
     */
    @Override
    public void checkpoint(boolean cleanup) throws IOException {
        this.letter.checkpoint(cleanup);
    }

    /**
     * @param context
     * @throws IOException
     * @see org.apache.activemq.store.PersistenceAdapter#commitTransaction(org.apache.activemq.broker.ConnectionContext)
     */
    @Override
    public void commitTransaction(ConnectionContext context) throws IOException {
        this.letter.commitTransaction(context);
    }

    /**
     * @param destination
     * @return MessageStore
     * @throws IOException
     * @see org.apache.activemq.store.PersistenceAdapter#createQueueMessageStore(org.apache.activemq.command.ActiveMQQueue)
     */
    @Override
    public MessageStore createQueueMessageStore(ActiveMQQueue destination) throws IOException {
        return this.letter.createQueueMessageStore(destination);
    }

    /**
     * @param destination
     * @return TopicMessageStore
     * @throws IOException
     * @see org.apache.activemq.store.PersistenceAdapter#createTopicMessageStore(org.apache.activemq.command.ActiveMQTopic)
     */
    @Override
    public TopicMessageStore createTopicMessageStore(ActiveMQTopic destination) throws IOException {
        return this.letter.createTopicMessageStore(destination);
    }

    /**
     * @return TransactionStore
     * @throws IOException
     * @see org.apache.activemq.store.PersistenceAdapter#createTransactionStore()
     */
    @Override
    public TransactionStore createTransactionStore() throws IOException {
        return this.letter.createTransactionStore();
    }

    /**
     * @throws IOException
     * @see org.apache.activemq.store.PersistenceAdapter#deleteAllMessages()
     */
    @Override
    public void deleteAllMessages() throws IOException {
        this.letter.deleteAllMessages();
    }

    /**
     * @return destinations
     * @see org.apache.activemq.store.PersistenceAdapter#getDestinations()
     */
    @Override
    public Set<ActiveMQDestination> getDestinations() {
        return this.letter.getDestinations();
    }

    /**
     * @return lastMessageBrokerSequenceId
     * @throws IOException
     * @see org.apache.activemq.store.PersistenceAdapter#getLastMessageBrokerSequenceId()
     */
    @Override
    public long getLastMessageBrokerSequenceId() throws IOException {
        return this.letter.getLastMessageBrokerSequenceId();
    }

    @Override
    public long getLastProducerSequenceId(ProducerId id) throws IOException {
        return this.letter.getLastProducerSequenceId(id);
    }

    @Override
    public void allowIOResumption() {
        this.letter.allowIOResumption();
    }

    /**
     * @param destination
     * @see org.apache.activemq.store.PersistenceAdapter#removeQueueMessageStore(org.apache.activemq.command.ActiveMQQueue)
     */
    @Override
    public void removeQueueMessageStore(ActiveMQQueue destination) {
        this.letter.removeQueueMessageStore(destination);
    }

    /**
     * @param destination
     * @see org.apache.activemq.store.PersistenceAdapter#removeTopicMessageStore(org.apache.activemq.command.ActiveMQTopic)
     */
    @Override
    public void removeTopicMessageStore(ActiveMQTopic destination) {
        this.letter.removeTopicMessageStore(destination);
    }

    /**
     * @param context
     * @throws IOException
     * @see org.apache.activemq.store.PersistenceAdapter#rollbackTransaction(org.apache.activemq.broker.ConnectionContext)
     */
    @Override
    public void rollbackTransaction(ConnectionContext context) throws IOException {
        this.letter.rollbackTransaction(context);
    }

    /**
     * @param brokerName
     * @see org.apache.activemq.store.PersistenceAdapter#setBrokerName(java.lang.String)
     */
    @Override
    public void setBrokerName(String brokerName) {
        this.letter.setBrokerName(brokerName);
    }

    /**
     * @param usageManager
     * @see org.apache.activemq.store.PersistenceAdapter#setUsageManager(org.apache.activemq.usage.SystemUsage)
     */
    @Override
    public void setUsageManager(SystemUsage usageManager) {
        this.letter.setUsageManager(usageManager);
    }

    /**
     * @return the size of the store
     * @see org.apache.activemq.store.PersistenceAdapter#size()
     */
    @Override
    public long size() {
        return this.letter.isStarted() ? this.letter.size() : 0l;
    }

    /**
     * @throws Exception
     * @see org.apache.activemq.Service#start()
     */
    @Override
    public void doStart() throws Exception {
        this.letter.start();

        if (brokerService != null && brokerService.isUseJmx()) {
            PersistenceAdapterView view = new PersistenceAdapterView(this);
            view.setInflightTransactionViewCallable(new Callable<String>() {
                @Override
                public String call() throws Exception {
                    return letter.getTransactions();
                }
            });
            view.setDataViewCallable(new Callable<String>() {
                @Override
                public String call() throws Exception {
                    return letter.getJournal().getFileMap().keySet().toString();
                }
            });

            view.setPersistenceAdapterStatistics(letter.persistenceAdapterStatistics);

            AnnotatedMBean.registerMBean(brokerService.getManagementContext(), view,
                    createPersistenceAdapterName(brokerService.getBrokerObjectName().toString(), toString()));
        }
    }

    /**
     * @throws Exception
     * @see org.apache.activemq.Service#stop()
     */
    @Override
    public void doStop(ServiceStopper stopper) throws Exception {
        this.letter.stop();

        if (brokerService != null && brokerService.isUseJmx()) {
            ObjectName brokerObjectName = brokerService.getBrokerObjectName();
            brokerService.getManagementContext().unregisterMBean(createPersistenceAdapterName(brokerObjectName.toString(), toString()));
        }
    }

    /**
     * Get the journalMaxFileLength
     *
     * @return the journalMaxFileLength
     */
    @Override
    public int getJournalMaxFileLength() {
        return this.letter.getJournalMaxFileLength();
    }

    /**
     * When set using Xbean, values of the form "20 Mb", "1024kb", and "1g" can
     * be used
     *
     * @org.apache.xbean.Property propertyEditor="org.apache.activemq.util.MemoryIntPropertyEditor"
     */
    public void setJournalMaxFileLength(int journalMaxFileLength) {
        this.letter.setJournalMaxFileLength(journalMaxFileLength);
    }

    /**
     * Set the max number of producers (LRU cache) to track for duplicate sends
     */
    public void setMaxFailoverProducersToTrack(int maxFailoverProducersToTrack) {
        this.letter.setMaxFailoverProducersToTrack(maxFailoverProducersToTrack);
    }

    public int getMaxFailoverProducersToTrack() {
        return this.letter.getMaxFailoverProducersToTrack();
    }

    /**
     * set the audit window depth for duplicate suppression (should exceed the max transaction
     * batch)
     */
    public void setFailoverProducersAuditDepth(int failoverProducersAuditDepth) {
        this.letter.setFailoverProducersAuditDepth(failoverProducersAuditDepth);
    }

    public int getFailoverProducersAuditDepth() {
        return this.letter.getFailoverProducersAuditDepth();
    }

    /**
     * Get the checkpointInterval
     *
     * @return the checkpointInterval
     */
    public long getCheckpointInterval() {
        return this.letter.getCheckpointInterval();
    }

    /**
     * Set the checkpointInterval
     *
     * @param checkpointInterval
     *            the checkpointInterval to set
     */
    public void setCheckpointInterval(long checkpointInterval) {
        this.letter.setCheckpointInterval(checkpointInterval);
    }

    /**
     * Get the cleanupInterval
     *
     * @return the cleanupInterval
     */
    public long getCleanupInterval() {
        return this.letter.getCleanupInterval();
    }

    /**
     * Set the cleanupInterval
     *
     * @param cleanupInterval
     *            the cleanupInterval to set
     */
    public void setCleanupInterval(long cleanupInterval) {
        this.letter.setCleanupInterval(cleanupInterval);
    }

    /**
     * Get the indexWriteBatchSize
     *
     * @return the indexWriteBatchSize
     */
    public int getIndexWriteBatchSize() {
        return this.letter.getIndexWriteBatchSize();
    }

    /**
     * Set the indexWriteBatchSize
     * When set using Xbean, values of the form "20 Mb", "1024kb", and "1g" can be used
     * @org.apache.xbean.Property propertyEditor="org.apache.activemq.util.MemoryPropertyEditor"
     * @param indexWriteBatchSize
     *            the indexWriteBatchSize to set
     */
    public void setIndexWriteBatchSize(int indexWriteBatchSize) {
        this.letter.setIndexWriteBatchSize(indexWriteBatchSize);
    }

    /**
     * Get the journalMaxWriteBatchSize
     *
     * @return the journalMaxWriteBatchSize
     */
    public int getJournalMaxWriteBatchSize() {
        return this.letter.getJournalMaxWriteBatchSize();
    }

    /**
     * Set the journalMaxWriteBatchSize
     *  * When set using Xbean, values of the form "20 Mb", "1024kb", and "1g" can be used
     * @org.apache.xbean.Property propertyEditor="org.apache.activemq.util.MemoryPropertyEditor"
     * @param journalMaxWriteBatchSize
     *            the journalMaxWriteBatchSize to set
     */
    public void setJournalMaxWriteBatchSize(int journalMaxWriteBatchSize) {
        this.letter.setJournalMaxWriteBatchSize(journalMaxWriteBatchSize);
    }

    /**
     * Get the enableIndexWriteAsync
     *
     * @return the enableIndexWriteAsync
     */
    public boolean isEnableIndexWriteAsync() {
        return this.letter.isEnableIndexWriteAsync();
    }

    /**
     * Set the enableIndexWriteAsync
     *
     * @param enableIndexWriteAsync
     *            the enableIndexWriteAsync to set
     */
    public void setEnableIndexWriteAsync(boolean enableIndexWriteAsync) {
        this.letter.setEnableIndexWriteAsync(enableIndexWriteAsync);
    }

    /**
     * Get the PersistenceAdapterStatistics
     *
     * @return the persistenceAdapterStatistics
     */
    public PersistenceAdapterStatistics getPersistenceAdapterStatistics() {
        return this.letter.getPersistenceAdapterStatistics();
    }

    /**
     * Get the directory
     *
     * @return the directory
     */
    @Override
    public File getDirectory() {
        return this.letter.getDirectory();
    }

    /**
     * @param dir
     * @see org.apache.activemq.store.PersistenceAdapter#setDirectory(java.io.File)
     */
    @Override
    public void setDirectory(File dir) {
        this.letter.setDirectory(dir);
    }

    /**
     * @return the currently configured location of the KahaDB index files.
     */
    public File getIndexDirectory() {
        return this.letter.getIndexDirectory();
    }

    /**
     * Sets the directory where KahaDB index files should be written.
     *
     * @param indexDirectory
     *        the directory where the KahaDB store index files should be written.
     */
    public void setIndexDirectory(File indexDirectory) {
        this.letter.setIndexDirectory(indexDirectory);
    }

    /**
     * Get the enableJournalDiskSyncs
     * @deprecated use {@link #getJournalDiskSyncStrategy} instead
     * @return the enableJournalDiskSyncs
     */
    public boolean isEnableJournalDiskSyncs() {
        return this.letter.isEnableJournalDiskSyncs();
    }

    /**
     * Set the enableJournalDiskSyncs
     *
     * @deprecated use {@link #setJournalDiskSyncStrategy} instead
     * @param enableJournalDiskSyncs
     *            the enableJournalDiskSyncs to set
     */
    public void setEnableJournalDiskSyncs(boolean enableJournalDiskSyncs) {
        this.letter.setEnableJournalDiskSyncs(enableJournalDiskSyncs);
    }

    /**
     * @return
     */
    public String getJournalDiskSyncStrategy() {
        return letter.getJournalDiskSyncStrategy();
    }

    public JournalDiskSyncStrategy getJournalDiskSyncStrategyEnum() {
        return letter.getJournalDiskSyncStrategyEnum();
    }

    /**
     * @param journalDiskSyncStrategy
     */
    public void setJournalDiskSyncStrategy(String journalDiskSyncStrategy) {
        letter.setJournalDiskSyncStrategy(journalDiskSyncStrategy);
    }

    /**
     * @return
     */
    public long getJournalDiskSyncInterval() {
        return letter.getJournalDiskSyncInterval();
    }

    /**
     * @param journalDiskSyncInterval
     */
    public void setJournalDiskSyncInterval(long journalDiskSyncInterval) {
        letter.setJournalDiskSyncInterval(journalDiskSyncInterval);
    }

    /**
     * Get the indexCacheSize
     *
     * @return the indexCacheSize
     */
    public int getIndexCacheSize() {
        return this.letter.getIndexCacheSize();
    }

    /**
     * Set the indexCacheSize
     * When set using Xbean, values of the form "20 Mb", "1024kb", and "1g" can be used
     * @org.apache.xbean.Property propertyEditor="org.apache.activemq.util.MemoryPropertyEditor"
     * @param indexCacheSize
     *            the indexCacheSize to set
     */
    public void setIndexCacheSize(int indexCacheSize) {
        this.letter.setIndexCacheSize(indexCacheSize);
    }

    /**
     * Get the ignoreMissingJournalfiles
     *
     * @return the ignoreMissingJournalfiles
     */
    public boolean isIgnoreMissingJournalfiles() {
        return this.letter.isIgnoreMissingJournalfiles();
    }

    /**
     * Set the ignoreMissingJournalfiles
     *
     * @param ignoreMissingJournalfiles
     *            the ignoreMissingJournalfiles to set
     */
    public void setIgnoreMissingJournalfiles(boolean ignoreMissingJournalfiles) {
        this.letter.setIgnoreMissingJournalfiles(ignoreMissingJournalfiles);
    }

    public boolean isChecksumJournalFiles() {
        return letter.isChecksumJournalFiles();
    }

    public boolean isCheckForCorruptJournalFiles() {
        return letter.isCheckForCorruptJournalFiles();
    }

    public void setChecksumJournalFiles(boolean checksumJournalFiles) {
        letter.setChecksumJournalFiles(checksumJournalFiles);
    }

    public void setCheckForCorruptJournalFiles(boolean checkForCorruptJournalFiles) {
        letter.setCheckForCorruptJournalFiles(checkForCorruptJournalFiles);
    }

    public String getPurgeRecoveredXATransactionStrategy() {
        return letter.getPurgeRecoveredXATransactionStrategy();
    }

    public void setPurgeRecoveredXATransactionStrategy(String purgeRecoveredXATransactionStrategy) {
        letter.setPurgeRecoveredXATransactionStrategy(purgeRecoveredXATransactionStrategy);
    }

    @Override
    public void setBrokerService(BrokerService brokerService) {
        super.setBrokerService(brokerService);
        letter.setBrokerService(brokerService);
    }

    public String getPreallocationScope() {
        return letter.getPreallocationScope();
    }

    public void setPreallocationScope(String preallocationScope) {
        this.letter.setPreallocationScope(preallocationScope);
    }

    public String getPreallocationStrategy() {
        return letter.getPreallocationStrategy();
    }

    public void setPreallocationStrategy(String preallocationStrategy) {
        this.letter.setPreallocationStrategy(preallocationStrategy);
    }

    public boolean isArchiveDataLogs() {
        return letter.isArchiveDataLogs();
    }

    public void setArchiveDataLogs(boolean archiveDataLogs) {
        letter.setArchiveDataLogs(archiveDataLogs);
    }

    public File getDirectoryArchive() {
        return letter.getDirectoryArchive();
    }

    public void setDirectoryArchive(File directoryArchive) {
        letter.setDirectoryArchive(directoryArchive);
    }

    public boolean isConcurrentStoreAndDispatchQueues() {
        return letter.isConcurrentStoreAndDispatchQueues();
    }

    public void setConcurrentStoreAndDispatchQueues(boolean concurrentStoreAndDispatch) {
        letter.setConcurrentStoreAndDispatchQueues(concurrentStoreAndDispatch);
    }

    public boolean isConcurrentStoreAndDispatchTopics() {
        return letter.isConcurrentStoreAndDispatchTopics();
    }

    public void setConcurrentStoreAndDispatchTopics(boolean concurrentStoreAndDispatch) {
        letter.setConcurrentStoreAndDispatchTopics(concurrentStoreAndDispatch);
    }

    public int getMaxAsyncJobs() {
        return letter.getMaxAsyncJobs();
    }
    /**
     * @param maxAsyncJobs
     *            the maxAsyncJobs to set
     */
    public void setMaxAsyncJobs(int maxAsyncJobs) {
        letter.setMaxAsyncJobs(maxAsyncJobs);
    }

    /**
     * @deprecated use {@link Locker#setLockAcquireSleepInterval(long)} instead
     *
     * @param databaseLockedWaitDelay the databaseLockedWaitDelay to set
     */
    @Deprecated
    public void setDatabaseLockedWaitDelay(int databaseLockedWaitDelay) throws IOException {
       getLocker().setLockAcquireSleepInterval(databaseLockedWaitDelay);
    }

    public boolean getForceRecoverIndex() {
        return letter.getForceRecoverIndex();
    }

    public void setForceRecoverIndex(boolean forceRecoverIndex) {
        letter.setForceRecoverIndex(forceRecoverIndex);
    }

    public boolean isArchiveCorruptedIndex() {
        return letter.isArchiveCorruptedIndex();
    }

    public void setArchiveCorruptedIndex(boolean archiveCorruptedIndex) {
        letter.setArchiveCorruptedIndex(archiveCorruptedIndex);
    }

    public float getIndexLFUEvictionFactor() {
        return letter.getIndexLFUEvictionFactor();
    }

    public void setIndexLFUEvictionFactor(float indexLFUEvictionFactor) {
        letter.setIndexLFUEvictionFactor(indexLFUEvictionFactor);
    }

    public boolean isUseIndexLFRUEviction() {
        return letter.isUseIndexLFRUEviction();
    }

    public void setUseIndexLFRUEviction(boolean useIndexLFRUEviction) {
        letter.setUseIndexLFRUEviction(useIndexLFRUEviction);
    }

    public void setEnableIndexDiskSyncs(boolean diskSyncs) {
        letter.setEnableIndexDiskSyncs(diskSyncs);
    }

    public boolean isEnableIndexDiskSyncs() {
        return letter.isEnableIndexDiskSyncs();
    }

    public void setEnableIndexRecoveryFile(boolean enable) {
        letter.setEnableIndexRecoveryFile(enable);
    }

    public boolean  isEnableIndexRecoveryFile() {
        return letter.isEnableIndexRecoveryFile();
    }

    public void setEnableIndexPageCaching(boolean enable) {
        letter.setEnableIndexPageCaching(enable);
    }

    public boolean isEnableIndexPageCaching() {
        return letter.isEnableIndexPageCaching();
    }

    public int getCompactAcksAfterNoGC() {
        return letter.getCompactAcksAfterNoGC();
    }

    /**
     * Sets the number of GC cycles where no journal logs were removed before an attempt to
     * move forward all the acks in the last log that contains them and is otherwise unreferenced.
     * <p>
     * A value of -1 will disable this feature.
     *
     * @param compactAcksAfterNoGC
     *      Number of empty GC cycles before we rewrite old ACKS.
     */
    public void setCompactAcksAfterNoGC(int compactAcksAfterNoGC) {
        this.letter.setCompactAcksAfterNoGC(compactAcksAfterNoGC);
    }

    public boolean isCompactAcksIgnoresStoreGrowth() {
        return this.letter.isCompactAcksIgnoresStoreGrowth();
    }

    /**
     * Configure if Ack compaction will occur regardless of continued growth of the
     * journal logs meaning that the store has not run out of space yet.  Because the
     * compaction operation can be costly this value is defaulted to off and the Ack
     * compaction is only done when it seems that the store cannot grow and larger.
     *
     * @param compactAcksIgnoresStoreGrowth the compactAcksIgnoresStoreGrowth to set
     */
    public void setCompactAcksIgnoresStoreGrowth(boolean compactAcksIgnoresStoreGrowth) {
        this.letter.setCompactAcksIgnoresStoreGrowth(compactAcksIgnoresStoreGrowth);
    }

    /**
     * Returns whether Ack compaction is enabled
     *
     * @return enableAckCompaction
     */
    public boolean isEnableAckCompaction() {
        return letter.isEnableAckCompaction();
    }

    /**
     * Configure if the Ack compaction task should be enabled to run
     *
     * @param enableAckCompaction
     */
    public void setEnableAckCompaction(boolean enableAckCompaction) {
        letter.setEnableAckCompaction(enableAckCompaction);
    }

    /**
     * Whether non-blocking subscription statistics have been enabled
     *
     * @return
     */
    public boolean isEnableSubscriptionStatistics() {
        return letter.isEnableSubscriptionStatistics();
    }

    /**
     * Enable caching statistics for each subscription to allow non-blocking
     * retrieval of metrics.  This could incur some overhead to compute if there are a lot
     * of subscriptions.
     *
     * @param enableSubscriptionStatistics
     */
    public void setEnableSubscriptionStatistics(boolean enableSubscriptionStatistics) {
        letter.setEnableSubscriptionStatistics(enableSubscriptionStatistics);
    }

    public KahaDBStore getStore() {
        return letter;
    }

    public KahaTransactionInfo createTransactionInfo(TransactionId txid) {
        if (txid == null) {
            return null;
        }
        KahaTransactionInfo rc = new KahaTransactionInfo();

        if (txid.isLocalTransaction()) {
            LocalTransactionId t = (LocalTransactionId) txid;
            KahaLocalTransactionId kahaTxId = new KahaLocalTransactionId();
            kahaTxId.setConnectionId(t.getConnectionId().getValue());
            kahaTxId.setTransactionId(t.getValue());
            rc.setLocalTransactionId(kahaTxId);
        } else {
            XATransactionId t = (XATransactionId) txid;
            KahaXATransactionId kahaTxId = new KahaXATransactionId();
            kahaTxId.setBranchQualifier(new Buffer(t.getBranchQualifier()));
            kahaTxId.setGlobalTransactionId(new Buffer(t.getGlobalTransactionId()));
            kahaTxId.setFormatId(t.getFormatId());
            rc.setXaTransactionId(kahaTxId);
        }
        return rc;
    }

    @Override
    public Locker createDefaultLocker() throws IOException {
        SharedFileLocker locker = new SharedFileLocker();
        locker.configure(this);
        return locker;
    }

    @Override
    public void init() throws Exception {}

    @Override
    public String toString() {
        String path = getDirectory() != null ? getDirectory().getAbsolutePath() : "DIRECTORY_NOT_SET";
        return "KahaDBPersistenceAdapter[" + path + (getIndexDirectory() != null ? ",Index:" + getIndexDirectory().getAbsolutePath() : "") +  "]";
    }

    @Override
    public void setTransactionIdTransformer(TransactionIdTransformer transactionIdTransformer) {
        getStore().setTransactionIdTransformer(transactionIdTransformer);
    }

    @Override
    public JobSchedulerStore createJobSchedulerStore() throws IOException, UnsupportedOperationException {
        return this.letter.createJobSchedulerStore();
    }

    /* (non-Javadoc)
     * @see org.apache.activemq.store.NoLocalSubscriptionAware#isPersistNoLocal()
     */
    @Override
    public boolean isPersistNoLocal() {
        return this.letter.isPersistNoLocal();
    }

    /*
     * When set, ensure that the cleanup/gc operation is executed during the stop procedure
     */
    public void setCleanupOnStop(boolean cleanupOnStop) {
        this.letter.setCleanupOnStop(cleanupOnStop);
    }

    public boolean getCleanupOnStop() {
        return this.letter.getCleanupOnStop();
    }

    public void setDataFileFactory(DataFileFactory dataFileFactory) {
        this.letter.setDataFileFactory(dataFileFactory);
    }
}
