org.exolab.jms.messagemgr
Class DurableConsumerEndpoint

java.lang.Object
  |
  +--org.exolab.jms.messagemgr.AbstractConsumerEndpoint
        |
        +--org.exolab.jms.messagemgr.AbstractTopicConsumerEndpoint
              |
              +--org.exolab.jms.messagemgr.DurableConsumerEndpoint

All Implemented Interfaces:

ConsumerEndpoint, org.exolab.jms.messagemgr.DestinationCacheEventListener, DestinationEventListener

public class DurableConsumerEndpoint

extends org.exolab.jms.messagemgr.AbstractTopicConsumerEndpoint

A ConsumerEndpoint for durable topic consumers. The state of durable topic consumers is maintained across server invocations by the persistent layer.

DurableConsumerEndpoints are always loaded in memory, whether they are active or inactive. When they are inactive they simply process persistent messages. Non-persistent message are ignored when the durable consumer is inactive.

Version:

$Revision: 1.4 $ $Date: 2005/11/12 12:27:40 $

Author:

Jim Alateras, Tim Anderson

Field Summary

protected java.util.Map

_caches Maintains a map of TopicDestinationCache that this endpoint subscribes to, keyed on JmsTopic.

Constructor Summary

DurableConsumerEndpoint(long consumerId, JmsTopic topic, java.lang.String name, DestinationManager destinations)
Construct a new DurableConsumerEndpoint.

Method Summary

void	activate(long connectionId, java.lang.String selector, boolean noLocal) Activate this durable consumer.
protected void	addMessage(MessageHandle handle) Add the handle to the cache.
void	cacheAdded(JmsDestination destination, DestinationCache cache) Invoked when a message cache is created.
void	cacheRemoved(JmsDestination destination, DestinationCache cache) Invoked when a message cache is removed.
boolean	canConsume(JmsDestination destination) Determines if this consumer can consume messages from the specified destination.
void	deactivate() Deactivate this durable consumer.
void	destinationAdded(JmsDestination destination) Invoked when a destination is created.
void	destinationRemoved(JmsDestination destination) Invoked when a destination is removed.
protected void	doClose() Closes this endpoint.
protected MessageHandle	doReceive(Condition cancel) Return the next available message to the client.
long	getConnectionId() Returns the identity of the connection that owns this consumer.
protected DestinationCache[]	getDestinationCaches() Returns the destination caches.
protected DestinationManager	getDestinationManager() Returns the destination manager.
int	getMessageCount() Return the number of unsent messages in the cache for this consumer.
java.lang.String	getPersistentId() Returns the persistent identifier for this consumer.
protected void	init() Registers this with the associated DestinationCaches.
boolean	isActive() Determines if the endpoint is active.
boolean	isPersistent() Determines if this is a persistent or non-persistent consumer.
boolean	messageAdded(MessageHandle handle, MessageImpl message) This event is called when a non-persistent message is added to the DestinationCache.
void	messageRemoved(java.lang.String messageId) This event is called when a message is removed from the DestinationCache.
boolean	persistentMessageAdded(MessageHandle handle, MessageImpl message) This event is called when a persistent message is added to the DestinationCache.
void	persistentMessageRemoved(java.lang.String messageId) This event is called when a message is removed from the DestinationCache.
void	returnMessage(MessageHandle handle) Return a delivered, but unacknowledged message to the cache.
protected void	setConnectionId(long connectionId) Set the connection identifier.

Methods inherited from class org.exolab.jms.messagemgr.AbstractConsumerEndpoint

close, getDestination, getId, getNoLocal, getSelector, isAsynchronous, isClosed, isWaitingForMessage, notifyMessageAvailable, receive, selects, setAsynchronous, setListener, setNoLocal, setSelector, setWaitingForMessage, toString

Methods inherited from class java.lang.Object

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

Field Detail

_caches

protected java.util.Map _caches

Maintains a map of TopicDestinationCache that this endpoint subscribes to, keyed on JmsTopic. A wildcard subscription may point to more than one.

Constructor Detail

DurableConsumerEndpoint

public DurableConsumerEndpoint(long consumerId,
                               JmsTopic topic,
                               java.lang.String name,
                               DestinationManager destinations)
                        throws javax.jms.InvalidSelectorException,
                               javax.jms.JMSException,
                               PersistenceException

Construct a new DurableConsumerEndpoint.

The consumer is inactive until made active via activate(long, java.lang.String, boolean).

Parameters:

consumerId - the identity of this consumer

topic - the topic to access

destinations - the destination manager

name - the well known name of the durable subscriber

Throws:

javax.jms.InvalidSelectorException - if the selector is invalid

javax.jms.JMSException - if the destination caches can't be constructed

PersistenceException - for any persistence error

Method Detail

isPersistent

public boolean isPersistent()

Determines if this is a persistent or non-persistent consumer.

If persistent, then the consumer is persistent accross subscriptions and server restarts, and getPersistentId() returns a non-null value

Overrides:

isPersistent in class AbstractConsumerEndpoint

Returns:

true

getPersistentId

public java.lang.String getPersistentId()

Returns the persistent identifier for this consumer.

This is the identity of the consumer which is persistent across subscriptions and server restarts.

This implementation returns the consumer name.

Overrides:

