org.apache.activemq.artemis.core.persistence

Interface StorageManager

All Superinterfaces:

ActiveMQComponent, IDGenerator

All Known Implementing Classes:

AbstractJournalStorageManager, JDBCJournalStorageManager, JournalStorageManager, NullStorageManager

public interface StorageManager
extends IDGenerator, ActiveMQComponent

A StorageManager Note about IDGEnerator I've changed StorageManager to extend IDGenerator, because in some places all we needed from the StorageManager was the idGeneration. I couldn't just get the IDGenerator from the inner part because the NullPersistent has its own sequence. So the best was to add the interface and adjust the callers for the method

Nested Class Summary

Nested Classes

Modifier and Type	Interface and Description
static class	StorageManager.LargeMessageExtension

Method Summary

Modifier and Type	Method and Description
void	addBytesToLargeMessage(SequentialFile appendFile, long messageID, byte[] bytes)
void	addGrouping(GroupBinding groupBinding)
void	addQueueBinding(long tx, Binding binding)
boolean	addToPage(PagingStore store, ServerMessage msg, Transaction tx, RouteContextList listCtx) Write message to page if we are paging.
void	afterCompleteOperations(IOCallback run)
void	afterPageRead() We need a safeguard in place to avoid too much concurrent IO happening on Paging, otherwise the system may become unresponsive if too many destinations are reading all the same time.
void	afterStoreOperations(IOCallback run) This is similar to afterComplete, however this only cares about the journal part.
ByteBuffer	allocateDirectBuffer(int size) AIO has an optimized buffer which has a method to release it instead of the way NIO will release data based on GC.
void	beforePageRead() We need a safeguard in place to avoid too much concurrent IO happening on Paging, otherwise the system may become unresponsive if too many destinations are reading all the same time.
void	clearContext()
void	commit(long txID)
void	commit(long txID, boolean lineUpContext)
void	commitBindings(long txID)
void	confirmPendingLargeMessage(long recordID) Confirms that a large message was finished
void	confirmPendingLargeMessageTX(Transaction transaction, long messageID, long recordID) Confirms that a large message was finished
SequentialFile	createFileForLargeMessage(long messageID, StorageManager.LargeMessageExtension extension) Instantiates a SequentialFile to be used for storing a LargeServerMessage.
LargeServerMessage	createLargeMessage()
LargeServerMessage	createLargeMessage(long id, MessageInternal message) Creates a new LargeMessage with the given id.
void	criticalError(Throwable error)
void	deleteAddressSetting(SimpleString addressMatch)
void	deleteCursorAcknowledge(long ackID)
void	deleteCursorAcknowledgeTransactional(long txID, long ackID)
void	deleteDuplicateID(long recordID)
void	deleteDuplicateIDTransactional(long txID, long recordID)
void	deleteGrouping(long tx, GroupBinding groupBinding)
void	deleteHeuristicCompletion(long id)
void	deleteID(long journalD)
void	deleteIncrementRecord(long txID, long recordID)
void	deleteMessage(long messageID)
void	deletePageComplete(long ackID)
void	deletePageCounter(long txID, long recordID)
void	deletePageTransactional(long recordID)
void	deletePendingPageCounter(long txID, long recordID)
void	deleteQueueBinding(long tx, long queueBindingID)
void	deleteQueueStatus(long recordID)
void	deleteSecurityRoles(SimpleString addressMatch)
void	freeDirectBuffer(ByteBuffer buffer) AIO has an optimized buffer which has a method to release it instead of the way NIO will release data based on GC.
Journal	getBindingsJournal()
OperationContext	getContext() Get the context associated with the thread for later reuse
Journal	getMessageJournal()
void	injectMonitor(FileStoreMonitor monitor)
void	lineUpContext()
JournalLoadInformation	loadBindingJournal(List<QueueBindingInfo> queueBindingInfos, List<GroupingInfo> groupingInfos)
JournalLoadInformation	loadMessageJournal(PostOffice postOffice, PagingManager pagingManager, ResourceManager resourceManager, Map<Long,QueueBindingInfo> queueInfos, Map<SimpleString,List<Pair<byte[],Long>>> duplicateIDMap, Set<Pair<Long,Long>> pendingLargeMessages, List<PageCountPending> pendingNonTXPageCounter, JournalLoader journalLoader)
OperationContext	newContext(Executor executor) It just creates an OperationContext without associating it
OperationContext	newSingleThreadContext()
void	pageClosed(SimpleString storeName, int pageNumber)
void	pageDeleted(SimpleString storeName, int pageNumber)
void	pageWrite(PagedMessage message, int pageNumber)
void	persistIdGenerator() Closes the IDGenerator persisting the current record ID.
void	prepare(long txID, Xid xid)
void	readLock() Read lock the StorageManager.
void	readUnLock() Unlock the manager.
List<PersistedAddressSetting>	recoverAddressSettings()
List<PersistedRoles>	recoverPersistedRoles()
void	rollback(long txID)
void	rollbackBindings(long txID)
void	setContext(OperationContext context) Set the context back to the thread
void	startReplication(ReplicationManager replicationManager, PagingManager pagingManager, String nodeID, boolean autoFailBack, long initialReplicationSyncTimeout)
void	stop(boolean ioCriticalError, boolean sendFailover)
void	stopReplication() Stops the replication of data from the live to the backup.
void	storeAcknowledge(long queueID, long messageID)
void	storeAcknowledgeTransactional(long txID, long queueID, long messageID)
void	storeAddressSetting(PersistedAddressSetting addressSetting)
void	storeCursorAcknowledge(long queueID, PagePosition position)
void	storeCursorAcknowledgeTransactional(long txID, long queueID, PagePosition position)
void	storeDuplicateID(SimpleString address, byte[] duplID, long recordID)
void	storeDuplicateIDTransactional(long txID, SimpleString address, byte[] duplID, long recordID)
long	storeHeuristicCompletion(Xid xid, boolean isCommit)
void	storeID(long journalID, long id) Stores the id from IDManager.
void	storeMessage(ServerMessage message)
void	storeMessageTransactional(long txID, ServerMessage message)
void	storePageCompleteTransactional(long txID, long queueID, PagePosition position)
long	storePageCounter(long txID, long queueID, long value)
long	storePageCounterInc(long queueID, int add)
long	storePageCounterInc(long txID, long queueID, int add)
void	storePageTransaction(long txID, PageTransactionInfo pageTransaction)
long	storePendingCounter(long queueID, long pageID, int inc)
long	storeQueueStatus(long queueID, QueueStatus status)
void	storeReference(long queueID, long messageID, boolean last)
void	storeReferenceTransactional(long txID, long queueID, long messageID)
void	storeSecurityRoles(PersistedRoles persistedRoles)
void	updateDeliveryCount(MessageReference ref)
void	updateDuplicateIDTransactional(long txID, SimpleString address, byte[] duplID, long recordID)
void	updatePageTransaction(long txID, PageTransactionInfo pageTransaction, int depage)
void	updatePageTransaction(PageTransactionInfo pageTransaction, int depage) FIXME Unused
void	updateScheduledDeliveryTime(MessageReference ref)
void	updateScheduledDeliveryTimeTransactional(long txID, MessageReference ref)
void	waitOnOperations() Block until the operations are done.
boolean	waitOnOperations(long timeout) Block until the operations are done.

