org.apache.activemq.store.kahadaptor
Class KahaTopicMessageStore

java.lang.Object
  org.apache.activemq.store.kahadaptor.KahaTopicMessageStore

All Implemented Interfaces:

Service, MessageStore, TopicMessageStore

public class KahaTopicMessageStore
extends java.lang.Object
implements TopicMessageStore

Version:

$Revision: 1.5 $

Constructor Summary

KahaTopicMessageStore(Store store, ListContainer messageContainer, ListContainer ackContainer, MapContainer subsContainer, ActiveMQDestination destination)

Method Summary

void	acknowledge(ConnectionContext context, java.lang.String clientId, java.lang.String subscriptionName, MessageId messageId) Stores the last acknowledged messgeID for the given subscription so that we can recover and commence dispatching messages from the last checkpoint
void	addMessage(ConnectionContext context, Message message) Adds a message to the message store
void	addMessageReference(ConnectionContext context, MessageId messageId, long expirationTime, java.lang.String messageRef) Adds a message reference to the message store
void	addSubsciption(java.lang.String clientId, java.lang.String subscriptionName, java.lang.String selector, boolean retroactive) Inserts the subscriber info due to a subscription change If this is a new subscription and the retroactive is false, then the last message sent to the topic should be set as the last message acknowledged by they new subscription.
protected void	addSubscriberMessageContainer(java.lang.Object key)
void	delete()
void	deleteSubscription(java.lang.String clientId, java.lang.String subscriptionName)
SubscriptionInfo[]	getAllSubscriptions() Lists all the durable subscirptions for a given destination.
ActiveMQDestination	getDestination() The destination that the message store is holding messages for.
Message	getMessage(MessageId identity) Looks up a message using either the String messageID or the messageNumber.
int	getMessageCount(java.lang.String clientId, java.lang.String subscriberName) Get the number of messages ready to deliver from the store to a durable subscriber
java.lang.String	getMessageReference(MessageId identity) Looks up a message using either the String messageID or the messageNumber.
MessageId	getNextMessageIdToDeliver(java.lang.String clientId, java.lang.String subscriptionName, MessageId id) Get the next messageId to deliver to a subscriber after the MessageId provided
MessageId	getPreviousMessageIdToDeliver(java.lang.String clientId, java.lang.String subscriptionName, MessageId id) Get the previous messageId to deliver to a subscriber before the MessageId provided
protected java.lang.String	getSubscriptionKey(java.lang.String clientId, java.lang.String subscriberName)
SubscriptionInfo	lookupSubscription(java.lang.String clientId, java.lang.String subscriptionName) Finds the subscriber entry for the given consumer info
void	recover(MessageRecoveryListener listener) Recover any messages to be delivered.
void	recoverNextMessages(java.lang.String clientId, java.lang.String subscriptionName, MessageId lastMessageId, int maxReturned, MessageRecoveryListener listener) For an active subscription - retrieve messages from the store for the subscriber after the lastMessageId messageId
void	recoverSubscription(java.lang.String clientId, java.lang.String subscriptionName, MessageRecoveryListener listener) For the new subscription find the last acknowledged message ID and then find any new messages since then and dispatch them to the subscription.
void	removeAllMessages(ConnectionContext context) Removes all the messages from the message store.
void	removeMessage(ConnectionContext context, MessageAck ack) Removes a message from the message store.
void	resetBatching(java.lang.String clientId, java.lang.String subscriptionName, MessageId nextToDispatch) A hint to the Store to reset any batching state for a durable subsriber
void	setUsageManager(UsageManager usageManager)
void	start()
void	stop()

Methods inherited from class java.lang.Object

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

Constructor Detail

KahaTopicMessageStore

public KahaTopicMessageStore(Store store,
                             ListContainer messageContainer,
                             ListContainer ackContainer,
                             MapContainer subsContainer,
                             ActiveMQDestination destination)
                      throws java.io.IOException

Throws:

java.io.IOException

Method Detail

addMessage

public void addMessage(ConnectionContext context,
                       Message message)
                throws java.io.IOException

Description copied from interface: MessageStore

Adds a message to the message store

Specified by:

addMessage in interface MessageStore

Parameters:

context - TODO

Throws:

java.io.IOException

acknowledge

public void acknowledge(ConnectionContext context,
                        java.lang.String clientId,
                        java.lang.String subscriptionName,
                        MessageId messageId)
                 throws java.io.IOException

