org.objectweb.joram.mom.dest
Class QueueImpl

java.lang.Object
  org.objectweb.joram.mom.dest.DestinationImpl
      org.objectweb.joram.mom.dest.QueueImpl

All Implemented Interfaces:

java.io.Serializable, DestinationImplMBean, QueueImplMBean

Direct Known Subclasses:

ClusterQueueImpl, JMSBridgeQueueImpl

public class QueueImpl
extends DestinationImpl
implements QueueImplMBean

The QueueImpl class implements the MOM queue behaviour, basically storing messages and delivering them upon clients requests.

See Also:

Serialized Form

Field Summary

protected long	arrivalsCounter Counter of messages arrivals.
protected java.util.Hashtable	consumers Table keeping the messages' consumers identifiers.
protected java.util.Hashtable	contexts Table keeping the messages' consumers contexts.
protected java.util.Hashtable	deliveredMsgs Table holding the delivered messages before acknowledgement.
static org.objectweb.util.monolog.api.Logger	logger
protected java.util.Vector	messages Vector holding the messages before delivery.
protected int	nbMaxMsg nb Max of Message store in queue (-1 no limit).
protected long	period period to run task at regular interval: cleaning, load-balancing, etc.
protected boolean	receiving true if the queue is currently receiving messages.
protected java.util.Vector	requests Vector holding the requests before reply or expiry.

Fields inherited from class org.objectweb.joram.mom.dest.DestinationImpl

_rights, agent, clients, creationDate, dmqId, freeReading, freeWriting, nbMsgsDeliverSinceCreation, nbMsgsReceiveSinceCreation, nbMsgsSentToDMQSinceCreation, READ, READWRITE, WRITE

Constructor Summary

QueueImpl(fr.dyade.aaa.agent.AgentId adminId, java.util.Properties prop)
Constructs a QueueImpl instance.

Method Summary

