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 */
017package org.apache.camel.component.scheduler;
018
019import java.util.Collection;
020import java.util.HashMap;
021import java.util.Map;
022import java.util.concurrent.ScheduledExecutorService;
023import java.util.concurrent.atomic.AtomicInteger;
024
025import org.apache.camel.Endpoint;
026import org.apache.camel.impl.UriEndpointComponent;
027
028public class SchedulerComponent extends UriEndpointComponent {
029
030    private final Map<String, ScheduledExecutorService> executors = new HashMap<String, ScheduledExecutorService>();
031    private final Map<String, AtomicInteger> refCounts = new HashMap<String, AtomicInteger>();
032
033    private int concurrentTasks = 1;
034
035    public SchedulerComponent() {
036        super(SchedulerEndpoint.class);
037    }
038
039    @Override
040    protected Endpoint createEndpoint(String uri, String remaining, Map<String, Object> parameters) throws Exception {
041        SchedulerEndpoint answer = new SchedulerEndpoint(uri, this, remaining);
042        answer.setConcurrentTasks(getConcurrentTasks());
043        setProperties(answer, parameters);
044        return answer;
045    }
046
047    public int getConcurrentTasks() {
048        return concurrentTasks;
049    }
050
051    /**
052     * Number of threads used by the scheduling thread pool.
053     * <p/>
054     * Is by default using a single thread
055     */
056    public void setConcurrentTasks(int concurrentTasks) {
057        this.concurrentTasks = concurrentTasks;
058    }
059
060    protected ScheduledExecutorService addConsumer(SchedulerConsumer consumer) {
061        String name = consumer.getEndpoint().getName();
062        int concurrentTasks = consumer.getEndpoint().getConcurrentTasks();
063
064        ScheduledExecutorService answer;
065        synchronized (executors) {
066            answer = executors.get(name);
067            if (answer == null) {
068                answer = getCamelContext().getExecutorServiceManager().newScheduledThreadPool(this, "scheduler://" + name, concurrentTasks);
069                executors.put(name, answer);
070                // store new reference counter
071                refCounts.put(name, new AtomicInteger(1));
072            } else {
073                // increase reference counter
074                AtomicInteger counter = refCounts.get(name);
075                if (counter != null) {
076                    counter.incrementAndGet();
077                }
078            }
079        }
080        return answer;
081    }
082
083    protected void removeConsumer(SchedulerConsumer consumer) {
084        String name = consumer.getEndpoint().getName();
085
086        synchronized (executors) {
087            // decrease reference counter
088            AtomicInteger counter = refCounts.get(name);
089            if (counter != null && counter.decrementAndGet() <= 0) {
090                refCounts.remove(name);
091                // remove scheduler as its no longer in use
092                ScheduledExecutorService scheduler = executors.remove(name);
093                if (scheduler != null) {
094                    getCamelContext().getExecutorServiceManager().shutdown(scheduler);
095                }
096            }
097        }
098    }
099
100    @Override
101    protected void doStop() throws Exception {
102        Collection<ScheduledExecutorService> collection = executors.values();
103        for (ScheduledExecutorService scheduler : collection) {
104            getCamelContext().getExecutorServiceManager().shutdown(scheduler);
105        }
106        executors.clear();
107        refCounts.clear();
108    }
109
110}