org.apache.activemq.store.kahadb

Class MessageDatabase

java.lang.Object

org.apache.activemq.util.ServiceSupport

org.apache.activemq.store.kahadb.MessageDatabase

All Implemented Interfaces:

BrokerServiceAware, Service

Direct Known Subclasses:

KahaDBStore

public abstract class MessageDatabase
extends ServiceSupport
implements BrokerServiceAware

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
protected class	MessageDatabase.LastAckMarshaller
protected static class	MessageDatabase.MessageKeysMarshaller
protected class	MessageDatabase.Metadata
protected class	MessageDatabase.StoredDestinationMarshaller

Field Summary

Fields

Modifier and Type	Field and Description
protected java.util.Set<java.lang.String>	ackedAndPrepared
protected boolean	archiveDataLogs
protected BrokerService	brokerService
protected java.lang.Thread	checkpointThread
static java.io.File	DEFAULT_DIRECTORY
protected boolean	deleteAllMessages
protected java.io.File	directory
protected java.io.File	directoryArchive
protected boolean	enableJournalDiskSyncs
protected boolean	failIfDatabaseIsLocked
protected boolean	forceRecoverIndex
protected java.util.concurrent.locks.ReentrantReadWriteLock	indexLock
protected Journal	journal
protected java.util.concurrent.atomic.AtomicLong	journalSize
static int	LOG_SLOW_ACCESS_TIME
protected MessageDatabase.Metadata	metadata
protected org.apache.activemq.store.kahadb.MessageDatabase.MetadataMarshaller	metadataMarshaller
protected java.util.concurrent.atomic.AtomicBoolean	opened
protected PageFile	pageFile
protected java.util.LinkedHashMap<TransactionId,java.util.List<org.apache.activemq.store.kahadb.MessageDatabase.Operation>>	preparedTransactions
static java.lang.String	PROPERTY_LOG_SLOW_ACCESS_TIME
protected static org.apache.activemq.protobuf.Buffer	UNMATCHED

Constructor Summary

Constructors

Constructor and Description
MessageDatabase()

Method Summary

Methods

