Class Queue

    • 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
      • logmsg

        static final Logger logmsg
      • defaultDMQId

        static AgentId defaultDMQId
        Static value holding the default DMQ identifier for a server.
      • LENIENT_REDELIVERY_DELAY

        public static final String LENIENT_REDELIVERY_DELAY
        Property allowing to delay the denied messages even if there are not considered 'redelivered' (see JMS specification). Normally, a message is considered 'redelivered' only if it has been delivered and rejected by the client. So, Messages 'denied' due to an error in the delivery mechanism are immediately re-delivered. The default value is false.
        See Also:
        Constant Field Values
      • lenientRedeliveryDelay

        private boolean lenientRedeliveryDelay
      • LENIENT_UNDELIVERABLE

        public static final String LENIENT_UNDELIVERABLE
        Property allowing to handle 'denied' messages as if they were considered 'redelivered' (see JMS specification). Normally, a message is considered 'redelivered' only if it has been delivered and rejected by the client, in this case its delivery counter is incremented. Then, a message is handled as 'undeliverable' only if it has been delivered and rejected several times by the client (the maximum number of attempts is defined by the queue's 'threshold' property). This property allows to increment the message delivery counter even when the message has been 'denied' due to an error in the delivery mechanism. The default value is false.
        See Also:
        Constant Field Values
      • lenientUndeliverable

        private boolean lenientUndeliverable
      • DELIVERY_TIMEOUT

        public static final String DELIVERY_TIMEOUT
        Property preventing unacceptable delivery times when a selector processes excessively long queues. If the time taken to retrieve a message from the queue exceeds the allotted time, the queue returns null, as if no message satisfied the query. If the timeout is negative, the queue is infinite. The default value is -1.
        See Also:
        Constant Field Values
      • DFLT_DELIVERY_TIMEOUT

        private long DFLT_DELIVERY_TIMEOUT
      • deliveryTimeout

        private long deliveryTimeout
      • threshold

        private int threshold
        Threshold above which messages are considered as undeliverable because constantly denied; 0 stands for no threshold, -1 for value not set.
      • defaultThreshold

        static int defaultThreshold
        Static value holding the default threshold for a server.
      • redeliveryDelay

        private int redeliveryDelay
        The re-delivery delay in seconds use to wait before re-delivering messages after a deny.
      • defaultRedeliveryDelay

        static int defaultRedeliveryDelay
        Static value holding the default redelivery delay for a server.
      • deliveryDelay

        private int deliveryDelay
        The delivery delay in milliseconds used to wait before delivering a message. If set the resulting delay is the max between this value and the message property.
      • pause

        private boolean pause
      • samePriorities

        private boolean samePriorities
        true if all the stored messages have the same priority. Note: messages list is ordered by priorities, so we could test if first and last message have the same priority.
      • nbExpirations

        protected int nbExpirations
        Number of stored messages with an expiration date.
      • priority

        private int priority
        Common priority value.
      • deliveryTable

        protected transient QueueDeliveryTable deliveryTable
        Table keeping the message deliveries
      • arrivalState

        protected transient QueueArrivalState arrivalState
        Counter of messages arrivals.
      • requests

        protected List<ReceiveRequest> requests
        List holding the requests before reply or expiry.
      • receiving

        protected transient boolean receiving
        true if the queue is currently handling a new received message.
      • messages

        protected transient List<Message> messages
        List holding the messages before delivery.
      • delayed

        protected transient List<Message> delayed
        List of delayed messages
      • nbMsgsDeniedSinceCreation

        protected long nbMsgsDeniedSinceCreation
      • nbMaxMsg

        protected int nbMaxMsg
        nb Max of Message store in queue (-1 no limit).
      • hprod

        long hprod
      • hcons

        long hcons
      • pload

        int pload
      • cload

        int cload
      • msgTxPrefixLength

        transient int msgTxPrefixLength
      • syncExceptionOnFullDest

        private boolean syncExceptionOnFullDest
        if true, throws an exception on sending message on full destination.
    • Constructor Detail

      • Queue

        public Queue()
        Creates a queue.
      • Queue

        protected Queue​(String name,
                        boolean fixed,
                        int stamp)
        Creates a queue with a specified stamp. It is used by the Encodable framework as the default constructor allocates a new stamp.
        Parameters:
        name - Name of topic;
        fixed - If true, topic is fixed in memory.
        stamp - Specific stamp for resulting agent.
    • Method Detail

      • getThreshold

        public int getThreshold()
        Returns the threshold value of this queue, -1 if not set.
        Specified by:
        getThreshold in interface QueueMBean
        Returns:
        the threshold value of this queue; -1 if not set.
      • setThreshold

        public void setThreshold​(int threshold)
        Sets or unsets the threshold for this queue.
        Specified by:
        setThreshold in interface QueueMBean
        Parameters:
        threshold - The threshold value to be set (-1 for unsetting previous value).
      • getDefaultThreshold

        public static int getDefaultThreshold()
        Static method returning the default threshold.
      • getDefaultDMQId

        public static AgentId getDefaultDMQId()
        Static method returning the default DMQ identifier.
      • getRedeliveryDelay

        public final int getRedeliveryDelay()
        Returns the delay in seconds use to wait before re-delivering messages after a deny.
        Specified by:
        getRedeliveryDelay in interface QueueMBean
        Returns:
        the reDeliveryDelay
      • setRedeliveryDelay

        public final void setRedeliveryDelay​(int redeliveryDelay)
        Sets the delay in seconds use to wait before re-delivering messages after a deny.
        Specified by:
        setRedeliveryDelay in interface QueueMBean
        Parameters:
        redeliveryDelay - the reDeliveryDelay to set
      • getDefaultRedeliveryDelay

        public static final int getDefaultRedeliveryDelay()
        Static method returning the default redelivery delay for a server.
      • setDefaultRedeliveryDelay

        public static final void setDefaultRedeliveryDelay​(int reDeliveryDelay)
      • getDeliveryDelay

        public final int getDeliveryDelay()
        Returns the Queue deliveryDelay in milliseconds.
        Specified by:
        getDeliveryDelay in interface QueueMBean
        Returns:
        the DeliveryDelay
      • setDeliveryDelay

        public final void setDeliveryDelay​(int deliveryDelay)
        Sets the Queue deliveryDelay in milliseconds.
        Specified by:
        setDeliveryDelay in interface QueueMBean
        Parameters:
        deliveryDelay - the deliveryDelay to set
      • setPause

        public void setPause​(boolean pause)
        Specified by:
        setPause in interface QueueMBean
      • react

        public void react​(AgentId from,
                          Notification not)
                   throws Exception
        Distributes the received notifications to the appropriate reactions.
        Overrides:
        react in class Destination
        Parameters:
        from - agent sending notification
        not - notification to react to
        Throws:
        Exception
      • agentSave

        protected void agentSave()
                          throws IOException
        Description copied from class: Agent
        Enables the sub-classes to save their state.
        Overrides:
        agentSave in class Agent
        Throws:
        IOException - if any error occurs.
      • cleanWaitingRequest

        public final void cleanWaitingRequest()
                                       throws Exception
        Removes all request that the expiration time is expired. Be careful,this method is part of the Queue MBean interface, it can be called outside the reactions of the engine thread. In order to avoid synchronization issues we should avoid direct manipulation of requests list (see JORAM-372, JORAM-373).
        Specified by:
        cleanWaitingRequest in interface QueueMBean
        Throws:
        Exception
        InterruptedException
      • 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 final int getWaitingRequestCount()
        Returns the number of waiting requests in the queue.
        Specified by:
        getWaitingRequestCount in interface QueueMBean
        Returns:
        The number of waiting requests.
      • getDelayedMessageCount

        public final int getDelayedMessageCount()
        Returns the number of messages waiting for a delay.
        Specified by:
        getDelayedMessageCount in interface QueueMBean
        Returns:
        The number of messages waiting for a delay.
      • cleanPendingMessage

        public final void cleanPendingMessage()
                                       throws Exception
        Removes all messages that the time-to-live is expired. Be careful,this method is part of the Queue MBean interface, it can be called outside the reactions of the engine thread. In order to avoid synchronization issues we should avoid direct manipulation of messages list (see JORAM-372, JORAM-373).
        Specified by:
        cleanPendingMessage in interface QueueMBean
        Throws:
        Exception
        InterruptedException
      • 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 final int getPendingMessageCount()
        Returns the number of pending messages in the queue.
        Specified by:
        getPendingMessageCount in interface QueueMBean
        Returns:
        The number of pending messages.
      • getDeliveredMessageCount

        public final int getDeliveredMessageCount()
        Returns the number of messages delivered and waiting for acknowledge.
        Specified by:
        getDeliveredMessageCount in interface QueueMBean
        Returns:
        The number of messages delivered.
      • getNbMsgsDeniedSinceCreation

        public final long getNbMsgsDeniedSinceCreation()
        Returns the number of messages denied since creation time of this destination.
        Specified by:
        getNbMsgsDeniedSinceCreation in interface QueueMBean
        Returns:
        the number of messages delivered since creation time.
      • getNbMsgsDeliverSinceCreation

        public long getNbMsgsDeliverSinceCreation()
        Description copied from class: Destination
        Returns the number of messages delivered since creation time of this destination. It includes messages all delivered messages to a consumer, already acknowledged or not.
        Specified by:
        getNbMsgsDeliverSinceCreation in interface DestinationMBean
        Overrides:
        getNbMsgsDeliverSinceCreation in class Destination
        Returns:
        the number of messages delivered since creation time.
      • getNbMaxMsg

        public final 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 QueueMBean
        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 QueueMBean
        Parameters:
        nbMaxMsg - the maximum number of message (-1 set no limit).
      • initialize

        protected void initialize​(boolean firstTime)
                           throws Exception
        Initializes the destination.
        Specified by:
        initialize in class Destination
        Parameters:
        firstTime - true when first called by the factory
        Throws:
        Exception
      • finalize

        protected void finalize​(boolean last)
        Finalizes the destination before it is garbaged.
        Specified by:
        finalize in class Destination
        Parameters:
        last - true if the destination is deleted
      • getProducerLoad

        public long getProducerLoad()
        Return the average producer's load during last moments.
        Specified by:
        getProducerLoad in interface QueueMBean
      • getConsumerLoad

        public long getConsumerLoad()
        Return the average consumer's load during last moments.
        Specified by:
        getConsumerLoad in interface QueueMBean
      • isValidJMXAttribute

        protected boolean isValidJMXAttribute​(String attrName)
        This method allows to exclude some JMX attribute of getJMXStatistics method. It excludes.
        Overrides:
        isValidJMXAttribute in class Destination
        Parameters:
        attrName - name of attribute to test.
        Returns:
        true if the attribute is a valid one.
      • receiveRequest

        protected void receiveRequest​(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

        protected void browseRequest​(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

        protected void acknowledgeRequest​(AcknowledgeRequest not)
        Method implementing the reaction to an AcknowledgeRequest instance, requesting messages to be acknowledged.
      • acknowledge

        private void acknowledge​(String msgId)
      • denyRequest

        protected void denyRequest​(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.

      • denyMessage

        private DMQManager denyMessage​(String msgid,
                                       boolean redelivered,
                                       DMQManager dmqManager)
        Denies the specified message.
        Parameters:
        msgid - The identifier of message to deny.
        redelivered - True if the message has been denied from user.
        dmqManager - The DMQManager if it exists, null otherwise.
        Returns:
        The DMQManager if it exists, null otherwise.
      • logmsg

        private void logmsg​(Level loggerLevel,
                            Level logmsgLevel,
                            String logMessage)
        Logs message either in logmsg if configured, or in logger.
        Parameters:
        loggerLevel -
        logmsgLevel -
        logMessage -
      • getQueueMessageIds

        private void getQueueMessageIds​(AgentId replyTo,
                                        String requestMsgId,
                                        String replyMsgId)
      • clear

        public void clear()
        Removes all pending messages.
        Specified by:
        clear in interface QueueMBean
      • clearQueue

        private void clearQueue​(AgentId replyTo,
                                String requestMsgId,
                                String replyMsgId)
      • doRightRequest

        protected void doRightRequest​(AgentId user,
                                      int right)
        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 Destination
        Parameters:
        user - The user about right modification.
        right - The right modification.
      • doClientMessages

        protected void doClientMessages​(AgentId from,
                                        ClientMessages not,
                                        boolean throwsExceptionOnFullDest)
                                 throws AccessException
        Method specifically processing a ClientMessages instance.

        This method stores the messages and launches a delivery sequence. This method is used when ClientMessages comes from a JMS client and/or LB notification in ClusterQueue (May be we should use addClienMessages). It is also used in old deprecated JMS bridge.

        Specified by:
        doClientMessages in class Destination
        Throws:
        AccessException
      • getIdString

        private static final String getIdString​(Message msg)
      • addDeliveryTimeMessage

        void addDeliveryTimeMessage​(Message msg,
                                    boolean throwsExceptionOnFullDest,
                                    boolean isHeader)
                             throws AccessException
        Adds the specified message to the list of delayed message.
        Parameters:
        msg - The message to add.
        throwsExceptionOnFullDest - if true, can throws an exception if destination is full.
        isHeader - If true, only save the message header (the message has already been saved).
        Throws:
        AccessException
      • doUnknownAgent

        protected void doUnknownAgent​(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 Destination
      • doDeleteNot

        protected void doDeleteNot​(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 Destination
      • getMsgTxPrefix

        protected final StringBuffer getMsgTxPrefix()
      • setMsgTxName

        protected final void setMsgTxName​(Message msg)
      • storeMessage

        protected final void storeMessage​(Message msg,
                                          boolean throwsExceptionOnFullDest)
                                   throws AccessException
        Actually stores a message in the deliverables list.
        Parameters:
        msg - The message to store.
        throwsExceptionOnFullDest - true, can throws an exception on sending message on full destination
        Throws:
        AccessException
      • isSyncExceptionOnFullDest

        public boolean isSyncExceptionOnFullDest()
        Returns:
        the syncExceptionOnFullDest
      • setSyncExceptionOnFullDest

        public void setSyncExceptionOnFullDest​(boolean syncExceptionOnFullDest)
        Parameters:
        syncExceptionOnFullDest - the syncExceptionOnFullDest to set
      • addMessage

        protected final boolean addMessage​(Message message,
                                           boolean throwsExceptionOnFullDest)
                                    throws AccessException
        Adds a message in the list of messages to deliver. This method take care of the message priority if needed.
        Parameters:
        message - the message to add.
        throwsExceptionOnFullDest - true, can throws an exception on sending message on full destination
        Returns:
        true if the message has been added. false if the queue is full.
        Throws:
        AccessException - If syncExceptionOnFullDest and the queue isFull
      • getClientMessages

        protected ClientMessages getClientMessages​(int nb,
                                                   String selector,
                                                   boolean remove)
        Get a client message contain nb messages. Only used in ClusterQueue.
        Parameters:
        nb - number of messages returned in ClientMessage.
        selector - jms selector
        remove - delete all messages returned if true
        Returns:
        ClientMessages (contains nb Messages)
      • getMessages

        private List<Message> getMessages​(int nb,
                                          String selector,
                                          boolean remove)
        get messages, if it's possible.
        Parameters:
        nb - Maximum number of messages to return, if -1 returns all messages.
        selector - JMS selector.
        remove -
        Returns:
        List of mom messages.
      • getMomMessage

        private Message getMomMessage​(String msgId)
      • getQueueMessage

        protected Message getQueueMessage​(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 CompositeData getMessage​(String msgId)
                                 throws Exception
        Returns the description of a particular pending message. The message is pointed out through its unique identifier.
        Specified by:
        getMessage in interface QueueMBean
        Parameters:
        msgId - The unique message's identifier.
        Returns:
        the description of the message.
        Throws:
        Exception
        See Also:
        MessageJMXWrapper
      • getMessagesView

        public List<? extends MessageView> getMessagesView()
        Only used from shell-mom MOMCommandsImpl.
      • getDelayedMessage

        public CompositeData getDelayedMessage​(String msgId)
                                        throws Exception
        Returns the description of a particular delayed message. The message is pointed out through its unique identifier.
        Specified by:
        getDelayedMessage in interface QueueMBean
        Parameters:
        msgId - The unique message's identifier.
        Returns:
        the description of the message.
        Throws:
        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​(String msgId)
        call in deliverMessages just after forward(msg), overload this method to process a specific treatment.
      • messageRemoved

        protected void messageRemoved​(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).
      • addClientMessages

        public void addClientMessages​(ClientMessages clientMsgs,
                                      boolean throwsExceptionOnFullDest)
                               throws AccessException
        Adds the client messages in the queue. This method is used when messages does not come from a JMS Client, for example for AcquisitionQueue, AliasQueue and AliasInQueue.
        Parameters:
        clientMsgs - client message notification.
        throwsExceptionOnFullDest - true, can throws an exception on sending message on full destination
        Throws:
        AccessException
      • handleExpiredNot

        protected void handleExpiredNot​(AgentId from,
                                        ExpiredNot not)
      • doExportMessages

        private void doExportMessages​(String dirpath,
                                      String selector,
                                      boolean binary)
                               throws IOException
        Exports all messages in the specified directory. Be careful, this method needs to be called from an agent reaction.
        Throws:
        IOException