Methods inherited from interface org.apache.activemq.artemis.utils.IDGenerator

generateID, getCurrentID

Methods inherited from interface org.apache.activemq.artemis.core.server.ActiveMQComponent

isStarted, start, stop

Method Detail

criticalError

void criticalError(Throwable error)

getContext

OperationContext getContext()

Get the context associated with the thread for later reuse

lineUpContext

void lineUpContext()

newContext

OperationContext newContext(Executor executor)

It just creates an OperationContext without associating it

newSingleThreadContext

OperationContext newSingleThreadContext()

setContext

void setContext(OperationContext context)

Set the context back to the thread

stop

void stop(boolean ioCriticalError,
          boolean sendFailover)
   throws Exception

Parameters:

ioCriticalError - is the server being stopped due to an IO critical error.

sendFailover - this is to send the replication stopping in case of replication.

Throws:

Exception

pageClosed

void pageClosed(SimpleString storeName,
                int pageNumber)

pageDeleted

void pageDeleted(SimpleString storeName,
                 int pageNumber)

pageWrite

void pageWrite(PagedMessage message,
               int pageNumber)

afterCompleteOperations

void afterCompleteOperations(IOCallback run)

afterStoreOperations

void afterStoreOperations(IOCallback run)

This is similar to afterComplete, however this only cares about the journal part.