Modifier and Type	Method and Description
protected void	checkpointCleanup(boolean cleanup)
void	close()
void	doStart()
void	doStop(ServiceStopper stopper)
void	forgetRecoveredAcks(java.util.ArrayList<MessageAck> acks)
long	getCheckpointInterval()
long	getCleanupInterval()
java.io.File	getDirectory()
java.io.File	getDirectoryArchive()
protected org.apache.activemq.store.kahadb.MessageDatabase.StoredDestination	getExistingStoredDestination(org.apache.activemq.store.kahadb.data.KahaDestination destination, Transaction tx)
int	getFailoverProducersAuditDepth()
int	getIndexCacheSize()
float	getIndexLFUEvictionFactor()
int	getIndexWriteBatchSize()
Location[]	getInProgressTxLocationRange()
Journal	getJournal()
java.util.HashSet<java.lang.Integer>	getJournalFilesBeingReplicated()
int	getJournalMaxFileLength()
int	getJournalMaxWriteBatchSize()
org.apache.activemq.store.kahadb.MessageDatabase.LastAck	getLastAck(Transaction tx, org.apache.activemq.store.kahadb.MessageDatabase.StoredDestination sd, java.lang.String subscriptionKey)
Location	getLastUpdatePosition()
int	getMaxFailoverProducersToTrack()
PageFile	getPageFile()
protected org.apache.activemq.store.kahadb.MessageDatabase.StoredDestination	getStoredDestination(org.apache.activemq.store.kahadb.data.KahaDestination destination, Transaction tx)
long	getStoredMessageCount(Transaction tx, org.apache.activemq.store.kahadb.MessageDatabase.StoredDestination sd, java.lang.String subscriptionKey)
java.lang.String	getTransactions()
void	incrementalRecover()
boolean	isArchiveCorruptedIndex()
boolean	isArchiveDataLogs()
boolean	isCheckForCorruptJournalFiles()
boolean	isChecksumJournalFiles()
boolean	isDeleteAllMessages()
boolean	isEnableIndexDiskSyncs()
boolean	isEnableIndexPageCaching()
boolean	isEnableIndexRecoveryFile()
boolean	isEnableJournalDiskSyncs()
boolean	isFailIfDatabaseIsLocked()
boolean	isIgnoreMissingJournalfiles()
boolean	isRewriteOnRedelivery()
boolean	isUseIndexLFRUEviction()
void	load()
JournalCommand<?>	load(Location location) Loads a previously stored JournalMessage
void	open()
protected void	process(org.apache.activemq.store.kahadb.data.KahaAddMessageCommand command, Location location)
protected void	process(org.apache.activemq.store.kahadb.data.KahaCommitCommand command, Location location, java.lang.Runnable after)
protected void	process(org.apache.activemq.store.kahadb.data.KahaPrepareCommand command, Location location)
protected void	process(org.apache.activemq.store.kahadb.data.KahaRemoveDestinationCommand command, Location location)
protected void	process(org.apache.activemq.store.kahadb.data.KahaRemoveMessageCommand command, Location location)
protected void	process(org.apache.activemq.store.kahadb.data.KahaRollbackCommand command, Location location)
protected void	process(org.apache.activemq.store.kahadb.data.KahaSubscriptionCommand command, Location location)
protected void	processLocation(Location location)
protected void	recoverIndex(Transaction tx)
void	setArchiveCorruptedIndex(boolean archiveCorruptedIndex)
void	setArchiveDataLogs(boolean archiveDataLogs)
void	setBrokerService(BrokerService brokerService)
void	setCheckForCorruptJournalFiles(boolean checkForCorruptJournalFiles)
void	setCheckpointInterval(long checkpointInterval)
void	setChecksumJournalFiles(boolean checksumJournalFiles)
void	setCleanupInterval(long cleanupInterval)
void	setDeleteAllMessages(boolean deleteAllMessages)
void	setDirectory(java.io.File directory)
void	setDirectoryArchive(java.io.File directoryArchive)
void	setEnableIndexDiskSyncs(boolean enableIndexDiskSyncs)
void	setEnableIndexPageCaching(boolean enableIndexPageCaching)
void	setEnableIndexRecoveryFile(boolean enableIndexRecoveryFile)
void	setEnableIndexWriteAsync(boolean enableIndexWriteAsync)
void	setEnableJournalDiskSyncs(boolean syncWrites)
void	setFailIfDatabaseIsLocked(boolean failIfDatabaseIsLocked)
void	setFailoverProducersAuditDepth(int failoverProducersAuditDepth)
void	setIgnoreMissingJournalfiles(boolean ignoreMissingJournalfiles)
void	setIndexCacheSize(int indexCacheSize)
void	setIndexLFUEvictionFactor(float indexLFUEvictionFactor)
void	setIndexWriteBatchSize(int setIndexWriteBatchSize)
void	setJournalMaxFileLength(int journalMaxFileLength)
void	setJournalMaxWriteBatchSize(int journalMaxWriteBatchSize)
void	setMaxFailoverProducersToTrack(int maxFailoverProducersToTrack)
void	setRewriteOnRedelivery(boolean rewriteOnRedelivery)
void	setUseIndexLFRUEviction(boolean useIndexLFRUEviction)
Location	store(JournalCommand<?> data)
Location	store(JournalCommand<?> data, boolean sync, java.lang.Runnable before, java.lang.Runnable after)
Location	store(JournalCommand<?> data, boolean sync, java.lang.Runnable before, java.lang.Runnable after, java.lang.Runnable onJournalStoreComplete) All updated are are funneled through this method.
Location	store(JournalCommand<?> data, java.lang.Runnable onJournalStoreComplete)
ByteSequence	toByteSequence(JournalCommand<?> data)
void	trackRecoveredAcks(java.util.ArrayList<MessageAck> acks)
void	unload()

Methods inherited from class org.apache.activemq.util.ServiceSupport

addServiceListener, dispose, isStarted, isStopped, isStopping, postStop, preStart, removeServiceListener, start, stop

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Field Detail

brokerService

protected BrokerService brokerService