void	abortReceiveRequest(fr.dyade.aaa.agent.AgentId from, AbortReceiveRequest not)
void	acknowledgeRequest(fr.dyade.aaa.agent.AgentId from, AcknowledgeRequest not) Method implementing the reaction to an AcknowledgeRequest instance, requesting messages to be acknowledged.
protected boolean	addMessage(Message message) Adds a message in the list of messages to deliver.
void	browseRequest(fr.dyade.aaa.agent.AgentId from, BrowseRequest not) Method implementing the queue reaction to a BrowseRequest instance, requesting an enumeration of the messages on the queue.
protected boolean	checkDelivery(Message msg) Returns true if conditions are ok to deliver the message.
void	cleanPendingMessage() Removes all messages that the time-to-live is expired.
protected DMQManager	cleanPendingMessage(long currentTime) Cleans the pending messages list.
void	cleanWaitingRequest() Removes all request that the expiration time is expired.
protected void	cleanWaitingRequest(long currentTime) Cleans the waiting request list.
protected void	deliverMessages(int index) Actually tries to answer the pending "receive" requests.
void	denyRequest(fr.dyade.aaa.agent.AgentId from, DenyRequest not) Method implementing the reaction to a DenyRequest instance, requesting messages to be denied.
void	destinationAdminRequestNot(fr.dyade.aaa.agent.AgentId from, DestinationAdminRequestNot not)
protected void	doClientMessages(fr.dyade.aaa.agent.AgentId from, ClientMessages not) Method specifically processing a ClientMessages instance.
protected void	doDeleteNot(fr.dyade.aaa.agent.DeleteNot not) Method specifically processing a fr.dyade.aaa.agent.DeleteNot instance.
protected void	doRightRequest(SetRightRequest not) Method specifically processing a SetRightRequest instance.
protected void	doUnknownAgent(fr.dyade.aaa.agent.UnknownAgent uA) Method specifically processing an UnknownAgent instance.
protected ClientMessages	getClientMessages(int nb, java.lang.String selector, boolean remove) Get a client message contain nb messages.
static fr.dyade.aaa.agent.AgentId	getDefaultDMQId() Static method returning the default DMQ identifier.
static java.lang.Integer	getDefaultThreshold() Static method returning the default threshold.
int	getDeliveredMessageCount() Returns the number of messages delivered and waiting for acknowledge.
javax.management.openmbean.CompositeData	getMessage(java.lang.String msgId) Returns the description of a particular pending message.
javax.management.openmbean.TabularData	getMessages() Returns the description of all pending messages.
protected java.lang.StringBuffer	getMsgTxPrefix()
int	getNbMaxMsg() Returns the maximum number of message for the destination.
long	getNbMsgsReceiveSinceCreation() Returns the number of messages received since creation time of this destination.
int	getPendingMessageCount() Returns the number of pending messages in the queue.
long	getPeriod() Returns the period value of this queue, -1 if not set.
protected Message	getQueueMessage(java.lang.String msgId, boolean remove) get mom message, delete if remove = true.
protected java.util.Hashtable	getStatisticHashtable() Overrides the destination method to add pending message counter
int	getThreshold() Returns the threshold value of this queue, -1 if not set.
int	getWaitingRequestCount() Returns the number of waiting requests in the queue.
protected void	handleExpiredNot(fr.dyade.aaa.agent.AgentId from, fr.dyade.aaa.agent.ExpiredNot not)
void	initialize(boolean firstTime) Initializes the destination.
protected boolean	isUndeliverable(Message message) Returns true if a given message is considered as undeliverable, because its delivery count matches the queue's threshold, if any, or the server's default threshold value (if any).
protected void	messageDelivered(java.lang.String msgId) call in deliverMessages just after forward(msg), overload this method to process a specific treatment.
protected void	messageRemoved(java.lang.String msgId) call in deliverMessages just after a remove message (invalid), overload this method to process a specific treatment.
void	monitGetDMQSettings(fr.dyade.aaa.agent.AgentId from, Monit_GetDMQSettings not) Overrides this DestinationImpl method for sending back the threshold along with the DMQ id.
void	monitGetNbMaxMsg(fr.dyade.aaa.agent.AgentId from, Monit_GetNbMaxMsg not) Method implementing the reaction to a Monit_GetNbMaxMsg notification requesting the number max of messages in this queue.
void	monitGetPendingMessages(fr.dyade.aaa.agent.AgentId from, Monit_GetPendingMessages not) Method implementing the reaction to a Monit_GetPendingMessages notification requesting the number of pending messages.
void	monitGetPendingRequests(fr.dyade.aaa.agent.AgentId from, Monit_GetPendingRequests not) Method implementing the reaction to a Monit_GetPendingRequests notification requesting the number of pending requests.
void	readBag(java.io.ObjectInputStream in)
void	receiveRequest(fr.dyade.aaa.agent.AgentId from, ReceiveRequest not) Method implementing the reaction to a ReceiveRequest instance, requesting a message.
protected void	setMsgTxName(Message msg)
void	setNbMaxMsg(int nbMaxMsg) Sets the maximum number of message for the destination.
void	setNbMaxMsgRequest(fr.dyade.aaa.agent.AgentId from, SetNbMaxMsgRequest req) Method implementing the reaction to a SetNbMaxMsgRequest instance setting the NbMaxMsg value for this queue.
void	setPeriod(long period) Sets or unsets the period for this queue.
void	setThreshold(int threshold) Sets or unsets the threshold for this queue.
void	setThreshRequest(fr.dyade.aaa.agent.AgentId from, SetThreshRequest req) Method implementing the reaction to a SetThreshRequest instance setting the threshold value for this queue.
protected void	storeMessage(Message message) Actually stores a message in the deliverables vector.
protected void	storeMessageHeader(Message message) Actually stores a message header in the deliverables vector.
java.lang.String	toString() Returns a string representation of this destination.
void	wakeUpNot(WakeUpNot not) wake up, and cleans the queue.
void	writeBag(java.io.ObjectOutputStream out)

Methods inherited from class org.objectweb.joram.mom.dest.DestinationImpl

canBeDeleted, clientMessages, deleteNot, forward, getCreationDate, getCreationTimeInMillis, getDestinationId, getDMQId, getId, getNbMsgsDeliverSinceCreation, getNbMsgsSentToDMQSinceCreation, getRight, getRights, handleDeniedMessage, isAdministrator, isFreeReading, isFreeWriting, isLocal, isReader, isWriter, monitFreeAccess, monitGetReaders, monitGetStat, monitGetWriters, postProcess, postProcess, preProcess, preProcess, processSetRight, replyToTopic, requestGroupNot, setAgent, setDMQRequest, setFreeReading, setFreeWriting, setRightRequest, specialAdminProcess, specialAdminRequest, unknownAgent

Methods inherited from class java.lang.Object

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

Methods inherited from interface org.objectweb.joram.mom.dest.DestinationImplMBean

getCreationDate, getCreationTimeInMillis, getDestinationId, getDMQId, getNbMsgsDeliverSinceCreation, getNbMsgsSentToDMQSinceCreation, getRight, getRights, isFreeReading, isFreeWriting, setFreeReading, setFreeWriting

Field Detail

logger