waitOnOperations

boolean waitOnOperations(long timeout)
                  throws Exception

Block until the operations are done. Warning: Don't use it inside an ordered executor, otherwise the system may lock up in case of the pools are full

Throws:

Exception

waitOnOperations

void waitOnOperations()
               throws Exception

Block until the operations are done. Warning: Don't use it inside an ordered executor, otherwise the system may lock up in case of the pools are full

Throws:

Exception

beforePageRead

void beforePageRead()
             throws Exception

We need a safeguard in place to avoid too much concurrent IO happening on Paging, otherwise the system may become unresponsive if too many destinations are reading all the same time. This is called before we read, so we can limit concurrent reads

Throws:

Exception

afterPageRead

void afterPageRead()
            throws Exception

We need a safeguard in place to avoid too much concurrent IO happening on Paging, otherwise the system may become unresponsive if too many destinations are reading all the same time. This is called after we read, so we can limit concurrent reads

Throws:

Exception

allocateDirectBuffer

ByteBuffer allocateDirectBuffer(int size)

AIO has an optimized buffer which has a method to release it instead of the way NIO will release data based on GC. These methods will use that buffer if the inner method supports it

freeDirectBuffer

void freeDirectBuffer(ByteBuffer buffer)

AIO has an optimized buffer which has a method to release it instead of the way NIO will release data based on GC. These methods will use that buffer if the inner method supports it

clearContext

void clearContext()

confirmPendingLargeMessageTX

void confirmPendingLargeMessageTX(Transaction transaction,
                                  long messageID,
                                  long recordID)
                           throws Exception

Confirms that a large message was finished

Throws:

Exception

confirmPendingLargeMessage

void confirmPendingLargeMessage(long recordID)
                         throws Exception

Confirms that a large message was finished

Throws:

Exception

storeMessage

void storeMessage(ServerMessage message)
           throws Exception

Throws:

Exception

storeReference

void storeReference(long queueID,
                    long messageID,
                    boolean last)
             throws Exception

Throws:

Exception

deleteMessage

void deleteMessage(long messageID)
            throws Exception

Throws:

Exception

storeAcknowledge

void storeAcknowledge(long queueID,
                      long messageID)
               throws Exception

Throws:

Exception

storeCursorAcknowledge

void storeCursorAcknowledge(long queueID,
                            PagePosition position)
                     throws Exception

Throws:

Exception

updateDeliveryCount

void updateDeliveryCount(MessageReference ref)
                  throws Exception

Throws:

Exception

updateScheduledDeliveryTime

void updateScheduledDeliveryTime(MessageReference ref)
                          throws Exception

Throws:

Exception

storeDuplicateID

void storeDuplicateID(SimpleString address,
                      byte[] duplID,
                      long recordID)
               throws Exception

Throws:

Exception

deleteDuplicateID

void deleteDuplicateID(long recordID)
                throws Exception

Throws:

Exception

storeMessageTransactional

void storeMessageTransactional(long txID,
                               ServerMessage message)
                        throws Exception

Throws:

Exception

storeReferenceTransactional

void storeReferenceTransactional(long txID,
                                 long queueID,
                                 long messageID)
                          throws Exception

Throws:

Exception

storeAcknowledgeTransactional

void storeAcknowledgeTransactional(long txID,
                                   long queueID,
                                   long messageID)
                            throws Exception

Throws:

Exception

storeCursorAcknowledgeTransactional

void storeCursorAcknowledgeTransactional(long txID,
                                         long queueID,
                                         PagePosition position)
                                  throws Exception

Throws:

Exception

deleteCursorAcknowledgeTransactional

void deleteCursorAcknowledgeTransactional(long txID,
                                          long ackID)
                                   throws Exception

Throws:

Exception

deleteCursorAcknowledge

void deleteCursorAcknowledge(long ackID)
                      throws Exception

Throws:

Exception

storePageCompleteTransactional

void storePageCompleteTransactional(long txID,
                                    long queueID,
                                    PagePosition position)
                             throws Exception

Throws:

Exception

deletePageComplete

void deletePageComplete(long ackID)
                 throws Exception

Throws:

Exception

updateScheduledDeliveryTimeTransactional

void updateScheduledDeliveryTimeTransactional(long txID,
                                              MessageReference ref)
                                       throws Exception

