org.apache.activemq.store.kahadaptor

Class KahaPersistenceAdapter

java.lang.Object

org.apache.activemq.store.kahadaptor.KahaPersistenceAdapter

All Implemented Interfaces:

BrokerServiceAware, Service, PersistenceAdapter

Direct Known Subclasses:

KahaReferenceStoreAdapter

public class KahaPersistenceAdapter
extends Object
implements PersistenceAdapter, BrokerServiceAware

Field Summary

Fields

Modifier and Type	Field and Description
protected ConcurrentHashMap<ActiveMQDestination,MessageStore>	messageStores
protected ConcurrentHashMap<ActiveMQQueue,MessageStore>	queues
protected ConcurrentHashMap<ActiveMQTopic,TopicMessageStore>	topics
protected KahaTransactionStore	transactionStore
protected OpenWireFormat	wireFormat

Constructor Summary

Constructors

Constructor and Description
KahaPersistenceAdapter()
KahaPersistenceAdapter(AtomicLong size)

Method Summary

Methods

Modifier and Type	Method and Description
void	beginTransaction(ConnectionContext context) This method starts a transaction on the persistent storage - which is nothing to do with JMS or XA transactions - its purely a mechanism to perform multiple writes to a persistent store in 1 transaction as a performance optimization.
void	checkpoint(boolean sync) checkpoint any
void	commitTransaction(ConnectionContext context) Commit a persistence transaction
protected Marshaller<Object>	createMessageMarshaller()
MessageStore	createQueueMessageStore(ActiveMQQueue destination) Factory method to create a new queue message store with the given destination name
protected Store	createStore()
TopicMessageStore	createTopicMessageStore(ActiveMQTopic destination) Factory method to create a new topic message store with the given destination name
TransactionStore	createTransactionStore() Factory method to create a new persistent prepared transaction store for XA recovery
void	deleteAllMessages() Delete's all the messages in the persistent store.
String	getBrokerName()
Set<ActiveMQDestination>	getDestinations() Returns a set of all the ActiveMQDestination objects that the persistence store is aware exist.
File	getDirectory()
long	getLastMessageBrokerSequenceId()
long	getLastProducerSequenceId(ProducerId id) return the last stored producer sequenceId for this producer Id used to suppress duplicate sends on failover reconnect at the transport when a reconnect occurs
protected ListContainer<TopicSubAck>	getListContainer(Object id, String containerName)
protected MapContainer<MessageId,Message>	getMapContainer(Object id, String containerName)
long	getMaxDataFileLength()
protected Store	getStore()
protected MapContainer	getSubsMapContainer(Object id, String containerName)
boolean	isPersistentIndex()
void	removeQueueMessageStore(ActiveMQQueue destination) Cleanup method to remove any state associated with the given destination
void	removeTopicMessageStore(ActiveMQTopic destination) Cleanup method to remove any state associated with the given destination
protected MessageStore	retrieveMessageStore(Object id)
void	rollbackTransaction(ConnectionContext context) Rollback a persistence transaction
void	setBrokerName(String brokerName) Set the name of the broker using the adapter
void	setBrokerService(BrokerService brokerService)
void	setDirectory(File directory) Set the directory where any data files should be created
void	setMaxDataFileLength(long maxDataFileLength) When set using Xbean, values of the form "20 Mb", "1024kb", and "1g" can be used
void	setPersistentIndex(boolean persistentIndex)
void	setUsageManager(SystemUsage usageManager)
long	size() A hint to return the size of the store on disk
void	start()
void	stop()
String	toString()

Methods inherited from class java.lang.Object

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

Field Detail

wireFormat

protected OpenWireFormat wireFormat

transactionStore

protected KahaTransactionStore transactionStore

topics

protected ConcurrentHashMap<ActiveMQTopic,TopicMessageStore> topics

queues

protected ConcurrentHashMap<ActiveMQQueue,MessageStore> queues

messageStores

protected ConcurrentHashMap<ActiveMQDestination,MessageStore> messageStores

Constructor Detail

KahaPersistenceAdapter

public KahaPersistenceAdapter(AtomicLong size)

KahaPersistenceAdapter

public KahaPersistenceAdapter()

Method Detail

getDestinations

public Set<ActiveMQDestination> getDestinations()

Description copied from interface: PersistenceAdapter

Returns a set of all the ActiveMQDestination objects that the persistence store is aware exist.

Specified by:

getDestinations in interface PersistenceAdapter

Returns:

active destinations

createQueueMessageStore

public MessageStore createQueueMessageStore(ActiveMQQueue destination)
                                     throws IOException

Description copied from interface: PersistenceAdapter

Factory method to create a new queue message store with the given destination name

Specified by:

createQueueMessageStore in interface PersistenceAdapter

Returns:

the message store

Throws:

IOException

createTopicMessageStore

public TopicMessageStore createTopicMessageStore(ActiveMQTopic destination)
                                          throws IOException

Description copied from interface: PersistenceAdapter

Factory method to create a new topic message store with the given destination name

Specified by:

createTopicMessageStore in interface PersistenceAdapter

Returns:

the topic message store

Throws:

IOException

removeQueueMessageStore

public void removeQueueMessageStore(ActiveMQQueue destination)

Cleanup method to remove any state associated with the given destination

Specified by:

removeQueueMessageStore in interface PersistenceAdapter

Parameters:

destination - Destination to forget

removeTopicMessageStore