public static org.objectweb.util.monolog.api.Logger logger

period

protected long period

period to run task at regular interval: cleaning, load-balancing, etc.

consumers

protected java.util.Hashtable consumers

Table keeping the messages' consumers identifiers.

contexts

protected java.util.Hashtable contexts

Table keeping the messages' consumers contexts.

arrivalsCounter

protected long arrivalsCounter

Counter of messages arrivals.

requests

protected java.util.Vector requests

Vector holding the requests before reply or expiry.

receiving

protected transient boolean receiving

true if the queue is currently receiving messages.

messages

protected transient java.util.Vector messages

Vector holding the messages before delivery.

deliveredMsgs

protected transient java.util.Hashtable deliveredMsgs

Table holding the delivered messages before acknowledgement.

nbMaxMsg

protected int nbMaxMsg

nb Max of Message store in queue (-1 no limit).

Constructor Detail

QueueImpl

public QueueImpl(fr.dyade.aaa.agent.AgentId adminId,
                 java.util.Properties prop)

Constructs a QueueImpl instance.

Parameters:

adminId - Identifier of the administrator of the queue.

prop - The initial set of properties.

Method Detail

getPeriod

public long getPeriod()

Returns the period value of this queue, -1 if not set.

Specified by:

getPeriod in interface QueueImplMBean

Returns:

the period value of this queue; -1 if not set.

setPeriod

public void setPeriod(long period)

Sets or unsets the period for this queue.

Specified by:

setPeriod in interface QueueImplMBean

Parameters:

period - The period value to be set or -1 for unsetting previous value.

getThreshold

public int getThreshold()

Returns the threshold value of this queue, -1 if not set.

Specified by:

getThreshold in interface QueueImplMBean

Returns:

the threshold value of this queue; -1 if not set.

getDefaultThreshold

public static java.lang.Integer getDefaultThreshold()

Static method returning the default threshold.

getDefaultDMQId

public static fr.dyade.aaa.agent.AgentId getDefaultDMQId()

Static method returning the default DMQ identifier.

setThreshold

public void setThreshold(int threshold)

Sets or unsets the threshold for this queue.

Specified by:

setThreshold in interface QueueImplMBean

Parameters:

threshold - The threshold value to be set (-1 for unsetting previous value).

cleanWaitingRequest

public void cleanWaitingRequest()

Removes all request that the expiration time is expired.

Specified by:

cleanWaitingRequest in interface QueueImplMBean

cleanWaitingRequest

protected void cleanWaitingRequest(long currentTime)

Cleans the waiting request list. Removes all request that the expiration time is less than the time given in parameter.

Parameters:

currentTime - The current time.

getWaitingRequestCount

public int getWaitingRequestCount()

Returns the number of waiting requests in the queue.

Specified by:

getWaitingRequestCount in interface QueueImplMBean

Returns:

The number of waiting requests.

cleanPendingMessage

public void cleanPendingMessage()

Removes all messages that the time-to-live is expired.

Specified by:

cleanPendingMessage in interface QueueImplMBean

cleanPendingMessage

protected DMQManager cleanPendingMessage(long currentTime)

Cleans the pending messages list. Removes all messages which expire before the date given in parameter.

Parameters:

currentTime - The current time.

Returns:

A DMQManager which contains the expired messages. null if there wasn't any.

getPendingMessageCount

public int getPendingMessageCount()

Returns the number of pending messages in the queue.

Specified by:

getPendingMessageCount in interface QueueImplMBean

Returns:

The number of pending messages.

getDeliveredMessageCount

public int getDeliveredMessageCount()

Returns the number of messages delivered and waiting for acknowledge.

Specified by:

getDeliveredMessageCount in interface QueueImplMBean

Returns:

The number of messages delivered.

getNbMaxMsg

public int getNbMaxMsg()

Returns the maximum number of message for the destination. If the limit is unset the method returns -1.

Specified by:

getNbMaxMsg in interface QueueImplMBean

Returns:

the maximum number of message for subscription if set; -1 otherwise.

setNbMaxMsg

public void setNbMaxMsg(int nbMaxMsg)

Sets the maximum number of message for the destination.

Specified by:

setNbMaxMsg in interface QueueImplMBean

Parameters:

nbMaxMsg - the maximum number of message (-1 set no limit).

initialize

public void initialize(boolean firstTime)

Initializes the destination.

Specified by:

initialize in class DestinationImpl

Parameters:

firstTime - true when first called by the factory

toString

public java.lang.String toString()

Returns a string representation of this destination.

Specified by:

toString in interface DestinationImplMBean

