Package org.objectweb.joram.mom.dest
Class ClusterQueue
- java.lang.Object
-
- fr.dyade.aaa.agent.Agent
-
- org.objectweb.joram.mom.dest.Destination
-
- org.objectweb.joram.mom.dest.Queue
-
- org.objectweb.joram.mom.dest.ClusterQueue
-
- All Implemented Interfaces:
AgentMBean,Encodable,Serializable,ClusterDestinationMBean,ClusterQueueMBean,DestinationMBean,QueueMBean
public class ClusterQueue extends Queue implements ClusterQueueMBean
TheClusterQueueclass implements the cluster queue behavior.- See Also:
- Serialized Form
-
-
Nested Class Summary
-
Nested classes/interfaces inherited from class org.objectweb.joram.mom.dest.Queue
Queue.QueueFactory
-
-
Field Summary
Fields Modifier and Type Field Description private longclusterDeliveryCountNumber of message send to clusterprotected Map<AgentId,Float>clusterskey = agentId of ClusterQueue value = rateOfFlow (Float)private static booleanDEBUGstatic booleanDEFAULT_AUTO_EVAL_THRESHOLDstatic intDEFAULT_CONSUM_THRESHOLDstatic intDEFAULT_PRODUC_THRESHOLDstatic longDEFAULT_TIME_THRESHOLDstatic longDEFAULT_WAIT_AFTER_CLUSTER_REQprotected LoadingFactorloadingFactorto evaluate the loading factor, overloading, ...private static Loggerlogger(package private) intreceivedFromCluster(package private) intsentToClusterprivate static longserialVersionUIDdefine serialVersionUID for interoperabilityprivate Map<String,Long>timeTablekey = msgId value = dateprivate longtimeThresholdMaximum period of time before forwarding a waiting message or request to other queues of the cluster.private Map<String,List<AgentId>>visitTablekey = msgId value = List (alreadyVisit)-
Fields inherited from class org.objectweb.joram.mom.dest.Queue
ARRIVAL_STATE_PREFIX, arrivalState, cload, defaultDMQId, defaultRedeliveryDelay, defaultThreshold, delayed, DELIVERY_TABLE_PREFIX, DELIVERY_TIMEOUT, deliveryTable, hcons, hprod, LENIENT_REDELIVERY_DELAY, LENIENT_UNDELIVERABLE, logmsg, messages, msgTxPrefix, msgTxPrefixLength, nbExpirations, nbMaxMsg, nbMsgsDeniedSinceCreation, pload, receiving, requests
-
Fields inherited from class org.objectweb.joram.mom.dest.Destination
clients, dmqId, freeReading, freeWriting, nbMsgsDeliverSinceCreation, nbMsgsSentToDMQSinceCreation, READ, READWRITE, strbuf, strictCounters, task, temporary, WRITE
-
Fields inherited from class fr.dyade.aaa.agent.Agent
agentProfiling, emptyString, fixed, logmon
-
Fields inherited from interface fr.dyade.aaa.common.encoding.Encodable
BOOLEAN_ENCODED_SIZE, BYTE_ENCODED_SIZE, DOUBLE_ENCODED_SIZE, FLOAT_ENCODED_SIZE, INT_ENCODED_SIZE, LONG_ENCODED_SIZE, SHORT_ENCODED_SIZE
-
-
Constructor Summary
Constructors Constructor Description ClusterQueue()
-
Method Summary
All Methods Instance Methods Concrete Methods Modifier and Type Method Description private voidclusterAdd(FwdAdminRequestNot req, String joiningQueue)Reaction to the request of adding a new cluster element.private voidclusterJoin(ClusterJoinNot not)Method implementing the reaction to aClusterJoinNotnotification, sent by a fellow queue for notifying this queue to join the cluster, doing a transitive closure of clusters, if any.private voidclusterJoinAck(ClusterJoinAck not)Method implementing the reaction to aClusterJoinAcknotification, doing a transitive closure with the current cluster and the one of the new cluster element.private voidclusterLeave()Ask this queue to leave the cluster.private List<String>clusterList()Returns the cluster list.private voidclusterRemove(AgentId queue)Remove the specified queue from current cluster.protected voiddoDeleteNot(DeleteNot not)Method specifically processing afr.dyade.aaa.agent.DeleteNotinstance.protected voiddoUnknownAgent(UnknownAgent uA)Method specifically processing anUnknownAgentinstance.protected ClientMessagesgetClientMessages(int nb, String selector, boolean remove)get a client message contain nb messages.longgetClusterDeliveryCount()return the number of Message send to cluster.String[]getClusterElements()StringgetConsumerStatus()Gets consumer status (NO, NORMAL, HIGH).intgetConsumThreshold()Gets the number of pending "receive" requests above which a queue will request messages from the other queues of the cluster.intgetEncodableClassId()Enables the sub classes not to implement this method.StringgetProducerStatus()Gets producer status (NO, NORMAL, HIGH).intgetProducThreshold()Gets the number of messages above which a queue is considered loaded.protected MessagegetQueueMessage(String msgId, boolean remove)get mom message, delete if remove = true.floatgetRateOfFlow()Gets an evaluation of the flow of messages handled by the queue.intgetReceivedFromCluster()intgetSentToCluster()StringgetStatus()Gets the status of the queue (RUN, INIT or WAIT).longgetWaitAfterClusterReq()Gets the time (in ms) during which a queue which requested something from the cluster is not authorized to do it again.voidhandleAdminRequestNot(AgentId from, FwdAdminRequestNot not)voidinitialize(boolean firstTime)Initializes the destination.booleanisAutoEvalThreshold()True if an automatic reevaluation of the queues' thresholds values is allowed according to their activity.private voidlBCycleLife(AgentId from, LBCycleLife not)If the messages are not consumed by an other cluster's queue in a period of time, try to consume in this queue.private voidlBMessageGive(AgentId from, LBMessageGive not)load balancing message give by an other cluster queue. process ClientMessages, no need to check if sender is writer.private voidlBMessageHope(AgentId from, LBMessageHope not)load balancing message hope by the "from" queue.protected voidmessageDelivered(String msgId)call in deliverMessages just after forward(msg), overload this method to process a specific treatment.protected voidmonitoringMsgSendToCluster(String msgId)voidpostProcess(ClientMessages not)call factorCheck to evaluate the loading factor, activity, ... and send message to cluster if need.ClientMessagespreProcess(AgentId from, ClientMessages not)overload preProcess(AgentId, ClientMessages) store all msgId in timeTable and visitTable.voidreact(AgentId from, Notification not)Distributes the received notifications to the appropriate reactions.voidreceiveRequest(AgentId from, ReceiveRequest not)Method implementing the reaction to aReceiveRequestinstance, requesting a message.protected voidsendToCluster(Notification not)Sends a notification to all queue in cluster.voidsetAutoEvalThreshold(boolean autoEvalThreshold)Automatic reevaluation of the queues' thresholds can be done according to their activity.voidsetConsumThreshold(int consumThreshold)Sets the number of pending "receive" requests above which a queue will request messages from the other queues of the cluster.voidsetProducThreshold(int producThreshold)Sets the number of messages above which a queue is considered loaded.voidsetProperties(Properties prop, boolean firstTime)Configures aClusterQueueinstance.voidsetWaitAfterClusterReq(long waitAfterClusterReq)Sets the time (in ms) during which a queue which requested something from the cluster is not authorized to do it again.private voidstoreMsgIdInTimeTable(String msgId, Long date)private voidstoreMsgIdInVisitTable(String msgId, AgentId destId)StringtoString()Returns a string representation of this destination.voidwakeUpNot(WakeUpNot not)wake up, and call factorCheck to evaluate the loading factor... if a message stays more than a period of time in timeTable, it is sent to an other (not visited) queue in cluster.-
Methods inherited from class org.objectweb.joram.mom.dest.Queue
abortReceiveRequest, acknowledgeRequest, addClientMessages, addDeliveryTimeMessage, addMessage, agentSave, browseRequest, checkDelivery, cleanPendingMessage, cleanPendingMessage, cleanWaitingRequest, cleanWaitingRequest, clear, decode, deliverMessages, denyRequest, doAdminSyncNotification, doClientMessages, doRightRequest, encode, exportMessages, finalize, getConsumerLoad, getDefaultDMQId, getDefaultRedeliveryDelay, getDefaultThreshold, getDelayedMessage, getDelayedMessageCount, getDelayedMessages, getDeliveredMessageCount, getDeliveryDelay, getEncodedSize, getMessage, getMessages, getMessagesView, getMsgTxPrefix, getNbMaxMsg, getNbMsgsDeliverSinceCreation, getNbMsgsDeniedSinceCreation, getNbMsgsReceiveSinceCreation, getPendingMessageCount, getProducerLoad, getRedeliveryDelay, getStats, getThreshold, getType, getWaitingRequestCount, handleExpiredNot, isPause, isSyncExceptionOnFullDest, isUndeliverable, isValidJMXAttribute, messageRemoved, processDeliveryTime, setDefaultRedeliveryDelay, setDeliveryDelay, setMsgTxName, setNbMaxMsg, setPause, setRedeliveryDelay, setSyncExceptionOnFullDest, setThreshold, storeMessage
-
Methods inherited from class org.objectweb.joram.mom.dest.Destination
agentFinalize, agentInitialize, clientMessages, delete, deleteNot, forward, getCreationDate, getCreationTimeInMillis, getDestinationId, getDMQAgentId, getDMQId, getJMXStatistics, getNbMsgsSentToDMQSinceCreation, getPeriod, getRight, getRights, getRights, handleDeniedMessage, interceptorsAvailable, isAdministrator, isFreeReading, isFreeWriting, isLocal, isReader, isWriter, processAdminCommand, processInterceptors, processPause, processSetRight, processStartHandler, processStopHandler, replyToTopic, requestGroupNot, setAdminId, setFreeReading, setFreeWriting, setPeriod, setRight, unknownAgent
-
Methods inherited from class fr.dyade.aaa.agent.Agent
delete, delete, deploy, deploy, getAgentId, getCommitTime, getId, getLogTopic, getName, getReactNb, getReactTime, hasName, incWorkInProgress, isAgentProfiling, isDeployed, isFixed, isUpdated, load, needToBeCommited, resetCommitTime, resetReactTime, resetTimer, save, sendTo, sendTo, sendTo, setAgentProfiling, setName, setNoSave, setSave
-
Methods inherited from class java.lang.Object
clone, equals, finalize, getClass, hashCode, notify, notifyAll, wait, wait, wait
-
Methods inherited from interface fr.dyade.aaa.agent.AgentMBean
getAgentId, getCommitTime, getReactNb, getReactTime, isAgentProfiling, isFixed, resetCommitTime, resetReactTime, resetTimer, setAgentProfiling
-
Methods inherited from interface org.objectweb.joram.mom.dest.DestinationMBean
delete, getCreationDate, getCreationTimeInMillis, getDestinationId, getDMQId, getName, getNbMsgsDeliverSinceCreation, getNbMsgsReceiveSinceCreation, getNbMsgsSentToDMQSinceCreation, getPeriod, getRight, getRights, getType, isFreeReading, isFreeWriting, setFreeReading, setFreeWriting, setPeriod
-
Methods inherited from interface org.objectweb.joram.mom.dest.QueueMBean
cleanPendingMessage, cleanWaitingRequest, clear, exportMessages, getConsumerLoad, getDelayedMessage, getDelayedMessageCount, getDelayedMessages, getDeliveredMessageCount, getDeliveryDelay, getMessage, getMessages, getNbMaxMsg, getNbMsgsDeniedSinceCreation, getPendingMessageCount, getProducerLoad, getRedeliveryDelay, getThreshold, getWaitingRequestCount, isPause, setDeliveryDelay, setNbMaxMsg, setPause, setRedeliveryDelay, setThreshold
-
-
-
-
Field Detail
-
serialVersionUID
private static final long serialVersionUID
define serialVersionUID for interoperability- See Also:
- Constant Field Values
-
logger
private static final Logger logger
-
DEBUG
private static final boolean DEBUG
-
clusters
protected Map<AgentId,Float> clusters
key = agentId of ClusterQueue value = rateOfFlow (Float)
-
loadingFactor
protected LoadingFactor loadingFactor
to evaluate the loading factor, overloading, ...
-
clusterDeliveryCount
private long clusterDeliveryCount
Number of message send to cluster
-
timeThreshold
private long timeThreshold
Maximum period of time before forwarding a waiting message or request to other queues of the cluster. By default it is set toQueue.period.
-
DEFAULT_PRODUC_THRESHOLD
public static final int DEFAULT_PRODUC_THRESHOLD
- See Also:
- Constant Field Values
-
DEFAULT_CONSUM_THRESHOLD
public static final int DEFAULT_CONSUM_THRESHOLD
- See Also:
- Constant Field Values
-
DEFAULT_AUTO_EVAL_THRESHOLD
public static final boolean DEFAULT_AUTO_EVAL_THRESHOLD
- See Also:
- Constant Field Values
-
DEFAULT_WAIT_AFTER_CLUSTER_REQ
public static final long DEFAULT_WAIT_AFTER_CLUSTER_REQ
- See Also:
- Constant Field Values
-
DEFAULT_TIME_THRESHOLD
public static final long DEFAULT_TIME_THRESHOLD
- See Also:
- Constant Field Values
-
receivedFromCluster
int receivedFromCluster
-
sentToCluster
int sentToCluster
-
-
Method Detail
-
setProperties
public void setProperties(Properties prop, boolean firstTime) throws Exception
Configures aClusterQueueinstance.- Overrides:
setPropertiesin classQueue- Parameters:
prop- The initial set of properties.- Throws:
Exception
-
initialize
public void initialize(boolean firstTime) throws ExceptionInitializes the destination.- Overrides:
initializein classQueue- Parameters:
firstTime- true when first called by the factory- Throws:
Exception
-
handleAdminRequestNot
public void handleAdminRequestNot(AgentId from, FwdAdminRequestNot not)
-
react
public void react(AgentId from, Notification not) throws Exception
Distributes the received notifications to the appropriate reactions.
-
toString
public String toString()
Description copied from class:QueueReturns a string representation of this destination.- Specified by:
toStringin interfaceAgentMBean- Specified by:
toStringin interfaceDestinationMBean- Overrides:
toStringin classQueue- Returns:
- a string representation of this destination.
-
clusterAdd
private void clusterAdd(FwdAdminRequestNot req, String joiningQueue)
Reaction to the request of adding a new cluster element.
-
clusterJoin
private void clusterJoin(ClusterJoinNot not)
Method implementing the reaction to aClusterJoinNotnotification, sent by a fellow queue for notifying this queue to join the cluster, doing a transitive closure of clusters, if any.
-
clusterJoinAck
private void clusterJoinAck(ClusterJoinAck not)
Method implementing the reaction to aClusterJoinAcknotification, doing a transitive closure with the current cluster and the one of the new cluster element.
-
getClusterElements
public String[] getClusterElements()
- Specified by:
getClusterElementsin interfaceClusterDestinationMBean- Returns:
- an array containing the ids of the cluster elements.
-
clusterLeave
private void clusterLeave()
Ask this queue to leave the cluster.
-
clusterRemove
private void clusterRemove(AgentId queue)
Remove the specified queue from current cluster.- Parameters:
queue- The queue which left the cluster
-
preProcess
public ClientMessages preProcess(AgentId from, ClientMessages not)
overload preProcess(AgentId, ClientMessages) store all msgId in timeTable and visitTable.- Overrides:
preProcessin classDestination- Parameters:
from-not-- Returns:
- The incoming messages after processing.
-
postProcess
public void postProcess(ClientMessages not)
call factorCheck to evaluate the loading factor, activity, ... and send message to cluster if need.- Overrides:
postProcessin classDestination- Parameters:
not-
-
wakeUpNot
public void wakeUpNot(WakeUpNot not)
wake up, and call factorCheck to evaluate the loading factor... if a message stays more than a period of time in timeTable, it is sent to an other (not visited) queue in cluster.
-
lBCycleLife
private void lBCycleLife(AgentId from, LBCycleLife not)
If the messages are not consumed by an other cluster's queue in a period of time, try to consume in this queue. update visitTable, and process clientMessages.- Parameters:
from-not-
-
receiveRequest
public void receiveRequest(AgentId from, ReceiveRequest not) throws AccessException
Description copied from class:QueueMethod implementing the reaction to aReceiveRequestinstance, requesting a message.This method stores the request and launches a delivery sequence.
- Overrides:
receiveRequestin classQueue- Parameters:
not- ReceiveRequest- Throws:
AccessException- If the sender is not a reader.
-
lBMessageGive
private void lBMessageGive(AgentId from, LBMessageGive not) throws UnknownNotificationException
load balancing message give by an other cluster queue. process ClientMessages, no need to check if sender is writer.- Parameters:
from- AgentIdnot- LBMessageGive- Throws:
UnknownNotificationException
-
lBMessageHope
private void lBMessageHope(AgentId from, LBMessageHope not)
load balancing message hope by the "from" queue.- Parameters:
from-not- LBMessageHope
-
getClientMessages
protected ClientMessages getClientMessages(int nb, String selector, boolean remove)
get a client message contain nb messages. add cluster monitoring value.- Overrides:
getClientMessagesin classQueue- Parameters:
nb- number of messages returned in ClientMessage.selector- jms selectorremove- delete all messages returned if true- Returns:
- ClientMessages (contains nb Messages)
-
getQueueMessage
protected Message getQueueMessage(String msgId, boolean remove)
get mom message, delete if remove = true. add cluster monitoring value.- Overrides:
getQueueMessagein classQueue- Parameters:
msgId- message identificationremove- if true delete message- Returns:
- mom message
-
sendToCluster
protected void sendToCluster(Notification not)
Sends a notification to all queue in cluster.- Parameters:
not- The notification to send.
-
doDeleteNot
protected void doDeleteNot(DeleteNot not)
Description copied from class:QueueMethod specifically processing afr.dyade.aaa.agent.DeleteNotinstance.ExceptionReplyreplies are sent to the pending receivers, and the remaining messages are sent to the DMQ and deleted.- Overrides:
doDeleteNotin classQueue
-
doUnknownAgent
protected void doUnknownAgent(UnknownAgent uA)
Description copied from class:QueueMethod specifically processing anUnknownAgentinstance.The specific processing is done when a
QueueMsgReplywas 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.- Overrides:
doUnknownAgentin classQueue
-
getClusterDeliveryCount
public long getClusterDeliveryCount()
return the number of Message send to cluster.
-
storeMsgIdInTimeTable
private void storeMsgIdInTimeTable(String msgId, Long date)
- Parameters:
msgId-date-
-
storeMsgIdInVisitTable
private void storeMsgIdInVisitTable(String msgId, AgentId destId)
- Parameters:
msgId-destId-
-
messageDelivered
protected void messageDelivered(String msgId)
Description copied from class:Queuecall in deliverMessages just after forward(msg), overload this method to process a specific treatment.- Overrides:
messageDeliveredin classQueue- Parameters:
msgId-
-
monitoringMsgSendToCluster
protected void monitoringMsgSendToCluster(String msgId)
- Parameters:
msgId-
-
setWaitAfterClusterReq
public void setWaitAfterClusterReq(long waitAfterClusterReq)
Description copied from interface:ClusterQueueMBeanSets the time (in ms) during which a queue which requested something from the cluster is not authorized to do it again.- Specified by:
setWaitAfterClusterReqin interfaceClusterQueueMBean- Parameters:
waitAfterClusterReq-
-
setProducThreshold
public void setProducThreshold(int producThreshold)
Description copied from interface:ClusterQueueMBeanSets the number of messages above which a queue is considered loaded.- Specified by:
setProducThresholdin interfaceClusterQueueMBean- Parameters:
producThreshold-
-
setConsumThreshold
public void setConsumThreshold(int consumThreshold)
Description copied from interface:ClusterQueueMBeanSets the number of pending "receive" requests above which a queue will request messages from the other queues of the cluster.- Specified by:
setConsumThresholdin interfaceClusterQueueMBean- Parameters:
consumThreshold-
-
setAutoEvalThreshold
public void setAutoEvalThreshold(boolean autoEvalThreshold)
Description copied from interface:ClusterQueueMBeanAutomatic reevaluation of the queues' thresholds can be done according to their activity.- Specified by:
setAutoEvalThresholdin interfaceClusterQueueMBean- Parameters:
autoEvalThreshold-
-
getProducThreshold
public int getProducThreshold()
Description copied from interface:ClusterQueueMBeanGets the number of messages above which a queue is considered loaded.- Specified by:
getProducThresholdin interfaceClusterQueueMBean- Returns:
- the produce threshold
-
getConsumThreshold
public int getConsumThreshold()
Description copied from interface:ClusterQueueMBeanGets the number of pending "receive" requests above which a queue will request messages from the other queues of the cluster.- Specified by:
getConsumThresholdin interfaceClusterQueueMBean- Returns:
- the consume threshold
-
isAutoEvalThreshold
public boolean isAutoEvalThreshold()
Description copied from interface:ClusterQueueMBeanTrue if an automatic reevaluation of the queues' thresholds values is allowed according to their activity.- Specified by:
isAutoEvalThresholdin interfaceClusterQueueMBean- Returns:
- true if auto evaluation of thresholds is allowed.
-
getWaitAfterClusterReq
public long getWaitAfterClusterReq()
Description copied from interface:ClusterQueueMBeanGets the time (in ms) during which a queue which requested something from the cluster is not authorized to do it again.- Specified by:
getWaitAfterClusterReqin interfaceClusterQueueMBean- Returns:
- the minimum time to wait before another cluster request.
-
getRateOfFlow
public float getRateOfFlow()
Description copied from interface:ClusterQueueMBeanGets an evaluation of the flow of messages handled by the queue.- Specified by:
getRateOfFlowin interfaceClusterQueueMBean- Returns:
- the rate of flow
-
getStatus
public String getStatus()
Description copied from interface:ClusterQueueMBeanGets the status of the queue (RUN, INIT or WAIT).- Specified by:
getStatusin interfaceClusterQueueMBean- Returns:
- the status of the queue
- See Also:
LoadingFactor.Status
-
getConsumerStatus
public String getConsumerStatus()
Description copied from interface:ClusterQueueMBeanGets consumer status (NO, NORMAL, HIGH).- Specified by:
getConsumerStatusin interfaceClusterQueueMBean- Returns:
- consumer status
- See Also:
LoadingFactor.ConsumerStatus
-
getProducerStatus
public String getProducerStatus()
Description copied from interface:ClusterQueueMBeanGets producer status (NO, NORMAL, HIGH).- Specified by:
getProducerStatusin interfaceClusterQueueMBean- Returns:
- producer status
- See Also:
LoadingFactor.ProducerStatus
-
getReceivedFromCluster
public int getReceivedFromCluster()
- Specified by:
getReceivedFromClusterin interfaceClusterQueueMBean
-
getSentToCluster
public int getSentToCluster()
- Specified by:
getSentToClusterin interfaceClusterQueueMBean
-
getEncodableClassId
public int getEncodableClassId()
Description copied from class:AgentEnables the sub classes not to implement this method.- Specified by:
getEncodableClassIdin interfaceEncodable- Overrides:
getEncodableClassIdin classQueue- Returns:
- -1
-
-