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.ArrayList;
020import java.util.Collection;
021import java.util.HashMap;
022import java.util.List;
023import java.util.Map;
024import java.util.concurrent.atomic.AtomicInteger;
025
026import org.apache.camel.CamelContext;
027import org.apache.camel.Endpoint;
028import org.apache.camel.NoSuchEndpointException;
029import org.apache.camel.Processor;
030import org.apache.camel.Route;
031import org.apache.camel.RuntimeCamelException;
032import org.apache.camel.ShutdownRoute;
033import org.apache.camel.ShutdownRunningTask;
034import org.apache.camel.model.FromDefinition;
035import org.apache.camel.model.ProcessorDefinition;
036import org.apache.camel.model.RouteDefinition;
037import org.apache.camel.processor.CamelInternalProcessor;
038import org.apache.camel.processor.Pipeline;
039import org.apache.camel.spi.InterceptStrategy;
040import org.apache.camel.spi.RouteContext;
041import org.apache.camel.spi.RoutePolicy;
042import org.apache.camel.util.CamelContextHelper;
043import org.apache.camel.util.ObjectHelper;
044
045/**
046 * The context used to activate new routing rules
047 *
048 * @version 
049 */
050public class DefaultRouteContext implements RouteContext {
051    private final Map<ProcessorDefinition<?>, AtomicInteger> nodeIndex = new HashMap<ProcessorDefinition<?>, AtomicInteger>();
052    private final RouteDefinition route;
053    private FromDefinition from;
054    private final Collection<Route> routes;
055    private Endpoint endpoint;
056    private final List<Processor> eventDrivenProcessors = new ArrayList<Processor>();
057    private CamelContext camelContext;
058    private List<InterceptStrategy> interceptStrategies = new ArrayList<InterceptStrategy>();
059    private InterceptStrategy managedInterceptStrategy;
060    private boolean routeAdded;
061    private Boolean trace;
062    private Boolean messageHistory;
063    private Boolean streamCache;
064    private Boolean handleFault;
065    private Long delay;
066    private Boolean autoStartup = Boolean.TRUE;
067    private List<RoutePolicy> routePolicyList = new ArrayList<RoutePolicy>();
068    private ShutdownRoute shutdownRoute;
069    private ShutdownRunningTask shutdownRunningTask;
070
071    public DefaultRouteContext(CamelContext camelContext, RouteDefinition route, FromDefinition from, Collection<Route> routes) {
072        this.camelContext = camelContext;
073        this.route = route;
074        this.from = from;
075        this.routes = routes;
076    }
077
078    /**
079     * Only used for lazy construction from inside ExpressionType
080     */
081    public DefaultRouteContext(CamelContext camelContext) {
082        this.camelContext = camelContext;
083        this.routes = new ArrayList<Route>();
084        this.route = new RouteDefinition("temporary");
085    }
086
087    public Endpoint getEndpoint() {
088        if (endpoint == null) {
089            endpoint = from.resolveEndpoint(this);
090        }
091        return endpoint;
092    }
093
094    public FromDefinition getFrom() {
095        return from;
096    }
097
098    public RouteDefinition getRoute() {
099        return route;
100    }
101
102    public CamelContext getCamelContext() {
103        return camelContext;
104    }
105
106    public Endpoint resolveEndpoint(String uri) {
107        return route.resolveEndpoint(getCamelContext(), uri);
108    }
109
110    public Endpoint resolveEndpoint(String uri, String ref) {
111        Endpoint endpoint = null;
112        if (uri != null) {
113            endpoint = resolveEndpoint(uri);
114            if (endpoint == null) {
115                throw new NoSuchEndpointException(uri);
116            }
117        }
118        if (ref != null) {
119            endpoint = lookup(ref, Endpoint.class);
120            if (endpoint == null) {
121                throw new NoSuchEndpointException("ref:" + ref, "check your camel registry with id " + ref);
122            }
123            // Check the endpoint has the right CamelContext 
124            if (!this.getCamelContext().equals(endpoint.getCamelContext())) {
125                throw new NoSuchEndpointException("ref:" + ref, "make sure the endpoint has the same camel context as the route does.");
126            }
127            try {
128                // need add the endpoint into service
129                getCamelContext().addService(endpoint);
130            } catch (Exception ex) {
131                throw new RuntimeCamelException(ex);
132            }
133        }
134        if (endpoint == null) {
135            throw new IllegalArgumentException("Either 'uri' or 'ref' must be specified on: " + this);
136        } else {
137            return endpoint;
138        }
139    }
140
141    public <T> T lookup(String name, Class<T> type) {
142        return getCamelContext().getRegistry().lookupByNameAndType(name, type);
143    }
144
145    public <T> Map<String, T> lookupByType(Class<T> type) {
146        return getCamelContext().getRegistry().findByTypeWithName(type);
147    }
148
149    @Override
150    public <T> T mandatoryLookup(String name, Class<T> type) {
151        return CamelContextHelper.mandatoryLookup(getCamelContext(), name, type);
152    }
153
154    public void commit() {
155        // now lets turn all of the event driven consumer processors into a single route
156        if (!eventDrivenProcessors.isEmpty()) {
157            Processor target = Pipeline.newInstance(getCamelContext(), eventDrivenProcessors);
158
159            // force creating the route id so its known ahead of the route is started
160            String routeId = route.idOrCreate(getCamelContext().getNodeIdFactory());
161
162            // and wrap it in a unit of work so the UoW is on the top, so the entire route will be in the same UoW
163            CamelInternalProcessor internal = new CamelInternalProcessor(target);
164            internal.addAdvice(new CamelInternalProcessor.UnitOfWorkProcessorAdvice(this));
165
166            // and then optionally add route policy processor if a custom policy is set
167            List<RoutePolicy> routePolicyList = getRoutePolicyList();
168            if (routePolicyList != null && !routePolicyList.isEmpty()) {
169                for (RoutePolicy policy : routePolicyList) {
170                    // add policy as service if we have not already done that (eg possible if two routes have the same service)
171                    // this ensures Camel can control the lifecycle of the policy
172                    if (!camelContext.hasService(policy)) {
173                        try {
174                            camelContext.addService(policy);
175                        } catch (Exception e) {
176                            throw ObjectHelper.wrapRuntimeCamelException(e);
177                        }
178                    }
179                }
180
181                internal.addAdvice(new CamelInternalProcessor.RoutePolicyAdvice(routePolicyList));
182            }
183
184            // wrap in route inflight processor to track number of inflight exchanges for the route
185            internal.addAdvice(new CamelInternalProcessor.RouteInflightRepositoryAdvice(camelContext.getInflightRepository(), routeId));
186
187            // wrap in JMX instrumentation processor that is used for performance stats
188            internal.addAdvice(new CamelInternalProcessor.InstrumentationAdvice("route"));
189
190            // wrap in route lifecycle
191            internal.addAdvice(new CamelInternalProcessor.RouteLifecycleAdvice());
192
193            // and create the route that wraps the UoW
194            Route edcr = new EventDrivenConsumerRoute(this, getEndpoint(), internal);
195            edcr.getProperties().put(Route.ID_PROPERTY, routeId);
196            edcr.getProperties().put(Route.PARENT_PROPERTY, Integer.toHexString(route.hashCode()));
197            edcr.getProperties().put(Route.DESCRIPTION_PROPERTY, route.getDescriptionText());
198            if (route.getGroup() != null) {
199                edcr.getProperties().put(Route.GROUP_PROPERTY, route.getGroup());
200            }
201            String rest = "false";
202            if (route.isRest() != null && route.isRest()) {
203                rest = "true";
204            }
205            edcr.getProperties().put(Route.REST_PROPERTY, rest);
206
207            // after the route is created then set the route on the policy processor so we get hold of it
208            CamelInternalProcessor.RoutePolicyAdvice task = internal.getAdvice(CamelInternalProcessor.RoutePolicyAdvice.class);
209            if (task != null) {
210                task.setRoute(edcr);
211            }
212            CamelInternalProcessor.RouteLifecycleAdvice task2 = internal.getAdvice(CamelInternalProcessor.RouteLifecycleAdvice.class);
213            if (task2 != null) {
214                task2.setRoute(edcr);
215            }
216
217            // invoke init on route policy
218            if (routePolicyList != null && !routePolicyList.isEmpty()) {
219                for (RoutePolicy policy : routePolicyList) {
220                    policy.onInit(edcr);
221                }
222            }
223
224            routes.add(edcr);
225        }
226    }
227
228    public void addEventDrivenProcessor(Processor processor) {
229        eventDrivenProcessors.add(processor);
230    }
231
232    public List<InterceptStrategy> getInterceptStrategies() {
233        return interceptStrategies;
234    }
235
236    public void setInterceptStrategies(List<InterceptStrategy> interceptStrategies) {
237        this.interceptStrategies = interceptStrategies;
238    }
239
240    public void addInterceptStrategy(InterceptStrategy interceptStrategy) {
241        getInterceptStrategies().add(interceptStrategy);
242    }
243
244    public void setManagedInterceptStrategy(InterceptStrategy interceptStrategy) {
245        this.managedInterceptStrategy = interceptStrategy;
246    }
247
248    public InterceptStrategy getManagedInterceptStrategy() {
249        return managedInterceptStrategy;
250    }
251
252    public boolean isRouteAdded() {
253        return routeAdded;
254    }
255
256    public void setIsRouteAdded(boolean routeAdded) {
257        this.routeAdded = routeAdded;
258    }
259
260    public void setTracing(Boolean tracing) {
261        this.trace = tracing;
262    }
263
264    public Boolean isTracing() {
265        if (trace != null) {
266            return trace;
267        } else {
268            // fallback to the option from camel context
269            return getCamelContext().isTracing();
270        }
271    }
272
273    public void setMessageHistory(Boolean messageHistory) {
274        this.messageHistory = messageHistory;
275    }
276
277    public Boolean isMessageHistory() {
278        if (messageHistory != null) {
279            return messageHistory;
280        } else {
281            // fallback to the option from camel context
282            return getCamelContext().isMessageHistory();
283        }
284    }
285
286    public void setStreamCaching(Boolean cache) {
287        this.streamCache = cache;
288    }
289
290    public Boolean isStreamCaching() {
291        if (streamCache != null) {
292            return streamCache;
293        } else {
294            // fallback to the option from camel context
295            return getCamelContext().isStreamCaching();
296        }
297    }
298
299    public void setHandleFault(Boolean handleFault) {
300        this.handleFault = handleFault;
301    }
302
303    public Boolean isHandleFault() {
304        if (handleFault != null) {
305            return handleFault;
306        } else {
307            // fallback to the option from camel context
308            return getCamelContext().isHandleFault();
309        }
310    }
311
312    public void setDelayer(Long delay) {
313        this.delay = delay;
314    }
315
316    public Long getDelayer() {
317        if (delay != null) {
318            return delay;
319        } else {
320            // fallback to the option from camel context
321            return getCamelContext().getDelayer();
322        }
323    }
324
325    public void setAutoStartup(Boolean autoStartup) {
326        this.autoStartup = autoStartup;
327    }
328
329    public Boolean isAutoStartup() {
330        if (autoStartup != null) {
331            return autoStartup;
332        }
333        // default to true
334        return true;
335    }
336
337    public void setShutdownRoute(ShutdownRoute shutdownRoute) {
338        this.shutdownRoute = shutdownRoute;
339    }
340
341    public void setAllowUseOriginalMessage(Boolean allowUseOriginalMessage) {
342        throw new IllegalArgumentException("This option can only be configured on CamelContext");
343    }
344
345    public Boolean isAllowUseOriginalMessage() {
346        return getCamelContext().isAllowUseOriginalMessage();
347    }
348
349    public ShutdownRoute getShutdownRoute() {
350        if (shutdownRoute != null) {
351            return shutdownRoute;
352        } else {
353            // fallback to the option from camel context
354            return getCamelContext().getShutdownRoute();
355        }
356    }
357
358    public void setShutdownRunningTask(ShutdownRunningTask shutdownRunningTask) {
359        this.shutdownRunningTask = shutdownRunningTask;
360    }
361
362    public ShutdownRunningTask getShutdownRunningTask() {
363        if (shutdownRunningTask != null) {
364            return shutdownRunningTask;
365        } else {
366            // fallback to the option from camel context
367            return getCamelContext().getShutdownRunningTask();
368        }
369    }
370    
371    public int getAndIncrement(ProcessorDefinition<?> node) {
372        AtomicInteger count = nodeIndex.get(node);
373        if (count == null) {
374            count = new AtomicInteger();
375            nodeIndex.put(node, count);
376        }
377        return count.getAndIncrement();
378    }
379
380    public void setRoutePolicyList(List<RoutePolicy> routePolicyList) {
381        this.routePolicyList = routePolicyList;
382    }
383
384    public List<RoutePolicy> getRoutePolicyList() {
385        return routePolicyList;
386    }
387}