Overrides:

toString in class java.lang.Object

wakeUpNot

public void wakeUpNot(WakeUpNot not)

wake up, and cleans the queue.

setThreshRequest

public void setThreshRequest(fr.dyade.aaa.agent.AgentId from,
                             SetThreshRequest req)
                      throws AccessException

Method implementing the reaction to a SetThreshRequest instance setting the threshold value for this queue.

Throws:

AccessException - If the requester is not the administrator.

setNbMaxMsgRequest

public void setNbMaxMsgRequest(fr.dyade.aaa.agent.AgentId from,
                               SetNbMaxMsgRequest req)
                        throws AccessException

Method implementing the reaction to a SetNbMaxMsgRequest instance setting the NbMaxMsg value for this queue.

Throws:

AccessException - If the requester is not the administrator.

monitGetDMQSettings

public void monitGetDMQSettings(fr.dyade.aaa.agent.AgentId from,
                                Monit_GetDMQSettings not)
                         throws AccessException

Overrides this DestinationImpl method for sending back the threshold along with the DMQ id.

Overrides:

monitGetDMQSettings in class DestinationImpl

Throws:

AccessException - If the requester is not the administrator.

monitGetPendingMessages

public void monitGetPendingMessages(fr.dyade.aaa.agent.AgentId from,
                                    Monit_GetPendingMessages not)
                             throws AccessException

Method implementing the reaction to a Monit_GetPendingMessages notification requesting the number of pending messages.

Throws:

AccessException - If the requester is not the administrator.

monitGetPendingRequests

public void monitGetPendingRequests(fr.dyade.aaa.agent.AgentId from,
                                    Monit_GetPendingRequests not)
                             throws AccessException

Method implementing the reaction to a Monit_GetPendingRequests notification requesting the number of pending requests.

Throws:

AccessException - If the requester is not the administrator.

monitGetNbMaxMsg

public void monitGetNbMaxMsg(fr.dyade.aaa.agent.AgentId from,
                             Monit_GetNbMaxMsg not)
                      throws AccessException

Method implementing the reaction to a Monit_GetNbMaxMsg notification requesting the number max of messages in this queue.

Throws:

AccessException - If the requester is not the administrator.

getStatisticHashtable

protected java.util.Hashtable getStatisticHashtable()

Overrides the destination method to add pending message counter

Overrides:

getStatisticHashtable in class DestinationImpl

Returns:

a Hashtable with some information statistics about the destination. Keys are String describing the values.

receiveRequest

public void receiveRequest(fr.dyade.aaa.agent.AgentId from,
                           ReceiveRequest not)
                    throws AccessException

Method implementing the reaction to a ReceiveRequest instance, requesting a message.

This method stores the request and launches a delivery sequence.

Throws:

AccessException - If the sender is not a reader.

browseRequest

public void browseRequest(fr.dyade.aaa.agent.AgentId from,
                          BrowseRequest not)
                   throws AccessException

Method implementing the queue reaction to a BrowseRequest instance, requesting an enumeration of the messages on the queue.

The method sends a BrowseReply back to the client. Expired messages are sent to the DMQ.

Throws:

AccessException - If the requester is not a reader.

acknowledgeRequest

public void acknowledgeRequest(fr.dyade.aaa.agent.AgentId from,
                               AcknowledgeRequest not)

Method implementing the reaction to an AcknowledgeRequest instance, requesting messages to be acknowledged.

denyRequest

public void denyRequest(fr.dyade.aaa.agent.AgentId from,
                        DenyRequest not)

Method implementing the reaction to a DenyRequest instance, requesting messages to be denied.

This method denies the messages and launches a delivery sequence. Messages considered as undeliverable are sent to the DMQ.

abortReceiveRequest

public void abortReceiveRequest(fr.dyade.aaa.agent.AgentId from,
                                AbortReceiveRequest not)

destinationAdminRequestNot

public void destinationAdminRequestNot(fr.dyade.aaa.agent.AgentId from,
                                       DestinationAdminRequestNot not)

Specified by:

destinationAdminRequestNot in class DestinationImpl

doRightRequest

protected void doRightRequest(SetRightRequest not)

Method specifically processing a SetRightRequest instance.

When a reader is removed, and receive requests of this reader are still on the queue, they are replied to by an ExceptionReply.

Specified by:

doRightRequest in class DestinationImpl

doClientMessages

protected void doClientMessages(fr.dyade.aaa.agent.AgentId from,
                                ClientMessages not)

Method specifically processing a ClientMessages instance.

This method stores the messages and launches a delivery sequence.

