jrun.jms.core.messagemgr
Class ConsumerEndpoint

java.lang.Object
  |
  +--jrun.jms.core.messagemgr.ConsumerEndpoint

All Implemented Interfaces:

DestinationCacheEventListener, Identifiable, java.lang.Runnable, java.io.Serializable

Direct Known Subclasses:

QueueBrowserEndpoint, QueueConsumerEndpoint, TopicConsumerEndpoint

public abstract class ConsumerEndpoint

extends java.lang.Object

implements java.io.Serializable, Identifiable, DestinationCacheEventListener, java.lang.Runnable

A Consumer is a message subscriber with a unique identity

Version:

$Revision: 1.15 $ $Date: 2001/09/18 10:50:34 $

Author:

Jim Alateras

See Also:

Serialized Form

Field Summary

protected int	_ackMode The acknowledgement mode for this endpoint.
protected int	_connectionId Holds the connection id of the connection that the endpoint belongs too
protected boolean	_destroyed Identifies this endpoint as being destroyed.
protected InternalMessageListener	_listener Holds the consumer's message listener.
protected MessageQueue	_messages Transient cache of all messages waiting to be sent to the consumer.
protected boolean	_nolocal The nolocal indicator, if set, inhibits consuming messages that have been published on the same connection
protected boolean	_scheduled Indicates whether the this cache has been scheduled with the dispatcher for asynchronous message delivery.
protected Scheduler	_scheduler This is the scheduler that is used to deliver messages if a consumer has a registered listener
protected JMSSelector	_selector The selector assoicated with this consumer.
protected JmsServerSession	_session caches the session that created this endpoint
protected int	_size Maintains the maximum size of this cache
protected boolean	_stopped This determines whether message delivery to the registered listener is enabled or disabled.
protected boolean	_transacted Indicates whether this endpoint belongs to a transacted session

Method Summary

abstract boolean	canGarbageCollect(MessageHandle handle) Check if this message is referenced by this endpoint.
protected void	clearMessages() Clear all messages in the cache, regardless of whether they are persistent or non-persistent
void	destroy() Destroy this object
int	getAckMode() Return the ackmode for this endpoint
long	getClientId() Return a reference to the client identity.
int	getConnectionId() Return the connection id that this endpoint belongs too
abstract JmsDestination	getDestination() Return the destination that this consumer is subscribed too
long	getId() Return the identity of the object as a long
java.lang.String	getIdAsString() Return the identity of the object as a string
protected Logger	getLogger() Return a reference to the logger
int	getMaximumSize() Return the cache's maximum size
int	getMessageCount() Return the number of unsent messages in the cache for this consumer
boolean	getNoLocal() Return the value of the nolocal indicator
JMSSelector	getSelector() Return the selector for this endpoint or null if one is not specified
JmsServerSession	getSession() Return a reference to the session owning this endpoint
boolean	getTransacted() Check whether this endpoint belongs to a transacted session
int	getUnprocessedMessageCount() Return the number of unsent and unacked messages for this consumer
int	getUnsentMessages() Return the number of unsent messages
int	hashCode()
boolean	messageAdded(MessageImpl message) This event is called when a non-persistent message is added to the DestinationCache.
boolean	messageRemoved(MessageImpl message) This event is called when a message is removed from the DestinationCache.
boolean	persistentMessageAdded(MessageImpl message) This event is called when a persistent message is added to the DestinationCache.
boolean	persistentMessageRemoved(MessageImpl message) This event is called when a message is removed from the DestinationCache.
abstract MessageHandle	receiveMessage() Return the next message to the client.
void	recover() This message will return all unacked messages to the queue and allow them to be resent to the consumer with the redelivery flag on.
protected void	schedule() Schedule asynchronouse message delivery
void	setAckMode(int ackmode) Set the ackMode for this endpoint
void	setCacheEvictionPolicy(CacheEvictionPolicy policy) Set the #CacheEvictionPolicy for this object.
void	setClientId(long id) Set the client identity for this identity
void	setConnectionId(int id) Set the connection that this endpooint belongs too
void	setMaximumSize(int size) Set the maximum size of the cache.
void	setMessageListener(InternalMessageListener listener) Set the message listener for this consmer.
void	setNoLocal(boolean nolocal) Set the no local indicator for this endpoint
void	setSelector(java.lang.String selector) Set the selector for this endpoint
void	setStopped(boolean stop) Start or stop message delivery
void	setTransacted(boolean transacted) Set the state if the transacted flag for this endpoint
void	shutdown() Close the cache and unregister all the consumers.
void	startMessageDelivery() Start delivering messages to the registered listener
void	stopMessageDelivery() Stop delivering messages to the registered listener
java.lang.String	toString() Return a stringified version of the consumer
abstract void	unregister() Unregister this consumer for the specified destination cache, so that it will stop receiving messages from it.

