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.camel.component.timer;
018    
019    import java.util.Date;
020    import java.util.Timer;
021    import java.util.TimerTask;
022    
023    import org.apache.camel.Exchange;
024    import org.apache.camel.Processor;
025    import org.apache.camel.impl.DefaultConsumer;
026    import org.apache.commons.logging.Log;
027    import org.apache.commons.logging.LogFactory;
028    
029    /**
030     * The timer consumer.
031     *
032     * @version $Revision: 747062 $
033     */
034    public class TimerConsumer extends DefaultConsumer {
035        private static final transient Log LOG = LogFactory.getLog(TimerConsumer.class);
036        private final TimerEndpoint endpoint;
037        private TimerTask task;
038    
039        public TimerConsumer(TimerEndpoint endpoint, Processor processor) {
040            super(endpoint, processor);
041            this.endpoint = endpoint;
042        }
043    
044        @Override
045        protected void doStart() throws Exception {
046            task = new TimerTask() {
047                @Override
048                public void run() {
049                    sendTimerExchange();
050                }
051            };
052    
053            Timer timer = endpoint.getTimer();
054            configureTask(task, timer);
055        }
056    
057        @Override
058        protected void doStop() throws Exception {
059            task.cancel();
060            task = null;
061        }
062    
063        protected void configureTask(TimerTask task, Timer timer) {
064            if (endpoint.isFixedRate()) {
065                if (endpoint.getTime() != null) {
066                    timer.scheduleAtFixedRate(task, endpoint.getTime(), endpoint.getPeriod());
067                } else {
068                    timer.scheduleAtFixedRate(task, endpoint.getDelay(), endpoint.getPeriod());
069                }
070            } else {
071                if (endpoint.getTime() != null) {
072                    if (endpoint.getPeriod() >= 0) {
073                        timer.schedule(task, endpoint.getTime(), endpoint.getPeriod());
074                    } else {
075                        timer.schedule(task, endpoint.getTime());
076                    }
077                } else {
078                    if (endpoint.getPeriod() >= 0) {
079                        timer.schedule(task, endpoint.getDelay(), endpoint.getPeriod());
080                    } else {
081                        timer.schedule(task, endpoint.getDelay());
082                    }
083                }
084            }
085        }
086    
087        protected void sendTimerExchange() {
088            Exchange exchange = endpoint.createExchange();
089            exchange.setProperty(Exchange.TIMER_NAME, endpoint.getTimerName());
090            exchange.setProperty(Exchange.TIMER_TIME, endpoint.getTime());
091            exchange.setProperty(Exchange.TIMER_PERIOD, endpoint.getPeriod());
092    
093            Date now = new Date();
094            exchange.setProperty(Exchange.TIMER_FIRED_TIME, now);
095            // also set now on in header with same key as quaartz to be consistent
096            exchange.getIn().setHeader("firedTime", now);
097    
098            if (LOG.isTraceEnabled()) {
099                LOG.trace("Timer " + endpoint.getTimerName() + " is firing");
100            }
101            try {
102                getProcessor().process(exchange);
103            } catch (Exception e) {
104                handleException(e);
105            }
106        }
107    }