PROPERTY_LOG_SLOW_ACCESS_TIME

public static final java.lang.String PROPERTY_LOG_SLOW_ACCESS_TIME

See Also:

Constant Field Values

LOG_SLOW_ACCESS_TIME

public static final int LOG_SLOW_ACCESS_TIME

DEFAULT_DIRECTORY

public static final java.io.File DEFAULT_DIRECTORY

UNMATCHED

protected static final org.apache.activemq.protobuf.Buffer UNMATCHED

pageFile

protected PageFile pageFile

journal

protected Journal journal

metadata

protected MessageDatabase.Metadata metadata

metadataMarshaller

protected org.apache.activemq.store.kahadb.MessageDatabase.MetadataMarshaller metadataMarshaller

failIfDatabaseIsLocked

protected boolean failIfDatabaseIsLocked

deleteAllMessages

protected boolean deleteAllMessages

directory

protected java.io.File directory

checkpointThread

protected java.lang.Thread checkpointThread

enableJournalDiskSyncs

protected boolean enableJournalDiskSyncs

archiveDataLogs

protected boolean archiveDataLogs

directoryArchive

protected java.io.File directoryArchive

journalSize

protected java.util.concurrent.atomic.AtomicLong journalSize

opened

protected java.util.concurrent.atomic.AtomicBoolean opened

forceRecoverIndex

protected boolean forceRecoverIndex

indexLock

protected final java.util.concurrent.locks.ReentrantReadWriteLock indexLock

preparedTransactions

protected final java.util.LinkedHashMap<TransactionId,java.util.List<org.apache.activemq.store.kahadb.MessageDatabase.Operation>> preparedTransactions

ackedAndPrepared

protected final java.util.Set<java.lang.String> ackedAndPrepared

Constructor Detail

MessageDatabase

public MessageDatabase()

Method Detail

doStart

public void doStart()
             throws java.lang.Exception

Specified by:

doStart in class ServiceSupport

Throws:

java.lang.Exception

doStop

public void doStop(ServiceStopper stopper)
            throws java.lang.Exception

Specified by:

doStop in class ServiceSupport

Throws:

java.lang.Exception

open

public void open()
          throws java.io.IOException

Throws:

java.io.IOException

load

public void load()
          throws java.io.IOException

Throws:

java.io.IOException

close

public void close()
           throws java.io.IOException,
                  java.lang.InterruptedException

Throws:

java.io.IOException

java.lang.InterruptedException

unload

public void unload()
            throws java.io.IOException,
                   java.lang.InterruptedException

Throws:

java.io.IOException

java.lang.InterruptedException

getInProgressTxLocationRange

public Location[] getInProgressTxLocationRange()

getTransactions

public java.lang.String getTransactions()

recoverIndex

protected void recoverIndex(Transaction tx)
                     throws java.io.IOException

Throws:

java.io.IOException

incrementalRecover

public void incrementalRecover()
                        throws java.io.IOException

Throws:

java.io.IOException

getLastUpdatePosition

public Location getLastUpdatePosition()
                               throws java.io.IOException

Throws:

java.io.IOException

checkpointCleanup

protected void checkpointCleanup(boolean cleanup)
                          throws java.io.IOException

Throws:

java.io.IOException

toByteSequence

public ByteSequence toByteSequence(JournalCommand<?> data)
                            throws java.io.IOException

Throws:

java.io.IOException

store

public Location store(JournalCommand<?> data)
               throws java.io.IOException

Throws:

java.io.IOException

store

public Location store(JournalCommand<?> data,
             java.lang.Runnable onJournalStoreComplete)
               throws java.io.IOException

Throws:

java.io.IOException

store

public Location store(JournalCommand<?> data,
             boolean sync,
             java.lang.Runnable before,
             java.lang.Runnable after)
               throws java.io.IOException

Throws:

java.io.IOException

store

public Location store(JournalCommand<?> data,
             boolean sync,
             java.lang.Runnable before,
             java.lang.Runnable after,
             java.lang.Runnable onJournalStoreComplete)
               throws java.io.IOException