Methods inherited from class java.lang.Object

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

Methods inherited from interface java.lang.Runnable

run

Field Detail

_selector

protected JMSSelector _selector

The selector assoicated with this consumer. A selector is used to filter messages.

_messages

protected MessageQueue _messages

Transient cache of all messages waiting to be sent to the consumer. This should be a priority cache

_listener

protected InternalMessageListener _listener

Holds the consumer's message listener. This means that messages will be pushed down

_size

protected int _size

Maintains the maximum size of this cache

_stopped

protected transient boolean _stopped

This determines whether message delivery to the registered listener is enabled or disabled.

_destroyed

protected transient boolean _destroyed

Identifies this endpoint as being destroyed.

_scheduler

protected transient Scheduler _scheduler

This is the scheduler that is used to deliver messages if a consumer has a registered listener

_scheduled

protected transient boolean _scheduled

Indicates whether the this cache has been scheduled with the dispatcher for asynchronous message delivery.

_ackMode

protected transient int _ackMode

The acknowledgement mode for this endpoint.

_nolocal

protected transient boolean _nolocal

The nolocal indicator, if set, inhibits consuming messages that have been published on the same connection

_transacted

protected transient boolean _transacted

Indicates whether this endpoint belongs to a transacted session

_connectionId

protected transient int _connectionId

Holds the connection id of the connection that the endpoint belongs too

_session

protected JmsServerSession _session

caches the session that created this endpoint

Method Detail

getDestination

public abstract JmsDestination getDestination()

Return the destination that this consumer is subscribed too

Returns:

JmsDestination

getId

public long getId()

Description copied from interface: Identifiable

Return the identity of the object as a long

Specified by:

getId in interface Identifiable

Following copied from interface: jrun.jms.core.Identifiable

Returns:

long - the object identity

getIdAsString

public java.lang.String getIdAsString()

Description copied from interface: Identifiable

Return the identity of the object as a string

Specified by:

getIdAsString in interface Identifiable

Following copied from interface: jrun.jms.core.Identifiable

Returns:

String - the identity

hashCode

public int hashCode()

Overrides:

hashCode in class java.lang.Object

toString

public java.lang.String toString()

Return a stringified version of the consumer

Overrides:

toString in class java.lang.Object

Returns:

String

unregister

public abstract void unregister()

Unregister this consumer for the specified destination cache, so that it will stop receiving messages from it.

getClientId

public long getClientId()

Return a reference to the client identity. This is an indirect reference back to the remote client, which can asynchronously receive messages

Returns:

long - identity of the client scoped to the session

setClientId

public void setClientId(long id)

Set the client identity for this identity

Parameters:

id - - the client identity

getSelector

public JMSSelector getSelector()

Return the selector for this endpoint or null if one is not specified

Returns:

String - the endpoint's selector

setSelector

public void setSelector(java.lang.String selector)
                 throws javax.jms.InvalidSelectorException

Set the selector for this endpoint

Parameters:

selector - - message selector as a string

Throws:

InvalidSelectorException - - raised when selector is not well-formed

getAckMode

public int getAckMode()

Return the ackmode for this endpoint

Returns:

int - acknowledgement mode

setAckMode

public void setAckMode(int ackmode)

Set the ackMode for this endpoint

Parameters:

ackmode - - the new ack mode for the endpoint

getConnectionId

public int getConnectionId()

Return the connection id that this endpoint belongs too

Returns:

int - connection id

setConnectionId

public void setConnectionId(int id)

Set the connection that this endpooint belongs too

Parameters:

id - - connection identity

getNoLocal

public boolean getNoLocal()

Return the value of the nolocal indicator

Returns:

boolean

setNoLocal

public void setNoLocal(boolean nolocal)

Set the no local indicator for this endpoint

Parameters:

nolocal - - true to inhibit messages published on this connection

getTransacted

public boolean getTransacted()

Check whether this endpoint belongs to a transacted session

Returns:

boolean - true if it does

setTransacted

public void setTransacted(boolean transacted)

Set the state if the transacted flag for this endpoint

