org.codehaus.activemq.service.impl
Class SubscriptionImpl

java.lang.Object
  org.codehaus.activemq.service.impl.SubscriptionImpl

All Implemented Interfaces:

Subscription

Direct Known Subclasses:

DurableTopicSubscription

public class SubscriptionImpl

extends java.lang.Object

implements Subscription

A Subscription holds messages to be dispatched to a a Client Consumer

Version:

$Revision: 1.17 $

Field Summary

protected Dispatcher	dispatch
(package private) Filter	filter
(package private) QueueList	messagePtrs
protected SynchronizedInt	unconsumedMessagesDispatched

Constructor Summary

SubscriptionImpl(Dispatcher dispatcher, ConsumerInfo info, Filter filter)
Create a Subscription object that holds messages to be dispatched to a Consumer

Method Summary

void	addMessage(MessageContainer container, ActiveMQMessage message) If the Subscription is a target for the message, the subscription will add a reference to the message and register an interest in the message to the container
void	clear() Called when the Subscription is discarded
protected boolean	clientIDsEqual(ActiveMQMessage message)
protected void	doMessageConsume(MessageAck ack, boolean remove) Consume a message.
protected static boolean	equal(java.lang.Object left, java.lang.Object right)
java.lang.String	getClientId()
java.lang.String	getConsumerId()
int	getConsumerNumber()
ActiveMQDestination	getDestination()
Filter	getFilter()
MessageIdentity	getLastMessageIdentity() Retreives the messageIdentity of the last message sent to this Queue based Subscription
ActiveMQMessage[]	getMessagesToDispatch() Retrieve messages to dispatch
protected ActiveMQMessage[]	getMessagesWithPrefetch()
java.lang.String	getPersistentKey() Returns the persistent key used to uniquely identify this durable topic subscription
java.lang.String	getSelector()
java.lang.String	getSubscriberName()
boolean	isActive()
boolean	isAtPrefetchLimit() Indicates the Subscription it's reached it's pre-fetch limit
boolean	isBrowser() Indicates the consumer is a browser only
boolean	isDurableTopic() Indicates the Subscriber is a Durable Subscriber
boolean	isNoLocal()
boolean	isReadyToDispatch() Indicates if this Subscription has more messages to send to the Consumer
boolean	isSameDurableSubscription(ConsumerInfo info) Checks if this subscription is a duplicate durable subscription of the given consumer info
boolean	isTarget(ActiveMQMessage message) determines if the Subscription is interested in the message
boolean	isWildcard()
void	messageConsumed(MessageAck ack) Indicates a message has been delivered to a MessageConsumer
void	onAcknowledgeTransactedMessageBeforeCommit(MessageAck ack) We have not yet committed and so the message acknowledgement has not really occurred yet but we need to let the dispatcher know that we can commence dispatching more messages to the client.
void	redeliverMessage(MessageContainer container, MessageAck ack) Forces the given message to be redelivered
void	reset() Called when an active subscriber has closed.
void	setActive(boolean active) set the state of the Subscription
void	setActiveConsumer(ConsumerInfo info) Set the active consumer info
void	setClientId(java.lang.String clientId)
void	setFilter(Filter filter)
void	setLastMessageIdentifier(MessageIdentity messageIdentity) Used for a Queue based Subscription to set the last acknowledged message ID
void	setNoLocal(boolean noLocal)
void	setSubscriberName(java.lang.String subscriberName)
java.lang.String	toString()

Methods inherited from class java.lang.Object

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

Field Detail

dispatch

protected Dispatcher dispatch

filter

Filter filter

unconsumedMessagesDispatched

protected SynchronizedInt unconsumedMessagesDispatched

messagePtrs

QueueList messagePtrs

Constructor Detail

SubscriptionImpl

public SubscriptionImpl(Dispatcher dispatcher,
                        ConsumerInfo info,
                        Filter filter)

Create a Subscription object that holds messages to be dispatched to a Consumer

Parameters:

dispatcher -

info -

filter -

Method Detail

setActiveConsumer

public void setActiveConsumer(ConsumerInfo info)

Set the active consumer info

Specified by:

setActiveConsumer in interface Subscription

Parameters:

info -

toString

public java.lang.String toString()

Returns:

pretty print of the Subscription

clear

public void clear()
           throws JMSException

Called when the Subscription is discarded

Specified by:

clear in interface Subscription

Throws:

JMSException

reset

public void reset()
           throws JMSException

Called when an active subscriber has closed. This resets all MessagePtrs

Specified by:

reset in interface Subscription

Throws:

JMSException

getClientId

public java.lang.String getClientId()

Specified by:

getClientId in interface Subscription

Returns:

Returns the clientId.

setClientId

public void setClientId(java.lang.String clientId)

Parameters:

clientId - The clientId to set.

getFilter

public Filter getFilter()

Returns:

Returns the filter.

setFilter

public void setFilter(Filter filter)

Parameters:

filter - The filter to set.

isWildcard

public boolean isWildcard()

Specified by:

isWildcard in interface Subscription

getPersistentKey

public java.lang.String getPersistentKey()

Description copied from interface: Subscription

Returns the persistent key used to uniquely identify this durable topic subscription

Specified by:

getPersistentKey in interface Subscription

Returns:

isSameDurableSubscription

public boolean isSameDurableSubscription(ConsumerInfo info)
                                  throws JMSException

Description copied from interface: Subscription

Checks if this subscription is a duplicate durable subscription of the given consumer info

Specified by:

isSameDurableSubscription in interface Subscription

Parameters:

info -

Returns:

true if this subscription is a durable topic subscription and the clientID and consumer names match

Throws:

JMSException

isNoLocal

public boolean isNoLocal()

