org.activemq.store.journal
Class JournalPersistenceAdapter

java.lang.Object
  org.activemq.store.journal.JournalPersistenceAdapter

All Implemented Interfaces:

org.activeio.journal.JournalEventListener, Service, PersistenceAdapter

public class JournalPersistenceAdapter
extends Object
implements PersistenceAdapter, org.activeio.journal.JournalEventListener

An implementation of PersistenceAdapter designed for use with a Journal and then checkpointing asynchronously on a timeout with some other long term persistent storage.

Version:

$Revision: 1.1 $

Field Summary

static String	DEFAULT_JOURNAL_TYPE
static String	HOWL_JOURNAL_TYPE

Constructor Summary

JournalPersistenceAdapter()

JournalPersistenceAdapter(File directory, PersistenceAdapter longTermPersistence)

Method Summary

void	beginTransaction() 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 optimisation.
void	checkpoint(boolean sync) When we checkpoint we move all the journaled data to long term storage.
void	commitTransaction() Commit a persistence transaction
protected org.activeio.journal.Journal	createJournal()
MessageStore	createQueueMessageStore(String destinationName) Factory method to create a new queue message store with the given destination name
protected JMSException	createRecoveryFailedException(Exception e)
TopicMessageStore	createTopicMessageStore(String destinationName) 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
protected JMSException	createWriteException(Packet packet, Exception e)
protected JMSException	createWriteException(String command, Exception e)
boolean	deadLetterAlreadySent(long seq, boolean useLocking) Verifies if a dead letter has already been sent for a message
long	getCheckpointInterval()
ClockDaemon	getClockDaemon()
File	getDirectory()
Map	getInitialDestinations() Returns a map, indexed by String name, of all the Destination objects active on startup.
String	getJournalType()
int	getLogFileCount()
int	getLogFileSize()
PersistenceAdapter	getLongTermPersistence()
JournalTransactionStore	getTransactionStore()
WireFormat	getWireFormat()
void	overflowNotification(org.activeio.journal.RecordLocation safeLocation) The Journal give us a call back so that we can move old data out of the journal.
Packet	readPacket(org.activeio.journal.RecordLocation location)
void	rollbackTransaction() Rollback a persistence transaction
void	setCheckpointInterval(long checkpointInterval)
void	setClockDaemon(ClockDaemon clockDaemon)
void	setDirectory(File directory)
void	setJournalType(String journalType)
void	setLogFileCount(int logFileCount)
void	setLogFileSize(int logFileSize)
void	setLongTermPersistence(PersistenceAdapter longTermPersistence)
void	start() Called to start the service
void	stop() Called to shutdown the service
org.activeio.journal.RecordLocation	writeCommand(String command, boolean sync)
org.activeio.journal.RecordLocation	writePacket(String destination, Packet packet, boolean sync)
org.activeio.journal.RecordLocation	writePacket(String destinationName, String subscription, MessageIdentity messageIdentity, boolean sync)
org.activeio.journal.RecordLocation	writeTxCommand(TxCommand command, boolean sync)

Methods inherited from class java.lang.Object

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

Field Detail

DEFAULT_JOURNAL_TYPE

public static final String DEFAULT_JOURNAL_TYPE

See Also:

Constant Field Values

HOWL_JOURNAL_TYPE

public static final String HOWL_JOURNAL_TYPE

See Also:

Constant Field Values

Constructor Detail

JournalPersistenceAdapter

public JournalPersistenceAdapter()

JournalPersistenceAdapter

public JournalPersistenceAdapter(File directory,
                                 PersistenceAdapter longTermPersistence)
                          throws IOException

Throws:

IOException

Method Detail

getInitialDestinations

public Map getInitialDestinations()

Description copied from interface: PersistenceAdapter

Returns a map, indexed by String name, of all the Destination objects active on startup.

Specified by:

getInitialDestinations in interface PersistenceAdapter

Returns:

createQueueMessageStore

public MessageStore createQueueMessageStore(String destinationName)
                                     throws JMSException

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

Throws:

JMSException

createTopicMessageStore

public TopicMessageStore createTopicMessageStore(String destinationName)
                                          throws JMSException

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

Throws:

JMSException

createTransactionStore

public TransactionStore createTransactionStore()
                                        throws JMSException

Description copied from interface: PersistenceAdapter

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

Specified by:

createTransactionStore in interface PersistenceAdapter

Throws:

JMSException