All updated are are funneled through this method. The updates are converted to a JournalMessage which is logged to the journal and then the data from the JournalMessage is used to update the index just like it would be done during a recovery process.

Throws:

java.io.IOException

load

public JournalCommand<?> load(Location location)
                       throws java.io.IOException

Loads a previously stored JournalMessage

Parameters:

location -

Returns:

Throws:

java.io.IOException

process

protected void process(org.apache.activemq.store.kahadb.data.KahaAddMessageCommand command,
           Location location)
                throws java.io.IOException

Throws:

java.io.IOException

process

protected void process(org.apache.activemq.store.kahadb.data.KahaRemoveMessageCommand command,
           Location location)
                throws java.io.IOException

Throws:

java.io.IOException

process

protected void process(org.apache.activemq.store.kahadb.data.KahaRemoveDestinationCommand command,
           Location location)
                throws java.io.IOException

Throws:

java.io.IOException

process

protected void process(org.apache.activemq.store.kahadb.data.KahaSubscriptionCommand command,
           Location location)
                throws java.io.IOException

Throws:

java.io.IOException

processLocation

protected void processLocation(Location location)

process

protected void process(org.apache.activemq.store.kahadb.data.KahaCommitCommand command,
           Location location,
           java.lang.Runnable after)
                throws java.io.IOException

Throws:

java.io.IOException

process

protected void process(org.apache.activemq.store.kahadb.data.KahaPrepareCommand command,
           Location location)

process

protected void process(org.apache.activemq.store.kahadb.data.KahaRollbackCommand command,
           Location location)
                throws java.io.IOException

Throws:

java.io.IOException

getJournalFilesBeingReplicated

public java.util.HashSet<java.lang.Integer> getJournalFilesBeingReplicated()

getStoredDestination

protected org.apache.activemq.store.kahadb.MessageDatabase.StoredDestination getStoredDestination(org.apache.activemq.store.kahadb.data.KahaDestination destination,
                                                                                      Transaction tx)
                                                                                           throws java.io.IOException

Throws:

java.io.IOException

getExistingStoredDestination

protected org.apache.activemq.store.kahadb.MessageDatabase.StoredDestination getExistingStoredDestination(org.apache.activemq.store.kahadb.data.KahaDestination destination,
                                                                                              Transaction tx)
                                                                                                   throws java.io.IOException

Throws:

java.io.IOException

getLastAck

public org.apache.activemq.store.kahadb.MessageDatabase.LastAck getLastAck(Transaction tx,
                                                                  org.apache.activemq.store.kahadb.MessageDatabase.StoredDestination sd,
                                                                  java.lang.String subscriptionKey)
                                                                    throws java.io.IOException

Throws:

java.io.IOException

getStoredMessageCount

public long getStoredMessageCount(Transaction tx,
                         org.apache.activemq.store.kahadb.MessageDatabase.StoredDestination sd,
                         java.lang.String subscriptionKey)
                           throws java.io.IOException

Throws:

java.io.IOException

trackRecoveredAcks

public void trackRecoveredAcks(java.util.ArrayList<MessageAck> acks)

forgetRecoveredAcks

public void forgetRecoveredAcks(java.util.ArrayList<MessageAck> acks)
                         throws java.io.IOException

Throws:

java.io.IOException

getJournalMaxWriteBatchSize

public int getJournalMaxWriteBatchSize()

setJournalMaxWriteBatchSize

public void setJournalMaxWriteBatchSize(int journalMaxWriteBatchSize)

getDirectory

public java.io.File getDirectory()

setDirectory

public void setDirectory(java.io.File directory)

isDeleteAllMessages

public boolean isDeleteAllMessages()

setDeleteAllMessages

public void setDeleteAllMessages(boolean deleteAllMessages)

setIndexWriteBatchSize

public void setIndexWriteBatchSize(int setIndexWriteBatchSize)

getIndexWriteBatchSize

public int getIndexWriteBatchSize()

setEnableIndexWriteAsync

public void setEnableIndexWriteAsync(boolean enableIndexWriteAsync)

isEnableJournalDiskSyncs

public boolean isEnableJournalDiskSyncs()

setEnableJournalDiskSyncs