Parameters:

transacted - - true if it is transacted

setMaximumSize

public void setMaximumSize(int size)

Set the maximum size of the cache. If there are more than this number of messages in the cache the CacheEvictionPolicy is enforced to remove messages.

Parameters:

size - - maximum number of messages a destination can hold

getMaximumSize

public int getMaximumSize()

Return the cache's maximum size

Returns:

int - size of cache

getUnsentMessages

public int getUnsentMessages()

Return the number of unsent messages

Returns:

int - number of unsent messages

setCacheEvictionPolicy

public void setCacheEvictionPolicy(CacheEvictionPolicy policy)

Set the #CacheEvictionPolicy for this object. This determines how messages are removed when the cache's upper limit is reached. If no policy is specified then the default #LifoEvictionPolicy is applied.

Parameters:

policy - - cache's eviction policy

getMessageCount

public int getMessageCount()

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

Returns:

int - nmber of unsent messages

getSession

public JmsServerSession getSession()

Return a reference to the session owning this endpoint

Returns:

JmsServerSession - the owning session

messageAdded

public boolean messageAdded(MessageImpl message)

Description copied from interface: DestinationCacheEventListener

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

Specified by:

messageAdded in interface DestinationCacheEventListener

Following copied from interface: jrun.jms.core.messagemgr.DestinationCacheEventListener

Parameters:

message - - message added to cache

messageRemoved

public boolean messageRemoved(MessageImpl message)

Description copied from interface: DestinationCacheEventListener

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

Specified by:

messageRemoved in interface DestinationCacheEventListener

Following copied from interface: jrun.jms.core.messagemgr.DestinationCacheEventListener

Parameters:

message - - message removed from cache

persistentMessageAdded

public boolean persistentMessageAdded(MessageImpl message)
                               throws PersistenceException

Description copied from interface: DestinationCacheEventListener

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

Specified by:

persistentMessageAdded in interface DestinationCacheEventListener

Following copied from interface: jrun.jms.core.messagemgr.DestinationCacheEventListener

Parameters:

connection - - the database connection

message - - message added to cache

Throws:

PersistenceException - - if there is a persistence related problem

persistentMessageRemoved

public boolean persistentMessageRemoved(MessageImpl message)
                                 throws PersistenceException

Description copied from interface: DestinationCacheEventListener

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

Specified by:

persistentMessageRemoved in interface DestinationCacheEventListener

Following copied from interface: jrun.jms.core.messagemgr.DestinationCacheEventListener

Parameters:

connection - - the database connection

message - - message to remove from cache

Throws:

PersistenceException - - if there is a persistence related problem

recover

public void recover()

This message will return all unacked messages to the queue and allow them to be resent to the consumer with the redelivery flag on.

destroy

public void destroy()

Destroy this object

setMessageListener

public void setMessageListener(InternalMessageListener listener)

Set the message listener for this consmer. If a message listener is set then messages will be scheduled to be sent to it when they are available

Each consumer cache can only have a single message listener. To remove the message listener call this method with null argument

Parameters:

listener - - the message listener to add.

receiveMessage

public abstract MessageHandle receiveMessage()

Return the next message to the client. This will also mark the message as sent and move it to the sent queue

Returns:

MessageHandle - handle to the next message in the list

canGarbageCollect

public abstract boolean canGarbageCollect(MessageHandle handle)

Check if this message is referenced by this endpoint. If it is then the message cannot be garbage collected.

Parameters:

handle - - message to check

Returns:

boolean - true if we can garbage collect

shutdown

public void shutdown()

Close the cache and unregister all the consumers. Notify any and all DestinationCacheLifecycleListeners.

Once the cache is closed it will no longger receive messages for this destination.

startMessageDelivery

public void startMessageDelivery()

Start delivering messages to the registered listener

stopMessageDelivery

public void stopMessageDelivery()

Stop delivering messages to the registered listener

getLogger

protected Logger getLogger()

Return a reference to the logger

Returns:

Logger

schedule

protected void schedule()

Schedule asynchronouse message delivery

setStopped

public void setStopped(boolean stop)

Start or stop message delivery

Parameters:

stop - - true to stop message delivery; false to start it

getUnprocessedMessageCount

public int getUnprocessedMessageCount()

Return the number of unsent and unacked messages for this consumer

Returns:

int - number of unacked or unsent messages

clearMessages

protected void clearMessages()

Clear all messages in the cache, regardless of whether they are persistent or non-persistent