001    /**
002     * Licensed to the Apache Software Foundation (ASF) under one or more
003     * contributor license agreements.  See the NOTICE file distributed with
004     * this work for additional information regarding copyright ownership.
005     * The ASF licenses this file to You under the Apache License, Version 2.0
006     * (the "License"); you may not use this file except in compliance with
007     * the License.  You may obtain a copy of the License at
008     *
009     *      http://www.apache.org/licenses/LICENSE-2.0
010     *
011     * Unless required by applicable law or agreed to in writing, software
012     * distributed under the License is distributed on an "AS IS" BASIS,
013     * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
014     * See the License for the specific language governing permissions and
015     * limitations under the License.
016     */
017    package org.apache.activemq.broker.util;
018    
019    import org.apache.activemq.broker.BrokerPluginSupport;
020    import org.apache.activemq.broker.ProducerBrokerExchange;
021    import org.apache.activemq.command.Message;
022    import org.slf4j.Logger;
023    import org.slf4j.LoggerFactory;
024    
025    /**
026     * A Broker interceptor which updates a JMS Client's timestamp on the message
027     * with a broker timestamp. Useful when the clocks on client machines are known
028     * to not be correct and you can only trust the time set on the broker machines.
029     * 
030     * Enabling this plugin will break JMS compliance since the timestamp that the
031     * producer sees on the messages after as send() will be different from the
032     * timestamp the consumer will observe when he receives the message. This plugin
033     * is not enabled in the default ActiveMQ configuration.
034     * 
035     * 2 new attributes have been added which will allow the administrator some override control
036     * over the expiration time for incoming messages:
037     *
038     * Attribute 'zeroExpirationOverride' can be used to apply an expiration
039     * time to incoming messages with no expiration defined (messages that would never expire)
040     *
041     * Attribute 'ttlCeiling' can be used to apply a limit to the expiration time
042     *
043     * @org.apache.xbean.XBean element="timeStampingBrokerPlugin"
044     * 
045     * 
046     */
047    public class TimeStampingBrokerPlugin extends BrokerPluginSupport {
048        private static final Logger LOG = LoggerFactory.getLogger(TimeStampingBrokerPlugin.class);
049        /** 
050        * variable which (when non-zero) is used to override
051        * the expiration date for messages that arrive with
052        * no expiration date set (in Milliseconds).
053        */
054        long zeroExpirationOverride = 0;
055    
056        /** 
057        * variable which (when non-zero) is used to limit
058        * the expiration date (in Milliseconds).  
059        */
060        long ttlCeiling = 0;
061        
062        /**
063         * If true, the plugin will not update timestamp to past values
064         * False by default
065         */
066        boolean futureOnly = false;
067        
068        
069        /**
070         * if true, update timestamp even if message has passed through a network
071         * default false
072         */
073        boolean processNetworkMessages = false;
074    
075        /** 
076        * setter method for zeroExpirationOverride
077        */
078        public void setZeroExpirationOverride(long ttl)
079        {
080            this.zeroExpirationOverride = ttl;
081        }
082    
083        /** 
084        * setter method for ttlCeiling
085        */
086        public void setTtlCeiling(long ttlCeiling)
087        {
088            this.ttlCeiling = ttlCeiling;
089        }
090    
091            public void setFutureOnly(boolean futureOnly) {
092                    this.futureOnly = futureOnly;
093            }
094            
095            public void setProcessNetworkMessages(Boolean processNetworkMessages) {
096                this.processNetworkMessages = processNetworkMessages;
097            }
098    
099            @Override
100        public void send(ProducerBrokerExchange producerExchange, Message message) throws Exception {
101            if (message.getTimestamp() > 0
102                && (processNetworkMessages || (message.getBrokerPath() == null || message.getBrokerPath().length == 0))) {
103                // timestamp not been disabled and has not passed through a network or processNetworkMessages=true
104                long oldExpiration = message.getExpiration();
105                long newTimeStamp = System.currentTimeMillis();
106                long timeToLive = zeroExpirationOverride;
107                long oldTimestamp = message.getTimestamp();
108                if (oldExpiration > 0) {
109                    timeToLive = oldExpiration - oldTimestamp;
110                }
111                if (timeToLive > 0 && ttlCeiling > 0 && timeToLive > ttlCeiling) {
112                    timeToLive = ttlCeiling;
113                }
114                long expiration = timeToLive + newTimeStamp;
115                            //In the scenario that the Broker is behind the clients we never want to set the Timestamp and Expiration in the past 
116                            if(!futureOnly || (expiration > oldExpiration)) {
117                                    if (timeToLive > 0 && expiration > 0) {
118                                            message.setExpiration(expiration);
119                                    }
120                                    message.setTimestamp(newTimeStamp);
121                                    if (LOG.isDebugEnabled()) {
122                                        LOG.debug("Set message " + message.getMessageId() + " timestamp from " + oldTimestamp + " to " + newTimeStamp);
123                                    }
124                            }
125            }
126            super.send(producerExchange, message);
127        }
128    }