public void setEnableJournalDiskSyncs(boolean syncWrites)

getCheckpointInterval

public long getCheckpointInterval()

setCheckpointInterval

public void setCheckpointInterval(long checkpointInterval)

getCleanupInterval

public long getCleanupInterval()

setCleanupInterval

public void setCleanupInterval(long cleanupInterval)

setJournalMaxFileLength

public void setJournalMaxFileLength(int journalMaxFileLength)

getJournalMaxFileLength

public int getJournalMaxFileLength()

setMaxFailoverProducersToTrack

public void setMaxFailoverProducersToTrack(int maxFailoverProducersToTrack)

getMaxFailoverProducersToTrack

public int getMaxFailoverProducersToTrack()

setFailoverProducersAuditDepth

public void setFailoverProducersAuditDepth(int failoverProducersAuditDepth)

getFailoverProducersAuditDepth

public int getFailoverProducersAuditDepth()

getPageFile

public PageFile getPageFile()

getJournal

public Journal getJournal()
                   throws java.io.IOException

Throws:

java.io.IOException

isFailIfDatabaseIsLocked

public boolean isFailIfDatabaseIsLocked()

setFailIfDatabaseIsLocked

public void setFailIfDatabaseIsLocked(boolean failIfDatabaseIsLocked)

isIgnoreMissingJournalfiles

public boolean isIgnoreMissingJournalfiles()

setIgnoreMissingJournalfiles

public void setIgnoreMissingJournalfiles(boolean ignoreMissingJournalfiles)

getIndexCacheSize

public int getIndexCacheSize()

setIndexCacheSize

public void setIndexCacheSize(int indexCacheSize)

isCheckForCorruptJournalFiles

public boolean isCheckForCorruptJournalFiles()

setCheckForCorruptJournalFiles

public void setCheckForCorruptJournalFiles(boolean checkForCorruptJournalFiles)

isChecksumJournalFiles

public boolean isChecksumJournalFiles()

setChecksumJournalFiles

public void setChecksumJournalFiles(boolean checksumJournalFiles)

setBrokerService

public void setBrokerService(BrokerService brokerService)

Specified by:

setBrokerService in interface BrokerServiceAware

isArchiveDataLogs

public boolean isArchiveDataLogs()

Returns:

the archiveDataLogs

setArchiveDataLogs

public void setArchiveDataLogs(boolean archiveDataLogs)

Parameters:

archiveDataLogs - the archiveDataLogs to set

getDirectoryArchive

public java.io.File getDirectoryArchive()

Returns:

the directoryArchive

setDirectoryArchive

public void setDirectoryArchive(java.io.File directoryArchive)

Parameters:

directoryArchive - the directoryArchive to set

isRewriteOnRedelivery

public boolean isRewriteOnRedelivery()

setRewriteOnRedelivery

public void setRewriteOnRedelivery(boolean rewriteOnRedelivery)

isArchiveCorruptedIndex

public boolean isArchiveCorruptedIndex()

setArchiveCorruptedIndex

public void setArchiveCorruptedIndex(boolean archiveCorruptedIndex)

getIndexLFUEvictionFactor

public float getIndexLFUEvictionFactor()

setIndexLFUEvictionFactor

public void setIndexLFUEvictionFactor(float indexLFUEvictionFactor)

isUseIndexLFRUEviction

public boolean isUseIndexLFRUEviction()

setUseIndexLFRUEviction

public void setUseIndexLFRUEviction(boolean useIndexLFRUEviction)

setEnableIndexDiskSyncs

public void setEnableIndexDiskSyncs(boolean enableIndexDiskSyncs)

setEnableIndexRecoveryFile

public void setEnableIndexRecoveryFile(boolean enableIndexRecoveryFile)

setEnableIndexPageCaching

public void setEnableIndexPageCaching(boolean enableIndexPageCaching)

isEnableIndexDiskSyncs

public boolean isEnableIndexDiskSyncs()

isEnableIndexRecoveryFile

public boolean isEnableIndexRecoveryFile()

isEnableIndexPageCaching

public boolean isEnableIndexPageCaching()