beginTransaction

public void beginTransaction()
                      throws JMSException

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 optimisation.

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 minimise 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

Throws:

JMSException

commitTransaction

public void commitTransaction()
                       throws JMSException

Description copied from interface: PersistenceAdapter

Commit a persistence transaction

Specified by:

commitTransaction in interface PersistenceAdapter

Throws:

JMSException

See Also:

PersistenceAdapter.beginTransaction()

rollbackTransaction

public void rollbackTransaction()

Description copied from interface: PersistenceAdapter

Rollback a persistence transaction

Specified by:

rollbackTransaction in interface PersistenceAdapter

See Also:

PersistenceAdapter.beginTransaction()

start

public void start()
           throws JMSException

Description copied from interface: Service

Called to start the service

Specified by:

start in interface Service

Throws:

JMSException

stop

public void stop()
          throws JMSException

Description copied from interface: Service

Called to shutdown the service

Specified by:

stop in interface Service

Throws:

JMSException

getLongTermPersistence

public PersistenceAdapter getLongTermPersistence()

setLongTermPersistence

public void setLongTermPersistence(PersistenceAdapter longTermPersistence)

getDirectory

public File getDirectory()

Returns:

Returns the directory.

setDirectory

public void setDirectory(File directory)

Parameters:

directory - The directory to set.

getWireFormat

public WireFormat getWireFormat()

Returns:

Returns the wireFormat.

getJournalType

public String getJournalType()

setJournalType

public void setJournalType(String journalType)

createJournal

protected org.activeio.journal.Journal createJournal()
                                              throws IOException

Throws:

IOException

overflowNotification

public void overflowNotification(org.activeio.journal.RecordLocation safeLocation)

The Journal give us a call back so that we can move old data out of the journal. Taking a checkpoint does this for us.

Specified by:

overflowNotification in interface org.activeio.journal.JournalEventListener

See Also:

org.activemq.journal.JournalEventListener#overflowNotification(org.activemq.journal.RecordLocation)

checkpoint

public void checkpoint(boolean sync)

When we checkpoint we move all the journaled data to long term storage.

Parameters:

b -

writePacket

public org.activeio.journal.RecordLocation writePacket(String destination,
                                                       Packet packet,
                                                       boolean sync)
                                                throws JMSException

Parameters:

destinationName -

message -

sync -

Throws:

JMSException

writeCommand

public org.activeio.journal.RecordLocation writeCommand(String command,
                                                        boolean sync)
                                                 throws JMSException

Parameters:

destinationName -

message -

sync -

Throws:

JMSException

readPacket

public Packet readPacket(org.activeio.journal.RecordLocation location)
                  throws JMSException

Parameters:

location -

Returns:

Throws:

JMSException

createWriteException

protected JMSException createWriteException(Packet packet,
                                            Exception e)

createWriteException

protected JMSException createWriteException(String command,
                                            Exception e)

createRecoveryFailedException

protected JMSException createRecoveryFailedException(Exception e)

getClockDaemon

public ClockDaemon getClockDaemon()

setClockDaemon

public void setClockDaemon(ClockDaemon clockDaemon)

writeTxCommand

public org.activeio.journal.RecordLocation writeTxCommand(TxCommand command,
                                                          boolean sync)
                                                   throws XAException

Parameters:

xid -

Returns:

Throws:

XAException

writePacket

public org.activeio.journal.RecordLocation writePacket(String destinationName,
                                                       String subscription,
                                                       MessageIdentity messageIdentity,
                                                       boolean sync)
                                                throws JMSException

Parameters:

destinationName -

persistentKey -

messageIdentity -

b -

Returns:

Throws:

JMSException

getTransactionStore

public JournalTransactionStore getTransactionStore()

getLogFileCount

public int getLogFileCount()

setLogFileCount

public void setLogFileCount(int logFileCount)

getLogFileSize

public int getLogFileSize()

setLogFileSize

public void setLogFileSize(int logFileSize)

deadLetterAlreadySent

public boolean deadLetterAlreadySent(long seq,
                                     boolean useLocking)

Verifies if a dead letter has already been sent for a message

Specified by:

deadLetterAlreadySent in interface PersistenceAdapter

Parameters:

seq -

useLocking - to prevent concurrency/dups

Returns:

getCheckpointInterval

public long getCheckpointInterval()

setCheckpointInterval

public void setCheckpointInterval(long checkpointInterval)