Throws:

Exception

storeDuplicateIDTransactional

void storeDuplicateIDTransactional(long txID,
                                   SimpleString address,
                                   byte[] duplID,
                                   long recordID)
                            throws Exception

Throws:

Exception

updateDuplicateIDTransactional

void updateDuplicateIDTransactional(long txID,
                                    SimpleString address,
                                    byte[] duplID,
                                    long recordID)
                             throws Exception

Throws:

Exception

deleteDuplicateIDTransactional

void deleteDuplicateIDTransactional(long txID,
                                    long recordID)
                             throws Exception

Throws:

Exception

createLargeMessage

LargeServerMessage createLargeMessage()

createLargeMessage

LargeServerMessage createLargeMessage(long id,
                                      MessageInternal message)
                               throws Exception

Creates a new LargeMessage with the given id.

Parameters:

id -

message - This is a temporary message that holds the parsed properties. The remoting layer can't create a ServerMessage directly, then this will be replaced.

Returns:

a large message object

Throws:

Exception

createFileForLargeMessage

SequentialFile createFileForLargeMessage(long messageID,
                                         StorageManager.LargeMessageExtension extension)

Instantiates a SequentialFile to be used for storing a LargeServerMessage.

Parameters:

messageID - the id of the message

extension - the extension to add to the file

Returns:

prepare

void prepare(long txID,
             Xid xid)
      throws Exception

Throws:

Exception

commit

void commit(long txID)
     throws Exception

Throws:

Exception

commit

void commit(long txID,
            boolean lineUpContext)
     throws Exception

Throws:

Exception

rollback

void rollback(long txID)
       throws Exception

Throws:

Exception

rollbackBindings

void rollbackBindings(long txID)
               throws Exception

Throws:

Exception

commitBindings

void commitBindings(long txID)
             throws Exception

Throws:

Exception

storePageTransaction

void storePageTransaction(long txID,
                          PageTransactionInfo pageTransaction)
                   throws Exception

Throws:

Exception

updatePageTransaction

void updatePageTransaction(long txID,
                           PageTransactionInfo pageTransaction,
                           int depage)
                    throws Exception

Throws:

Exception

updatePageTransaction

void updatePageTransaction(PageTransactionInfo pageTransaction,
                           int depage)
                    throws Exception

FIXME Unused

Throws:

Exception

deletePageTransactional

void deletePageTransactional(long recordID)
                      throws Exception

Throws:

Exception

loadMessageJournal

JournalLoadInformation loadMessageJournal(PostOffice postOffice,
                                          PagingManager pagingManager,
                                          ResourceManager resourceManager,
                                          Map<Long,QueueBindingInfo> queueInfos,
                                          Map<SimpleString,List<Pair<byte[],Long>>> duplicateIDMap,
                                          Set<Pair<Long,Long>> pendingLargeMessages,
                                          List<PageCountPending> pendingNonTXPageCounter,
                                          JournalLoader journalLoader)
                                   throws Exception

Throws:

Exception

storeHeuristicCompletion

long storeHeuristicCompletion(Xid xid,
                              boolean isCommit)
                       throws Exception

Throws:

Exception

deleteHeuristicCompletion

void deleteHeuristicCompletion(long id)
                        throws Exception

Throws:

Exception

addQueueBinding

void addQueueBinding(long tx,
                     Binding binding)
              throws Exception

Throws:

Exception

deleteQueueBinding

void deleteQueueBinding(long tx,
                        long queueBindingID)
                 throws Exception

Throws:

Exception

storeQueueStatus

long storeQueueStatus(long queueID,
                      QueueStatus status)
               throws Exception

Parameters:

queueID - The id of the queue

status - The current status of the queue. (Reserved for future use, ATM we only use this record for PAUSED)

Returns:

the id of the journal

Throws:

Exception

deleteQueueStatus

void deleteQueueStatus(long recordID)
                throws Exception

Throws:

Exception

loadBindingJournal

JournalLoadInformation loadBindingJournal(List<QueueBindingInfo> queueBindingInfos,
                                          List<GroupingInfo> groupingInfos)
                                   throws Exception

Throws:

Exception

addGrouping

void addGrouping(GroupBinding groupBinding)
          throws Exception

Throws:

Exception

deleteGrouping