getPersistentId in class AbstractConsumerEndpoint

Returns:

the persistent identifier for this consumer

activate

public void activate(long connectionId,
                     java.lang.String selector,
                     boolean noLocal)
              throws javax.jms.JMSException

Activate this durable consumer.

Parameters:

connectionId - the identity of the connection that owns this consumer

selector - the message selector. May be null

noLocal - if true, inhibits the delivery of messages published by its own connection.

Throws:

javax.jms.JMSException - if the consumer can't be activated

javax.jms.InvalidSelectorException - if the selector is invalid

deactivate

public void deactivate()
                throws javax.jms.JMSException

Deactivate this durable consumer.

Throws:

javax.jms.JMSException - if the consumer can't be deactivated

isActive

public boolean isActive()

Determines if the endpoint is active.

Returns:

true if the endpoint is active, false if it is inactive

getConnectionId

public long getConnectionId()

Returns the identity of the connection that owns this consumer.

Returns:

the identity of the connection, or -1 if this is not currently associated with a connection.

See Also:

ServerConnection.getConnectionId()

canConsume

public boolean canConsume(JmsDestination destination)

Determines if this consumer can consume messages from the specified destination.

Overrides:

canConsume in class AbstractConsumerEndpoint

Parameters:

destination - the destination

Returns:

true if the consumer can consume messages from destination; otherwise false

returnMessage

public void returnMessage(MessageHandle handle)

Return a delivered, but unacknowledged message to the cache.

Parameters:

handle - the handle of the message to return

getMessageCount

public int getMessageCount()

Return the number of unsent messages in the cache for this consumer.

Returns:

the number of unsent messages

messageAdded

public boolean messageAdded(MessageHandle handle,
                            MessageImpl message)
                     throws javax.jms.JMSException

This event is called when a non-persistent message is added to the DestinationCache.

Parameters:

handle - a handle to the message

message - the added message

Returns:

true if the listener accepted the message; otherwise false

Throws:

javax.jms.JMSException - if the listener fails to handle the message

messageRemoved

public void messageRemoved(java.lang.String messageId)
                    throws javax.jms.JMSException

This event is called when a message is removed from the DestinationCache.

Parameters:

messageId - the identifier of the removed message

Throws:

javax.jms.JMSException - if the listener fails to handle the message

persistentMessageAdded

public boolean persistentMessageAdded(MessageHandle handle,
                                      MessageImpl message)
                               throws javax.jms.JMSException,
                                      PersistenceException

This event is called when a persistent message is added to the DestinationCache.

Parameters:

handle - a handle to the added message

message - the added message

Returns:

true if the listener accepted the message;

Throws:

javax.jms.JMSException - if the listener fails to handle the message

PersistenceException - if there is a persistence related problem

persistentMessageRemoved

public void persistentMessageRemoved(java.lang.String messageId)
                              throws javax.jms.JMSException,
                                     PersistenceException

This event is called when a message is removed from the DestinationCache.

Parameters:

messageId - the identifier of the removed message

Throws:

javax.jms.JMSException - if the listener fails to handle the message

PersistenceException - if there is a persistence related problem

destinationAdded

public void destinationAdded(JmsDestination destination)

Invoked when a destination is created.

Specified by:

destinationAdded in interface DestinationEventListener

Parameters:

destination - the destination that was added

destinationRemoved

public void destinationRemoved(JmsDestination destination)

Invoked when a destination is removed.

Specified by:

destinationRemoved in interface DestinationEventListener

Parameters:

destination - the destination that was removed

cacheAdded

public void cacheAdded(JmsDestination destination,
                       DestinationCache cache)

Invoked when a message cache is created.

Specified by:

cacheAdded in interface DestinationEventListener

Parameters:

destination - the destination that messages are being cached for

cache - the corresponding cache

cacheRemoved

public void cacheRemoved(JmsDestination destination,
                         DestinationCache cache)

Invoked when a message cache is removed.

Specified by:

cacheRemoved in interface DestinationEventListener

Parameters:

destination - the destination that messages are no longer being cached for

cache - the corresponding cache

init

protected void init()
             throws javax.jms.JMSException

Registers this with the associated DestinationCaches. The consumer may receive messages immediately.

Throws:

javax.jms.JMSException - for any JMS error

setConnectionId

protected void setConnectionId(long connectionId)

Set the connection identifier.

Parameters:

connectionId - the identity of the connection that owns this consumer

See Also:

AbstractTopicConsumerEndpoint.getConnectionId()

addMessage

protected void addMessage(MessageHandle handle)

Add the handle to the cache.

Parameters:

handle - the message handle to add

doReceive

protected MessageHandle doReceive(Condition cancel)
                           throws javax.jms.JMSException

Return the next available message to the client.

Overrides:

doReceive in class AbstractConsumerEndpoint

Parameters:

cancel -

Returns:

the next message, or null if none is available

Throws:

javax.jms.JMSException - for any error

doClose

protected void doClose()

Closes this endpoint.

Overrides:

doClose in class AbstractConsumerEndpoint

getDestinationManager

protected DestinationManager getDestinationManager()

Returns the destination manager.

Returns:

the destination manager

getDestinationCaches

protected DestinationCache[] getDestinationCaches()

Returns the destination caches.

Returns:

the destination caches