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.impl;
018
019import java.util.EventObject;
020import java.util.LinkedHashSet;
021import java.util.Set;
022import java.util.concurrent.locks.Lock;
023import java.util.concurrent.locks.ReentrantLock;
024
025import org.apache.camel.CamelContext;
026import org.apache.camel.CamelContextAware;
027import org.apache.camel.Consumer;
028import org.apache.camel.Exchange;
029import org.apache.camel.LoggingLevel;
030import org.apache.camel.Route;
031import org.apache.camel.management.event.ExchangeCompletedEvent;
032import org.apache.camel.support.EventNotifierSupport;
033import org.apache.camel.support.RoutePolicySupport;
034import org.apache.camel.util.CamelLogger;
035import org.apache.camel.util.ObjectHelper;
036import org.apache.camel.util.ServiceHelper;
037import org.slf4j.LoggerFactory;
038
039/**
040 * A throttle based {@link org.apache.camel.spi.RoutePolicy} which is capable of dynamic
041 * throttling a route based on number of current inflight exchanges.
042 * <p/>
043 * This implementation supports two scopes {@link ThrottlingScope#Context} and {@link ThrottlingScope#Route} (is default).
044 * If context scope is selected then this implementation will use a {@link org.apache.camel.spi.EventNotifier} to listen
045 * for events when {@link Exchange}s is done, and trigger the {@link #throttle(org.apache.camel.Route, org.apache.camel.Exchange)}
046 * method. If the route scope is selected then <b>no</b> {@link org.apache.camel.spi.EventNotifier} is in use, as there is already
047 * a {@link org.apache.camel.spi.Synchronization} callback on the current {@link Exchange} which triggers the
048 * {@link #throttle(org.apache.camel.Route, org.apache.camel.Exchange)} when the current {@link Exchange} is done.
049 *
050 * @version 
051 */
052public class ThrottlingInflightRoutePolicy extends RoutePolicySupport implements CamelContextAware {
053
054    public enum ThrottlingScope {
055        Context, Route
056    }
057
058    private final Set<Route> routes = new LinkedHashSet<Route>();
059    private ContextScopedEventNotifier eventNotifier;
060    private CamelContext camelContext;
061    private final Lock lock = new ReentrantLock();
062    private ThrottlingScope scope = ThrottlingScope.Route;
063    private int maxInflightExchanges = 1000;
064    private int resumePercentOfMax = 70;
065    private int resumeInflightExchanges = 700;
066    private LoggingLevel loggingLevel = LoggingLevel.INFO;
067    private CamelLogger logger;
068
069    public ThrottlingInflightRoutePolicy() {
070    }
071
072    @Override
073    public String toString() {
074        return "ThrottlingInflightRoutePolicy[" + maxInflightExchanges + " / " + resumePercentOfMax + "% using scope " + scope + "]";
075    }
076
077    public CamelContext getCamelContext() {
078        return camelContext;
079    }
080
081    public void setCamelContext(CamelContext camelContext) {
082        this.camelContext = camelContext;
083    }
084
085    @Override
086    public void onInit(Route route) {
087        // we need to remember the routes we apply for
088        routes.add(route);
089    }
090
091    @Override
092    public void onExchangeDone(Route route, Exchange exchange) {
093        // if route scoped then throttle directly
094        // as context scoped is handled using an EventNotifier instead
095        if (scope == ThrottlingScope.Route) {
096            throttle(route, exchange);
097        }
098    }
099
100    /**
101     * Throttles the route when {@link Exchange}s is done.
102     *
103     * @param route  the route
104     * @param exchange the exchange
105     */
106    protected void throttle(Route route, Exchange exchange) {
107        // this works the best when this logic is executed when the exchange is done
108        Consumer consumer = route.getConsumer();
109
110        int size = getSize(route, exchange);
111        boolean stop = maxInflightExchanges > 0 && size > maxInflightExchanges;
112        if (log.isTraceEnabled()) {
113            log.trace("{} > 0 && {} > {} evaluated as {}", new Object[]{maxInflightExchanges, size, maxInflightExchanges, stop});
114        }
115        if (stop) {
116            try {
117                lock.lock();
118                stopConsumer(size, consumer);
119            } catch (Exception e) {
120                handleException(e);
121            } finally {
122                lock.unlock();
123            }
124        }
125
126        // reload size in case a race condition with too many at once being invoked
127        // so we need to ensure that we read the most current size and start the consumer if we are already to low
128        size = getSize(route, exchange);
129        boolean start = size <= resumeInflightExchanges;
130        if (log.isTraceEnabled()) {
131            log.trace("{} <= {} evaluated as {}", new Object[]{size, resumeInflightExchanges, start});
132        }
133        if (start) {
134            try {
135                lock.lock();
136                startConsumer(size, consumer);
137            } catch (Exception e) {
138                handleException(e);
139            } finally {
140                lock.unlock();
141            }
142        }
143    }
144
145    public int getMaxInflightExchanges() {
146        return maxInflightExchanges;
147    }
148
149    /**
150     * Sets the upper limit of number of concurrent inflight exchanges at which point reached
151     * the throttler should suspend the route.
152     * <p/>
153     * Is default 1000.
154     *
155     * @param maxInflightExchanges the upper limit of concurrent inflight exchanges
156     */
157    public void setMaxInflightExchanges(int maxInflightExchanges) {
158        this.maxInflightExchanges = maxInflightExchanges;
159        // recalculate, must be at least at 1
160        this.resumeInflightExchanges = Math.max(resumePercentOfMax * maxInflightExchanges / 100, 1);
161    }
162
163    public int getResumePercentOfMax() {
164        return resumePercentOfMax;
165    }
166
167    /**
168     * Sets at which percentage of the max the throttler should start resuming the route.
169     * <p/>
170     * Will by default use 70%.
171     *
172     * @param resumePercentOfMax the percentage must be between 0 and 100
173     */
174    public void setResumePercentOfMax(int resumePercentOfMax) {
175        if (resumePercentOfMax < 0 || resumePercentOfMax > 100) {
176            throw new IllegalArgumentException("Must be a percentage between 0 and 100, was: " + resumePercentOfMax);
177        }
178
179        this.resumePercentOfMax = resumePercentOfMax;
180        // recalculate, must be at least at 1
181        this.resumeInflightExchanges = Math.max(resumePercentOfMax * maxInflightExchanges / 100, 1);
182    }
183
184    public ThrottlingScope getScope() {
185        return scope;
186    }
187
188    /**
189     * Sets which scope the throttling should be based upon, either route or total scoped.
190     *
191     * @param scope the scope
192     */
193    public void setScope(ThrottlingScope scope) {
194        this.scope = scope;
195    }
196
197    public LoggingLevel getLoggingLevel() {
198        return loggingLevel;
199    }
200
201    public CamelLogger getLogger() {
202        if (logger == null) {
203            logger = createLogger();
204        }
205        return logger;
206    }
207
208    /**
209     * Sets the logger to use for logging throttling activity.
210     *
211     * @param logger the logger
212     */
213    public void setLogger(CamelLogger logger) {
214        this.logger = logger;
215    }
216
217    /**
218     * Sets the logging level to report the throttling activity.
219     * <p/>
220     * Is default <tt>INFO</tt> level.
221     *
222     * @param loggingLevel the logging level
223     */
224    public void setLoggingLevel(LoggingLevel loggingLevel) {
225        this.loggingLevel = loggingLevel;
226    }
227
228    protected CamelLogger createLogger() {
229        return new CamelLogger(LoggerFactory.getLogger(ThrottlingInflightRoutePolicy.class), getLoggingLevel());
230    }
231
232    private int getSize(Route route, Exchange exchange) {
233        if (scope == ThrottlingScope.Context) {
234            return exchange.getContext().getInflightRepository().size();
235        } else {
236            return exchange.getContext().getInflightRepository().size(route.getId());
237        }
238    }
239
240    private void startConsumer(int size, Consumer consumer) throws Exception {
241        boolean started = super.startConsumer(consumer);
242        if (started) {
243            getLogger().log("Throttling consumer: " + size + " <= " + resumeInflightExchanges + " inflight exchange by resuming consumer: " + consumer);
244        }
245    }
246
247    private void stopConsumer(int size, Consumer consumer) throws Exception {
248        boolean stopped = super.stopConsumer(consumer);
249        if (stopped) {
250            getLogger().log("Throttling consumer: " + size + " > " + maxInflightExchanges + " inflight exchange by suspending consumer: " + consumer);
251        }
252    }
253
254    @Override
255    protected void doStart() throws Exception {
256        ObjectHelper.notNull(camelContext, "CamelContext", this);
257        if (scope == ThrottlingScope.Context) {
258            eventNotifier = new ContextScopedEventNotifier();
259            // must start the notifier before it can be used
260            ServiceHelper.startService(eventNotifier);
261            // we are in context scope, so we need to use an event notifier to keep track
262            // when any exchanges is done on the camel context.
263            // This ensures we can trigger accordingly to context scope
264            camelContext.getManagementStrategy().addEventNotifier(eventNotifier);
265        }
266    }
267
268    @Override
269    protected void doStop() throws Exception {
270        ObjectHelper.notNull(camelContext, "CamelContext", this);
271        if (scope == ThrottlingScope.Context) {
272            camelContext.getManagementStrategy().removeEventNotifier(eventNotifier);
273        }
274    }
275
276    /**
277     * {@link org.apache.camel.spi.EventNotifier} to keep track on when {@link Exchange}
278     * is done, so we can throttle accordingly.
279     */
280    private class ContextScopedEventNotifier extends EventNotifierSupport {
281
282        @Override
283        public void notify(EventObject event) throws Exception {
284            ExchangeCompletedEvent completedEvent = (ExchangeCompletedEvent) event;
285            for (Route route : routes) {
286                throttle(route, completedEvent.getExchange());
287            }
288        }
289
290        @Override
291        public boolean isEnabled(EventObject event) {
292            return event instanceof ExchangeCompletedEvent;
293        }
294
295        @Override
296        protected void doStart() throws Exception {
297            // noop
298        }
299
300        @Override
301        protected void doStop() throws Exception {
302            // noop
303        }
304
305        @Override
306        public String toString() {
307            return "ContextScopedEventNotifier";
308        }
309    }
310
311}