Description copied from interface: TopicMessageStore

Stores the last acknowledged messgeID for the given subscription so that we can recover and commence dispatching messages from the last checkpoint

Specified by:

acknowledge in interface TopicMessageStore

Throws:

java.io.IOException

lookupSubscription

public SubscriptionInfo lookupSubscription(java.lang.String clientId,
                                           java.lang.String subscriptionName)
                                    throws java.io.IOException

Description copied from interface: TopicMessageStore

Finds the subscriber entry for the given consumer info

Specified by:

lookupSubscription in interface TopicMessageStore

Returns:

the SubscriptionInfo

Throws:

java.io.IOException

addSubsciption

public void addSubsciption(java.lang.String clientId,
                           java.lang.String subscriptionName,
                           java.lang.String selector,
                           boolean retroactive)
                    throws java.io.IOException

Description copied from interface: TopicMessageStore

Inserts the subscriber info due to a subscription change

If this is a new subscription and the retroactive is false, then the last message sent to the topic should be set as the last message acknowledged by they new subscription. Otherwise, if retroactive is true, then create the subscription without it having an acknowledged message so that on recovery, all message recorded for the topic get replayed.

Specified by:

addSubsciption in interface TopicMessageStore

Throws:

java.io.IOException

deleteSubscription

public void deleteSubscription(java.lang.String clientId,
                               java.lang.String subscriptionName)

Specified by:

deleteSubscription in interface TopicMessageStore

recoverSubscription

public void recoverSubscription(java.lang.String clientId,
                                java.lang.String subscriptionName,
                                MessageRecoveryListener listener)
                         throws java.lang.Exception

Description copied from interface: TopicMessageStore

For the new subscription find the last acknowledged message ID and then find any new messages since then and dispatch them to the subscription.

e.g. if we dispatched some messages to a new durable topic subscriber, then went down before acknowledging any messages, we need to know the correct point from which to recover from.

Specified by:

recoverSubscription in interface TopicMessageStore

Throws:

java.lang.Exception

recoverNextMessages

public void recoverNextMessages(java.lang.String clientId,
                                java.lang.String subscriptionName,
                                MessageId lastMessageId,
                                int maxReturned,
                                MessageRecoveryListener listener)
                         throws java.lang.Exception

Description copied from interface: TopicMessageStore

For an active subscription - retrieve messages from the store for the subscriber after the lastMessageId messageId

Specified by:

recoverNextMessages in interface TopicMessageStore

Throws:

java.lang.Exception

delete

public void delete()

getAllSubscriptions

public SubscriptionInfo[] getAllSubscriptions()
                                       throws java.io.IOException

Description copied from interface: TopicMessageStore

Lists all the durable subscirptions for a given destination.

Specified by:

getAllSubscriptions in interface TopicMessageStore

Returns:

an array SubscriptionInfos

Throws:

java.io.IOException

getSubscriptionKey

protected java.lang.String getSubscriptionKey(java.lang.String clientId,
                                              java.lang.String subscriberName)

addSubscriberMessageContainer

protected void addSubscriberMessageContainer(java.lang.Object key)
                                      throws java.io.IOException

Throws:

java.io.IOException

getMessageCount

public int getMessageCount(java.lang.String clientId,
                           java.lang.String subscriberName)
                    throws java.io.IOException

Description copied from interface: TopicMessageStore

Get the number of messages ready to deliver from the store to a durable subscriber

Specified by:

getMessageCount in interface TopicMessageStore

Returns:

the outstanding message count

Throws:

java.io.IOException

addMessageReference

public void addMessageReference(ConnectionContext context,
                                MessageId messageId,
                                long expirationTime,
                                java.lang.String messageRef)
                         throws java.io.IOException

Description copied from interface: MessageStore

Adds a message reference to the message store

Specified by:

addMessageReference in interface MessageStore

Parameters:

context -

messageId -

expirationTime -

messageRef -

Throws:

java.io.IOException

See Also:

MessageStore.addMessageReference(org.apache.activemq.broker.ConnectionContext, org.apache.activemq.command.MessageId, long, java.lang.String)

getDestination

public ActiveMQDestination getDestination()

Description copied from interface: MessageStore

The destination that the message store is holding messages for.

Specified by:

getDestination in interface MessageStore

Returns:

the destination

See Also:

MessageStore.getDestination()

getMessage