void deleteGrouping(long tx,
                    GroupBinding groupBinding)
             throws Exception

Throws:

Exception

storeAddressSetting

void storeAddressSetting(PersistedAddressSetting addressSetting)
                  throws Exception

Throws:

Exception

deleteAddressSetting

void deleteAddressSetting(SimpleString addressMatch)
                   throws Exception

Throws:

Exception

recoverAddressSettings

List<PersistedAddressSetting> recoverAddressSettings()
                                              throws Exception

Throws:

Exception

storeSecurityRoles

void storeSecurityRoles(PersistedRoles persistedRoles)
                 throws Exception

Throws:

Exception

deleteSecurityRoles

void deleteSecurityRoles(SimpleString addressMatch)
                  throws Exception

Throws:

Exception

recoverPersistedRoles

List<PersistedRoles> recoverPersistedRoles()
                                    throws Exception

Throws:

Exception

storePageCounter

long storePageCounter(long txID,
                      long queueID,
                      long value)
               throws Exception

Returns:

The ID with the stored counter

Throws:

Exception

storePendingCounter

long storePendingCounter(long queueID,
                         long pageID,
                         int inc)
                  throws Exception

Throws:

Exception

deleteIncrementRecord

void deleteIncrementRecord(long txID,
                           long recordID)
                    throws Exception

Throws:

Exception

deletePageCounter

void deletePageCounter(long txID,
                       long recordID)
                throws Exception

Throws:

Exception

deletePendingPageCounter

void deletePendingPageCounter(long txID,
                              long recordID)
                       throws Exception

Throws:

Exception

storePageCounterInc

long storePageCounterInc(long txID,
                         long queueID,
                         int add)
                  throws Exception

Returns:

the ID with the increment record

Throws:

Exception

storePageCounterInc

long storePageCounterInc(long queueID,
                         int add)
                  throws Exception

Returns:

the ID with the increment record

Throws:

Exception

getBindingsJournal

Journal getBindingsJournal()

Returns:

the bindings journal

getMessageJournal

Journal getMessageJournal()

Returns:

the message journal

startReplication

void startReplication(ReplicationManager replicationManager,
                      PagingManager pagingManager,
                      String nodeID,
                      boolean autoFailBack,
                      long initialReplicationSyncTimeout)
               throws Exception

Throws:

Exception

See Also:

JournalStorageManager.startReplication(org.apache.activemq.artemis.core.replication.ReplicationManager, org.apache.activemq.artemis.core.paging.PagingManager, String, boolean, long)

addToPage

boolean addToPage(PagingStore store,
                  ServerMessage msg,
                  Transaction tx,
                  RouteContextList listCtx)
           throws Exception

Write message to page if we are paging.

This is primarily a PagingStore call, but as with any other call writing persistent data, it must go through here. Both for the sake of replication, and also to ensure that it takes the locks (storage manager and pagingStore) in the right order. Avoiding thus the creation of dead-locks.

Returns:

true if we are paging and have handled the data, false if the data needs to be sent to the journal

Throws:

Exception

stopReplication

void stopReplication()

Stops the replication of data from the live to the backup.

Typical scenario is a broken connection.

addBytesToLargeMessage

void addBytesToLargeMessage(SequentialFile appendFile,
                            long messageID,
                            byte[] bytes)
                     throws Exception

Parameters:

appendFile -

messageID -

bytes -

Throws:

Exception

storeID

void storeID(long journalID,
             long id)
      throws Exception

Stores the id from IDManager.

Parameters:

journalID -

id -

Throws:

Exception

deleteID

void deleteID(long journalD)
       throws Exception

Throws:

Exception

readLock

void readLock()

Read lock the StorageManager. USE WITH CARE!

The main lock is used to write lock the whole manager when starting replication. Sub-systems, say Paging classes, that use locks of their own AND also write through the StorageManager MUST first read lock the storageManager before taking their own locks. Otherwise, we may dead-lock when starting replication sync.

readUnLock

void readUnLock()

Unlock the manager.

See Also:

readLock()

persistIdGenerator

void persistIdGenerator()

Closes the IDGenerator persisting the current record ID.

Effectively a "pre-stop" method. Necessary due to the "stop"-order at ActiveMQServerImpl

injectMonitor

void injectMonitor(FileStoreMonitor monitor)
            throws Exception

Throws:

Exception