public void removeTopicMessageStore(ActiveMQTopic destination)

Cleanup method to remove any state associated with the given destination

Specified by:

removeTopicMessageStore in interface PersistenceAdapter

Parameters:

destination - Destination to forget

retrieveMessageStore

protected MessageStore retrieveMessageStore(Object id)

createTransactionStore

public TransactionStore createTransactionStore()
                                        throws IOException

Description copied from interface: PersistenceAdapter

Factory method to create a new persistent prepared transaction store for XA recovery

Specified by:

createTransactionStore in interface PersistenceAdapter

Returns:

transaction store

Throws:

IOException

beginTransaction

public void beginTransaction(ConnectionContext context)

Description copied from interface: PersistenceAdapter

This method starts a transaction on the persistent storage - which is nothing to do with JMS or XA transactions - its purely a mechanism to perform multiple writes to a persistent store in 1 transaction as a performance optimization.

Typically one transaction will require one disk synchronization point and so for real high performance its usually faster to perform many writes within the same transaction to minimize latency caused by disk synchronization. This is especially true when using tools like Berkeley Db or embedded JDBC servers.

Specified by:

beginTransaction in interface PersistenceAdapter

commitTransaction

public void commitTransaction(ConnectionContext context)
                       throws IOException

Description copied from interface: PersistenceAdapter

Commit a persistence transaction

Specified by:

commitTransaction in interface PersistenceAdapter

Throws:

IOException

See Also:

PersistenceAdapter.beginTransaction(ConnectionContext context)

rollbackTransaction

public void rollbackTransaction(ConnectionContext context)

Description copied from interface: PersistenceAdapter

Rollback a persistence transaction

Specified by:

rollbackTransaction in interface PersistenceAdapter

See Also:

PersistenceAdapter.beginTransaction(ConnectionContext context)

start

public void start()
           throws Exception

Specified by:

start in interface Service

Throws:

Exception

stop

public void stop()
          throws Exception

Specified by:

stop in interface Service

Throws:

Exception

getLastMessageBrokerSequenceId

public long getLastMessageBrokerSequenceId()
                                    throws IOException

Specified by:

getLastMessageBrokerSequenceId in interface PersistenceAdapter

Returns:

last broker sequence

Throws:

IOException

deleteAllMessages

public void deleteAllMessages()
                       throws IOException

Description copied from interface: PersistenceAdapter

Delete's all the messages in the persistent store.

Specified by:

deleteAllMessages in interface PersistenceAdapter

Throws:

IOException

getMapContainer

protected MapContainer<MessageId,Message> getMapContainer(Object id,
                                              String containerName)
                                                   throws IOException

Throws:

IOException

getSubsMapContainer

protected MapContainer getSubsMapContainer(Object id,
                               String containerName)
                                    throws IOException

Throws:

IOException

createMessageMarshaller

protected Marshaller<Object> createMessageMarshaller()

getListContainer

protected ListContainer<TopicSubAck> getListContainer(Object id,
                                          String containerName)
                                               throws IOException

Throws:

IOException

setUsageManager

public void setUsageManager(SystemUsage usageManager)

Specified by:

setUsageManager in interface PersistenceAdapter

Parameters:

usageManager - The UsageManager that is controlling the broker's memory usage.

getMaxDataFileLength

public long getMaxDataFileLength()

Returns:

the maxDataFileLength

isPersistentIndex

public boolean isPersistentIndex()

setPersistentIndex

public void setPersistentIndex(boolean persistentIndex)

setMaxDataFileLength

public void setMaxDataFileLength(long maxDataFileLength)

When set using Xbean, values of the form "20 Mb", "1024kb", and "1g" can be used

getStore

protected final Store getStore()
                        throws IOException

Throws:

IOException

createStore

protected final Store createStore()
                           throws IOException

Throws:

IOException

toString

public String toString()

Overrides:

toString in class Object

setBrokerName

public void setBrokerName(String brokerName)

Description copied from interface: PersistenceAdapter

Set the name of the broker using the adapter

Specified by:

setBrokerName in interface PersistenceAdapter

getBrokerName

public String getBrokerName()

getDirectory

public File getDirectory()

Specified by:

getDirectory in interface PersistenceAdapter

Returns:

the directory used by the persistence adaptor

setDirectory

public void setDirectory(File directory)

Description copied from interface: PersistenceAdapter

Set the directory where any data files should be created

Specified by:

setDirectory in interface PersistenceAdapter

checkpoint

public void checkpoint(boolean sync)
                throws IOException

Description copied from interface: PersistenceAdapter

checkpoint any

Specified by:

checkpoint in interface PersistenceAdapter

Throws:

IOException

size

public long size()

Description copied from interface: PersistenceAdapter

A hint to return the size of the store on disk

Specified by:

size in interface PersistenceAdapter

Returns:

disk space used in bytes of 0 if not implemented

setBrokerService

public void setBrokerService(BrokerService brokerService)

Specified by:

setBrokerService in interface BrokerServiceAware

getLastProducerSequenceId

public long getLastProducerSequenceId(ProducerId id)

Description copied from interface: PersistenceAdapter

return the last stored producer sequenceId for this producer Id used to suppress duplicate sends on failover reconnect at the transport when a reconnect occurs

Specified by:

getLastProducerSequenceId in interface PersistenceAdapter

Parameters:

id - the producerId to find a sequenceId for

Returns:

the last stored sequence id or -1 if no suppression needed