Returns:

Returns the noLocal.

setNoLocal

public void setNoLocal(boolean noLocal)

Parameters:

noLocal - The noLocal to set.

getSubscriberName

public java.lang.String getSubscriberName()

Specified by:

getSubscriberName in interface Subscription

Returns:

Returns the subscriberName.

setSubscriberName

public void setSubscriberName(java.lang.String subscriberName)

Parameters:

subscriberName - The subscriberName to set.

isTarget

public boolean isTarget(ActiveMQMessage message)
                 throws JMSException

determines if the Subscription is interested in the message

Specified by:

isTarget in interface Subscription

Parameters:

message -

Returns:

true if this Subscription will accept the message

Throws:

JMSException

addMessage

public void addMessage(MessageContainer container,
                       ActiveMQMessage message)
                throws JMSException

If the Subscription is a target for the message, the subscription will add a reference to the message and register an interest in the message to the container

Specified by:

addMessage in interface Subscription

Parameters:

container -

message -

Throws:

JMSException

messageConsumed

public void messageConsumed(MessageAck ack)
                     throws JMSException

Indicates a message has been delivered to a MessageConsumer

Specified by:

messageConsumed in interface Subscription

Parameters:

ack -

Throws:

JMSException

onAcknowledgeTransactedMessageBeforeCommit

public void onAcknowledgeTransactedMessageBeforeCommit(MessageAck ack)
                                                throws JMSException

Description copied from interface: Subscription

We have not yet committed and so the message acknowledgement has not really occurred yet but we need to let the dispatcher know that we can commence dispatching more messages to the client. This is so that we can have a prefetch value of 1 yet we can still consume 1000 messages inside a transaction, with all the acks coming after the commit() on the client.

Specified by:

onAcknowledgeTransactedMessageBeforeCommit in interface Subscription

Parameters:

ack -

Throws:

JMSException

redeliverMessage

public void redeliverMessage(MessageContainer container,
                             MessageAck ack)
                      throws JMSException

Description copied from interface: Subscription

Forces the given message to be redelivered

Specified by:

redeliverMessage in interface Subscription

Parameters:

container -

ack -

Throws:

JMSException

getMessagesToDispatch

public ActiveMQMessage[] getMessagesToDispatch()
                                        throws JMSException

Retrieve messages to dispatch

Specified by:

getMessagesToDispatch in interface Subscription

Returns:

the messages to dispatch

Throws:

JMSException

getMessagesWithPrefetch

protected ActiveMQMessage[] getMessagesWithPrefetch()
                                             throws JMSException

Throws:

JMSException

isAtPrefetchLimit

public boolean isAtPrefetchLimit()
                          throws JMSException

Indicates the Subscription it's reached it's pre-fetch limit

Specified by:

isAtPrefetchLimit in interface Subscription

Returns:

true/false

Throws:

JMSException

isReadyToDispatch

public boolean isReadyToDispatch()
                          throws JMSException

Indicates if this Subscription has more messages to send to the Consumer

Specified by:

isReadyToDispatch in interface Subscription

Returns:

true if more messages available to dispatch

Throws:

JMSException

getDestination

public ActiveMQDestination getDestination()

Specified by:

getDestination in interface Subscription

Returns:

Returns the destination.

getSelector

public java.lang.String getSelector()

Specified by:

getSelector in interface Subscription

Returns:

Returns the selector.

isActive

public boolean isActive()

Specified by:

isActive in interface Subscription

Returns:

Returns the active.

setActive

public void setActive(boolean active)
               throws JMSException

Description copied from interface: Subscription

set the state of the Subscription

Specified by:

setActive in interface Subscription

Parameters:

active - The active to set.

Throws:

JMSException

getConsumerNumber

public int getConsumerNumber()

Specified by:

getConsumerNumber in interface Subscription

Returns:

Returns the consumerNumber.

getConsumerId

public java.lang.String getConsumerId()

Specified by:

getConsumerId in interface Subscription

Returns:

the consumer Id for the active consumer

isDurableTopic

public boolean isDurableTopic()
                       throws JMSException

Indicates the Subscriber is a Durable Subscriber

Specified by:

isDurableTopic in interface Subscription

Returns:

true if the subscriber is a durable topic

Throws:

JMSException

isBrowser

public boolean isBrowser()
                  throws JMSException

Indicates the consumer is a browser only

Specified by:

isBrowser in interface Subscription

Returns:

true if a Browser

Throws:

JMSException

getLastMessageIdentity

public MessageIdentity getLastMessageIdentity()
                                       throws JMSException

Description copied from interface: Subscription

Retreives the messageIdentity of the last message sent to this Queue based Subscription

Specified by:

getLastMessageIdentity in interface Subscription

Returns:

the messageId of the last message or null

Throws:

JMSException

setLastMessageIdentifier

public void setLastMessageIdentifier(MessageIdentity messageIdentity)
                              throws JMSException

Description copied from interface: Subscription

Used for a Queue based Subscription to set the last acknowledged message ID

Specified by:

setLastMessageIdentifier in interface Subscription

Parameters:

messageIdentity -

Throws:

JMSException

doMessageConsume

protected void doMessageConsume(MessageAck ack,
                                boolean remove)
                         throws JMSException

Consume a message. If we are inside a transaction then we just update the consumed messages dispatched counter and we don't actually remove the message until a future call.

Parameters:

ack - the ack command

remove - whether we should actually remove the message (i.e. really consume the message) or should we just update the counters for the dispatcher / prefetch logic to work

Throws:

JMSException

clientIDsEqual

protected boolean clientIDsEqual(ActiveMQMessage message)

equal

protected static final boolean equal(java.lang.Object left,
                                     java.lang.Object right)