| /** |
| * 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.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 = 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(); |
| } |
| } |
