Class ClusterQueue

    • 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, ...
      • timeTable

        private Map<String,​Long> timeTable
        key = msgId value = date
      • visitTable

        private Map<String,​List<AgentId>> visitTable
        key = msgId value = List (alreadyVisit)
      • 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 to Queue.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
    • Constructor Detail

      • ClusterQueue

        public ClusterQueue()
    • Method Detail

      • setProperties

        public void setProperties​(Properties prop,
                                  boolean firstTime)
                           throws Exception
        Configures a ClusterQueue instance.
        Overrides:
        setProperties in class Queue
        Parameters:
        prop - The initial set of properties.
        Throws:
        Exception
      • initialize

        public void initialize​(boolean firstTime)
                        throws Exception
        Initializes the destination.
        Overrides:
        initialize in class Queue
        Parameters:
        firstTime - true when first called by the factory
        Throws:
        Exception
      • react

        public void react​(AgentId from,
                          Notification not)
                   throws Exception
        Distributes the received notifications to the appropriate reactions.
        Overrides:
        react in class Queue
        Parameters:
        from - agent sending notification
        not - notification to react to
        Throws:
        Exception
      • 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 a ClusterJoinNot notification, 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 a ClusterJoinAck notification, doing a transitive closure with the current cluster and the one of the new cluster element.
      • clusterList

        private List<String> clusterList()
        Returns the cluster list.
        Returns:
        the cluster list.
      • 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:
        preProcess in class Destination
        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:
        postProcess in class Destination
        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.
        Overrides:
        wakeUpNot in class Queue
        Parameters:
        not -
      • 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: Queue
        Method implementing the reaction to a ReceiveRequest instance, requesting a message.

        This method stores the request and launches a delivery sequence.

        Overrides:
        receiveRequest in class Queue
        Parameters:
        not - ReceiveRequest
        Throws:
        AccessException - If the sender is not a reader.
      • 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:
        getClientMessages in class Queue
        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​(String msgId,
                                          boolean remove)
        get mom message, delete if remove = true. add cluster monitoring value.
        Overrides:
        getQueueMessage in class Queue
        Parameters:
        msgId - message identification
        remove - 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: Queue
        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.

        Overrides:
        doDeleteNot in class Queue
      • doUnknownAgent

        protected void doUnknownAgent​(UnknownAgent uA)
        Description copied from class: Queue
        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.

        Overrides:
        doUnknownAgent in class Queue
      • 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: Queue
        call in deliverMessages just after forward(msg), overload this method to process a specific treatment.
        Overrides:
        messageDelivered in class Queue
        Parameters:
        msgId -
      • monitoringMsgSendToCluster

        protected void monitoringMsgSendToCluster​(String msgId)
        Parameters:
        msgId -
      • setWaitAfterClusterReq

        public void setWaitAfterClusterReq​(long waitAfterClusterReq)
        Description copied from interface: ClusterQueueMBean
        Sets the time (in ms) during which a queue which requested something from the cluster is not authorized to do it again.
        Specified by:
        setWaitAfterClusterReq in interface ClusterQueueMBean
        Parameters:
        waitAfterClusterReq -
      • setProducThreshold

        public void setProducThreshold​(int producThreshold)
        Description copied from interface: ClusterQueueMBean
        Sets the number of messages above which a queue is considered loaded.
        Specified by:
        setProducThreshold in interface ClusterQueueMBean
        Parameters:
        producThreshold -
      • setConsumThreshold

        public void setConsumThreshold​(int consumThreshold)
        Description copied from interface: ClusterQueueMBean
        Sets the number of pending "receive" requests above which a queue will request messages from the other queues of the cluster.
        Specified by:
        setConsumThreshold in interface ClusterQueueMBean
        Parameters:
        consumThreshold -
      • setAutoEvalThreshold

        public void setAutoEvalThreshold​(boolean autoEvalThreshold)
        Description copied from interface: ClusterQueueMBean
        Automatic reevaluation of the queues' thresholds can be done according to their activity.
        Specified by:
        setAutoEvalThreshold in interface ClusterQueueMBean
        Parameters:
        autoEvalThreshold -
      • getProducThreshold

        public int getProducThreshold()
        Description copied from interface: ClusterQueueMBean
        Gets the number of messages above which a queue is considered loaded.
        Specified by:
        getProducThreshold in interface ClusterQueueMBean
        Returns:
        the produce threshold
      • getConsumThreshold

        public int getConsumThreshold()
        Description copied from interface: ClusterQueueMBean
        Gets the number of pending "receive" requests above which a queue will request messages from the other queues of the cluster.
        Specified by:
        getConsumThreshold in interface ClusterQueueMBean
        Returns:
        the consume threshold
      • isAutoEvalThreshold

        public boolean isAutoEvalThreshold()
        Description copied from interface: ClusterQueueMBean
        True if an automatic reevaluation of the queues' thresholds values is allowed according to their activity.
        Specified by:
        isAutoEvalThreshold in interface ClusterQueueMBean
        Returns:
        true if auto evaluation of thresholds is allowed.
      • getWaitAfterClusterReq

        public long getWaitAfterClusterReq()
        Description copied from interface: ClusterQueueMBean
        Gets the time (in ms) during which a queue which requested something from the cluster is not authorized to do it again.
        Specified by:
        getWaitAfterClusterReq in interface ClusterQueueMBean
        Returns:
        the minimum time to wait before another cluster request.
      • getRateOfFlow

        public float getRateOfFlow()
        Description copied from interface: ClusterQueueMBean
        Gets an evaluation of the flow of messages handled by the queue.
        Specified by:
        getRateOfFlow in interface ClusterQueueMBean
        Returns:
        the rate of flow