public Message getMessage(MessageId identity)
                   throws java.io.IOException

Description copied from interface: MessageStore

Looks up a message using either the String messageID or the messageNumber. Implementations are encouraged to fill in the missing key if its easy to do so.

Specified by:

getMessage in interface MessageStore

Parameters:

identity -

Returns:

the Message

Throws:

java.io.IOException

See Also:

MessageStore.getMessage(org.apache.activemq.command.MessageId)

getMessageReference

public java.lang.String getMessageReference(MessageId identity)
                                     throws java.io.IOException

Description copied from interface: MessageStore

Looks up a message using either the String messageID or the messageNumber. Implementations are encouraged to fill in the missing key if its easy to do so.

Specified by:

getMessageReference in interface MessageStore

Parameters:

identity -

Returns:

String

Throws:

java.io.IOException

See Also:

MessageStore.getMessageReference(org.apache.activemq.command.MessageId)

recover

public void recover(MessageRecoveryListener listener)
             throws java.lang.Exception

Description copied from interface: MessageStore

Recover any messages to be delivered.

Specified by:

recover in interface MessageStore

Throws:

java.lang.Exception

See Also:

MessageStore.recover(org.apache.activemq.store.MessageRecoveryListener)

removeAllMessages

public void removeAllMessages(ConnectionContext context)
                       throws java.io.IOException

Description copied from interface: MessageStore

Removes all the messages from the message store.

Specified by:

removeAllMessages in interface MessageStore

Parameters:

context -

Throws:

java.io.IOException

See Also:

MessageStore.removeAllMessages(org.apache.activemq.broker.ConnectionContext)

removeMessage

public void removeMessage(ConnectionContext context,
                          MessageAck ack)
                   throws java.io.IOException

Description copied from interface: MessageStore

Removes a message from the message store.

Specified by:

removeMessage in interface MessageStore

Parameters:

context -

ack -

Throws:

java.io.IOException

See Also:

MessageStore.removeMessage(org.apache.activemq.broker.ConnectionContext, org.apache.activemq.command.MessageAck)

setUsageManager

public void setUsageManager(UsageManager usageManager)

Specified by:

setUsageManager in interface MessageStore

Parameters:

usageManager -

See Also:

MessageStore.setUsageManager(org.apache.activemq.memory.UsageManager)

start

public void start()
           throws java.lang.Exception

Specified by:

start in interface Service

Throws:

java.lang.Exception

See Also:

Service.start()

stop

public void stop()
          throws java.lang.Exception

Specified by:

stop in interface Service

Throws:

java.lang.Exception

See Also:

Service.stop()

resetBatching

public void resetBatching(java.lang.String clientId,
                          java.lang.String subscriptionName,
                          MessageId nextToDispatch)

Description copied from interface: TopicMessageStore

A hint to the Store to reset any batching state for a durable subsriber

Specified by:

resetBatching in interface TopicMessageStore

Parameters:

clientId -

subscriptionName -

See Also:

org.apache.activemq.store.TopicMessageStore#resetBatching(java.lang.String, java.lang.String)

getNextMessageIdToDeliver

public MessageId getNextMessageIdToDeliver(java.lang.String clientId,
                                           java.lang.String subscriptionName,
                                           MessageId id)
                                    throws java.io.IOException

Description copied from interface: TopicMessageStore

Get the next messageId to deliver to a subscriber after the MessageId provided

Specified by:

getNextMessageIdToDeliver in interface TopicMessageStore

Parameters:

clientId -

subscriptionName -

id -

Returns:

next messageId

Throws:

java.io.IOException

See Also:

TopicMessageStore.getNextMessageIdToDeliver(java.lang.String, java.lang.String, org.apache.activemq.command.MessageId)

getPreviousMessageIdToDeliver

public MessageId getPreviousMessageIdToDeliver(java.lang.String clientId,
                                               java.lang.String subscriptionName,
                                               MessageId id)
                                        throws java.io.IOException

Description copied from interface: TopicMessageStore

Get the previous messageId to deliver to a subscriber before the MessageId provided

Specified by:

getPreviousMessageIdToDeliver in interface TopicMessageStore

Parameters:

clientId -

subscriptionName -

id -

Returns:

previous messageId

Throws:

java.io.IOException

See Also:

TopicMessageStore.getPreviousMessageIdToDeliver(java.lang.String, java.lang.String, org.apache.activemq.command.MessageId)