Specified by:

doClientMessages in class DestinationImpl

doUnknownAgent

protected void doUnknownAgent(fr.dyade.aaa.agent.UnknownAgent uA)

Method specifically processing an UnknownAgent instance.

The specific processing is done when a QueueMsgReply was sent to a requester which does not exist anymore. In that case, the messages sent to this requester and not yet acknowledged are marked as "denied" for delivery to an other requester, and a new delivery sequence is launched. Messages considered as undeliverable are removed and sent to the DMQ.

Specified by:

doUnknownAgent in class DestinationImpl

doDeleteNot

protected void doDeleteNot(fr.dyade.aaa.agent.DeleteNot not)

Method specifically processing a fr.dyade.aaa.agent.DeleteNot instance.

ExceptionReply replies are sent to the pending receivers, and the remaining messages are sent to the DMQ and deleted.

Specified by:

doDeleteNot in class DestinationImpl

getMsgTxPrefix

protected final java.lang.StringBuffer getMsgTxPrefix()

setMsgTxName

protected final void setMsgTxName(Message msg)

storeMessage

protected final void storeMessage(Message message)

Actually stores a message in the deliverables vector.

Parameters:

message - The message to store.

storeMessageHeader

protected final void storeMessageHeader(Message message)

Actually stores a message header in the deliverables vector.

Parameters:

message - The message to store.

addMessage

protected final boolean addMessage(Message message)

Adds a message in the list of messages to deliver.

Parameters:

message - the message to add.

Returns:

true if the message has been added. false if the queue is full.

getClientMessages

protected ClientMessages getClientMessages(int nb,
                                           java.lang.String selector,
                                           boolean remove)

Get a client message contain nb messages.

Parameters:

nb - number of messages returned in ClientMessage.

selector - jms selector

remove - delete all messages returned if true

Returns:

ClientMessages (contains nb Messages)

getQueueMessage

protected Message getQueueMessage(java.lang.String msgId,
                                  boolean remove)

get mom message, delete if remove = true.

Parameters:

msgId - message identification

remove - if true delete message

Returns:

mom message

getMessage

public javax.management.openmbean.CompositeData getMessage(java.lang.String msgId)
                                                    throws java.lang.Exception

Returns the description of a particular pending message. The message is pointed out through its unique identifier.

Specified by:

getMessage in interface QueueImplMBean

Parameters:

msgId - The unique message's identifier.

Returns:

the description of the message.

Throws:

java.lang.Exception

See Also:

MessageJMXWrapper

getMessages

public javax.management.openmbean.TabularData getMessages()
                                                   throws java.lang.Exception

Returns the description of all pending messages.

Specified by:

getMessages in interface QueueImplMBean

Returns:

the description of the message.

Throws:

java.lang.Exception

See Also:

MessageJMXWrapper

deliverMessages

protected void deliverMessages(int index)

Actually tries to answer the pending "receive" requests.

The method may send QueueMsgReply replies to clients.

Parameters:

index - Index where starting to "browse" the requests.

checkDelivery

protected boolean checkDelivery(Message msg)

Returns true if conditions are ok to deliver the message. This method must be overloaded in subclasses. Be careful only the message header is accessible.

messageDelivered

protected void messageDelivered(java.lang.String msgId)

call in deliverMessages just after forward(msg), overload this method to process a specific treatment.

messageRemoved

protected void messageRemoved(java.lang.String msgId)

call in deliverMessages just after a remove message (invalid), overload this method to process a specific treatment.

isUndeliverable

protected boolean isUndeliverable(Message message)

Returns true if a given message is considered as undeliverable, because its delivery count matches the queue's threshold, if any, or the server's default threshold value (if any).

readBag

public void readBag(java.io.ObjectInputStream in)
             throws java.io.IOException,
                    java.lang.ClassNotFoundException

Throws:

java.io.IOException

java.lang.ClassNotFoundException

writeBag

public void writeBag(java.io.ObjectOutputStream out)
              throws java.io.IOException

Throws:

java.io.IOException

getNbMsgsReceiveSinceCreation

public long getNbMsgsReceiveSinceCreation()

Description copied from class: DestinationImpl

Returns the number of messages received since creation time of this destination.

Specified by:

getNbMsgsReceiveSinceCreation in interface DestinationImplMBean

Specified by:

getNbMsgsReceiveSinceCreation in class DestinationImpl

Returns:

the number of messages received since creation time.

handleExpiredNot

protected void handleExpiredNot(fr.dyade.aaa.agent.AgentId from,
                                fr.dyade.aaa.agent.ExpiredNot not)