public class KahaPersistenceAdapter extends Object implements PersistenceAdapter, BrokerServiceAware
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 and Description |
---|
KahaPersistenceAdapter() |
KahaPersistenceAdapter(AtomicLong size) |
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() |
protected OpenWireFormat wireFormat
protected KahaTransactionStore transactionStore
protected ConcurrentHashMap<ActiveMQTopic,TopicMessageStore> topics
protected ConcurrentHashMap<ActiveMQQueue,MessageStore> queues
protected ConcurrentHashMap<ActiveMQDestination,MessageStore> messageStores
public KahaPersistenceAdapter(AtomicLong size)
public KahaPersistenceAdapter()
public Set<ActiveMQDestination> getDestinations()
PersistenceAdapter
ActiveMQDestination
objects that the persistence store is aware exist.getDestinations
in interface PersistenceAdapter
public MessageStore createQueueMessageStore(ActiveMQQueue destination) throws IOException
PersistenceAdapter
createQueueMessageStore
in interface PersistenceAdapter
IOException
public TopicMessageStore createTopicMessageStore(ActiveMQTopic destination) throws IOException
PersistenceAdapter
createTopicMessageStore
in interface PersistenceAdapter
IOException
public void removeQueueMessageStore(ActiveMQQueue destination)
removeQueueMessageStore
in interface PersistenceAdapter
destination
- Destination to forgetpublic void removeTopicMessageStore(ActiveMQTopic destination)
removeTopicMessageStore
in interface PersistenceAdapter
destination
- Destination to forgetprotected MessageStore retrieveMessageStore(Object id)
public TransactionStore createTransactionStore() throws IOException
PersistenceAdapter
createTransactionStore
in interface PersistenceAdapter
IOException
public void beginTransaction(ConnectionContext context)
PersistenceAdapter
beginTransaction
in interface PersistenceAdapter
public void commitTransaction(ConnectionContext context) throws IOException
PersistenceAdapter
commitTransaction
in interface PersistenceAdapter
IOException
PersistenceAdapter.beginTransaction(ConnectionContext context)
public void rollbackTransaction(ConnectionContext context)
PersistenceAdapter
rollbackTransaction
in interface PersistenceAdapter
PersistenceAdapter.beginTransaction(ConnectionContext context)
public void start() throws Exception
public long getLastMessageBrokerSequenceId() throws IOException
getLastMessageBrokerSequenceId
in interface PersistenceAdapter
IOException
public void deleteAllMessages() throws IOException
PersistenceAdapter
deleteAllMessages
in interface PersistenceAdapter
IOException
protected MapContainer<MessageId,Message> getMapContainer(Object id, String containerName) throws IOException
IOException
protected MapContainer getSubsMapContainer(Object id, String containerName) throws IOException
IOException
protected Marshaller<Object> createMessageMarshaller()
protected ListContainer<TopicSubAck> getListContainer(Object id, String containerName) throws IOException
IOException
public void setUsageManager(SystemUsage usageManager)
setUsageManager
in interface PersistenceAdapter
usageManager
- The UsageManager that is controlling the broker's
memory usage.public long getMaxDataFileLength()
public boolean isPersistentIndex()
public void setPersistentIndex(boolean persistentIndex)
public void setMaxDataFileLength(long maxDataFileLength)
protected final Store getStore() throws IOException
IOException
protected final Store createStore() throws IOException
IOException
public void setBrokerName(String brokerName)
PersistenceAdapter
setBrokerName
in interface PersistenceAdapter
public String getBrokerName()
public File getDirectory()
getDirectory
in interface PersistenceAdapter
public void setDirectory(File directory)
PersistenceAdapter
setDirectory
in interface PersistenceAdapter
public void checkpoint(boolean sync) throws IOException
PersistenceAdapter
checkpoint
in interface PersistenceAdapter
IOException
public long size()
PersistenceAdapter
size
in interface PersistenceAdapter
public void setBrokerService(BrokerService brokerService)
setBrokerService
in interface BrokerServiceAware
public long getLastProducerSequenceId(ProducerId id)
PersistenceAdapter
getLastProducerSequenceId
in interface PersistenceAdapter
id
- the producerId to find a sequenceId forCopyright © 2005-2013 The Apache Software Foundation. All Rights Reserved.