Java 类org.apache.camel.spi.RouteContext 实例源码

private static FlowBeginProcessor createFlowBeginProcessor(RouteContext routeContext) {
    CamelContext camelContext = routeContext.getCamelContext();

    // Try to obtain a FlowBeginProcessor bean (its definition is optional)
    FlowBeginProcessor processor = ContextUtils.beanOrNull(FlowBeginProcessor.class, camelContext);

    if (processor != null) {
        return processor;
    }

    // No FlowBeginProcessor bean found so let's create one. We need a
    // - reference to a ReplayStrategyRegistry
    // - reference to a FlowManager
    processor = new FlowBeginProcessor();
    processor.setCamelContext(camelContext);
    processor.setFlowManager(ContextUtils.bean(FlowManager.class, camelContext));
    processor.setRegistry(ContextUtils.bean(ReplayStrategyRegistry.class, camelContext));
    return processor;
}

@Override
@SuppressWarnings("unchecked")
public Processor createProcessor(RouteContext routeContext) throws Exception {
    // lookup in registry
    if (ObjectHelper.isNotEmpty(comparatorRef)) {
        comparator = routeContext.getCamelContext().getRegistry().lookupByNameAndType(comparatorRef, Comparator.class);
    }

    // if no comparator then default on to string representation
    if (comparator == null) {
        comparator = new Comparator<T>() {
            public int compare(T o1, T o2) {
                return ObjectHelper.compare(o1, o2);
            }
        };
    }

    // if no expression provided then default to body expression
    Expression exp;
    if (getExpression() == null) {
        exp = bodyExpression();
    } else {
        exp = getExpression().createExpression(routeContext);
    }
    return new SortProcessor<T>(exp, getComparator());
}

@Override
public Processor createProcessor(RouteContext routeContext) throws Exception {
    ObjectHelper.notEmpty(uri, "uri", this);

    Expression exp = createExpression(routeContext);

    SendDynamicProcessor processor = new SendDynamicProcessor(uri, exp);
    processor.setCamelContext(routeContext.getCamelContext());
    processor.setPattern(pattern);
    if (cacheSize != null) {
        processor.setCacheSize(cacheSize);
    }
    if (ignoreInvalidEndpoint != null) {
        processor.setIgnoreInvalidEndpoint(ignoreInvalidEndpoint);
    }
    return processor;
}

protected ServiceCallServerListStrategy configureServerListStrategy(C conf, RouteContext routeContext, ServiceCallConfigurationDefinition config)  throws Exception {
    ServiceCallServerListStrategy sl = config.getServerListStrategy();
    String ref = config.getServerListStrategyRef();
    if (sl == null && ref != null) {
        sl = builtInServerListStrategy(
                conf,
                ref)
            .orElseGet(() -> CamelContextHelper.mandatoryLookup(
                routeContext.getCamelContext(),
                ref,
                ServiceCallServerListStrategy.class)
        );
    }

    return sl;
}

protected DataFormat createDataFormat(RouteContext routeContext) {
    if (classType == null && clazz == null) {
        throw new IllegalArgumentException("Either packages or classType must be specified");
    }

    if (type == BindyType.Csv) {
        setDataFormatName("bindy-csv");
    } else if (type == BindyType.Fixed) {
        setDataFormatName("bindy-fixed");
    } else {
        setDataFormatName("bindy-kvp");
    }

    if (clazz == null && classType != null) {
        try {
            clazz = routeContext.getCamelContext().getClassResolver().resolveMandatoryClass(classType);
        } catch (ClassNotFoundException e) {
            throw ObjectHelper.wrapRuntimeCamelException(e);
        }
    }
    return super.createDataFormat(routeContext);
}

/**
 * Factory method to create the load balancer from the loadBalancerTypeName
 */
protected LoadBalancer createLoadBalancer(RouteContext routeContext) {
    ObjectHelper.notEmpty(loadBalancerTypeName, "loadBalancerTypeName", this);

    LoadBalancer answer = null;
    if (loadBalancerTypeName != null) {
        Class<?> type = routeContext.getCamelContext().getClassResolver().resolveClass(loadBalancerTypeName, LoadBalancer.class);
        if (type == null) {
            throw new IllegalArgumentException("Cannot find class: " + loadBalancerTypeName + " in the classpath");
        }
        answer = (LoadBalancer) routeContext.getCamelContext().getInjector().newInstance(type);
        configureLoadBalancer(answer);
    }

    return answer;
}

protected Processor createCompositeProcessor(RouteContext routeContext, List<Processor> list) throws Exception {
    final AggregationStrategy strategy = createAggregationStrategy(routeContext);

    boolean isParallelProcessing = getParallelProcessing() != null && getParallelProcessing();
    boolean isShareUnitOfWork = getShareUnitOfWork() != null && getShareUnitOfWork();
    boolean isStreaming = getStreaming() != null && getStreaming();
    boolean isStopOnException = getStopOnException() != null && getStopOnException();
    boolean isParallelAggregate = getParallelAggregate() != null && getParallelAggregate();

    boolean shutdownThreadPool = ProcessorDefinitionHelper.willCreateNewThreadPool(routeContext, this, isParallelProcessing);
    ExecutorService threadPool = ProcessorDefinitionHelper.getConfiguredExecutorService(routeContext, "Multicast", this, isParallelProcessing);

    long timeout = getTimeout() != null ? getTimeout() : 0;
    if (timeout > 0 && !isParallelProcessing) {
        throw new IllegalArgumentException("Timeout is used but ParallelProcessing has not been enabled.");
    }
    if (onPrepareRef != null) {
        onPrepare = CamelContextHelper.mandatoryLookup(routeContext.getCamelContext(), onPrepareRef, Processor.class);
    }

    MulticastProcessor answer = new MulticastProcessor(routeContext.getCamelContext(), list, strategy, isParallelProcessing,
                                  threadPool, shutdownThreadPool, isStreaming, isStopOnException, timeout, onPrepare, isShareUnitOfWork, isParallelAggregate);
    return answer;
}

protected ServiceCallServerListStrategy configureServerListStrategy(C conf, RouteContext routeContext, ServiceCallDefinition sd)  throws Exception {
    ServiceCallServerListStrategy sl = null;
    String ref;
    if (sd != null) {
        sl = sd.getServerListStrategy();
        ref = sd.getServerListStrategyRef();
        if (sl == null && ref != null) {
            sl = builtInServerListStrategy(
                    conf,
                    ref)
                .orElseGet(() -> CamelContextHelper.mandatoryLookup(
                    routeContext.getCamelContext(),
                    ref,
                    ServiceCallServerListStrategy.class)
            );
        }
    }

    return sl;
}

@Override
public Processor createProcessor(RouteContext routeContext) throws Exception {

    // if no timeout then we should block, and there use a negative timeout
    long time = timeout != null ? timeout : -1;
    boolean isIgnoreInvalidEndpoint = getIgnoreInvalidEndpoint() != null && getIgnoreInvalidEndpoint();
    Expression exp = getExpression().createExpression(routeContext);

    PollEnricher enricher = new PollEnricher(exp, time);

    AggregationStrategy strategy = createAggregationStrategy(routeContext);
    if (strategy == null) {
        enricher.setDefaultAggregationStrategy();
    } else {
        enricher.setAggregationStrategy(strategy);
    }
    if (getAggregateOnException() != null) {
        enricher.setAggregateOnException(getAggregateOnException());
    }
    if (getCacheSize() != null) {
        enricher.setCacheSize(getCacheSize());
    }
    enricher.setIgnoreInvalidEndpoint(isIgnoreInvalidEndpoint);

    return enricher;
}

protected ServiceCallLoadBalancer configureLoadBalancer(C conf, RouteContext routeContext, ServiceCallDefinition sd)  throws Exception {
    ServiceCallLoadBalancer lb = null;
    String ref;

    if (sd != null) {
        lb = sd.getLoadBalancer();
        ref = sd.getLoadBalancerRef();
        if (lb == null && ref != null) {
            lb = builtInLoadBalancer(
                    conf,
                    ref)
                .orElseGet(() -> CamelContextHelper.mandatoryLookup(
                    routeContext.getCamelContext(),
                    ref,
                    ServiceCallLoadBalancer.class)
            );
        }
    }

    return lb;
}

public void onErrorHandlerAdd(RouteContext routeContext, Processor errorHandler, ErrorHandlerFactory errorHandlerBuilder) {
    if (!shouldRegister(errorHandler, null)) {
        // avoid registering if not needed
        return;
    }

    Object me = getManagementObjectStrategy().getManagedObjectForErrorHandler(camelContext, routeContext, errorHandler, errorHandlerBuilder);

    // skip already managed services, for example if a route has been restarted
    if (getManagementStrategy().isManaged(me, null)) {
        LOG.trace("The error handler builder is already managed: {}", errorHandlerBuilder);
        return;
    }

    try {
        manageObject(me);
    } catch (Exception e) {
        LOG.warn("Could not register error handler builder: " + errorHandlerBuilder + " as ErrorHandler MBean.", e);
    }
}

/**
 * Will lookup and get the configured {@link java.util.concurrent.ExecutorService} from the given definition.
 * <p/>
 * This method will lookup for configured thread pool in the following order
 * <ul>
 * <li>from the definition if any explicit configured executor service.</li>
 * <li>from the {@link org.apache.camel.spi.Registry} if found</li>
 * <li>from the known list of {@link org.apache.camel.spi.ThreadPoolProfile ThreadPoolProfile(s)}.</li>
 * <li>if none found, then <tt>null</tt> is returned.</li>
 * </ul>
 * The various {@link ExecutorServiceAwareDefinition} should use this helper method to ensure they support
 * configured executor services in the same coherent way.
 *
 * @param routeContext the route context
 * @param name         name which is appended to the thread name, when the {@link java.util.concurrent.ExecutorService}
 *                     is created based on a {@link org.apache.camel.spi.ThreadPoolProfile}.
 * @param definition   the node definition which may leverage executor service.
 * @param useDefault   whether to fallback and use a default thread pool, if no explicit configured
 * @return the configured executor service, or <tt>null</tt> if none was configured.
 * @throws IllegalArgumentException is thrown if lookup of executor service in {@link org.apache.camel.spi.Registry} was not found
 */
public static ExecutorService getConfiguredExecutorService(RouteContext routeContext, String name,
                                                           ExecutorServiceAwareDefinition<?> definition,
                                                           boolean useDefault) throws IllegalArgumentException {
    ExecutorServiceManager manager = routeContext.getCamelContext().getExecutorServiceManager();
    ObjectHelper.notNull(manager, "ExecutorServiceManager", routeContext.getCamelContext());

    // prefer to use explicit configured executor on the definition
    if (definition.getExecutorService() != null) {
        return definition.getExecutorService();
    } else if (definition.getExecutorServiceRef() != null) {
        // lookup in registry first and use existing thread pool if exists
        ExecutorService answer = lookupExecutorServiceRef(routeContext, name, definition, definition.getExecutorServiceRef());
        if (answer == null) {
            throw new IllegalArgumentException("ExecutorServiceRef " + definition.getExecutorServiceRef() + " not found in registry or as a thread pool profile.");
        }
        return answer;
    } else if (useDefault) {
        return manager.newDefaultThreadPool(definition, name);
    }

    return null;
}

/**
 * Will lookup and get the configured {@link java.util.concurrent.ScheduledExecutorService} from the given definition.
 * <p/>
 * This method will lookup for configured thread pool in the following order
 * <ul>
 * <li>from the definition if any explicit configured executor service.</li>
 * <li>from the {@link org.apache.camel.spi.Registry} if found</li>
 * <li>from the known list of {@link org.apache.camel.spi.ThreadPoolProfile ThreadPoolProfile(s)}.</li>
 * <li>if none found, then <tt>null</tt> is returned.</li>
 * </ul>
 * The various {@link ExecutorServiceAwareDefinition} should use this helper method to ensure they support
 * configured executor services in the same coherent way.
 *
 * @param routeContext the rout context
 * @param name         name which is appended to the thread name, when the {@link java.util.concurrent.ExecutorService}
 *                     is created based on a {@link org.apache.camel.spi.ThreadPoolProfile}.
 * @param definition   the node definition which may leverage executor service.
 * @param useDefault   whether to fallback and use a default thread pool, if no explicit configured
 * @return the configured executor service, or <tt>null</tt> if none was configured.
 * @throws IllegalArgumentException is thrown if the found instance is not a ScheduledExecutorService type,
 *                                  or lookup of executor service in {@link org.apache.camel.spi.Registry} was not found
 */
public static ScheduledExecutorService getConfiguredScheduledExecutorService(RouteContext routeContext, String name,
                                                                             ExecutorServiceAwareDefinition<?> definition,
                                                                             boolean useDefault) throws IllegalArgumentException {
    ExecutorServiceManager manager = routeContext.getCamelContext().getExecutorServiceManager();
    ObjectHelper.notNull(manager, "ExecutorServiceManager", routeContext.getCamelContext());

    // prefer to use explicit configured executor on the definition
    if (definition.getExecutorService() != null) {
        ExecutorService executorService = definition.getExecutorService();
        if (executorService instanceof ScheduledExecutorService) {
            return (ScheduledExecutorService) executorService;
        }
        throw new IllegalArgumentException("ExecutorServiceRef " + definition.getExecutorServiceRef() + " is not an ScheduledExecutorService instance");
    } else if (definition.getExecutorServiceRef() != null) {
        ScheduledExecutorService answer = lookupScheduledExecutorServiceRef(routeContext, name, definition, definition.getExecutorServiceRef());
        if (answer == null) {
            throw new IllegalArgumentException("ExecutorServiceRef " + definition.getExecutorServiceRef() + " not found in registry or as a thread pool profile.");
        }
        return answer;
    } else if (useDefault) {
        return manager.newDefaultScheduledThreadPool(definition, name);
    }

    return null;
}

@Override
public Processor createProcessor(RouteContext routeContext) throws Exception {
    Policy policy = resolvePolicy(routeContext);
    ObjectHelper.notNull(policy, "policy", this);

    // before wrap
    policy.beforeWrap(routeContext, this);

    // create processor after the before wrap
    Processor childProcessor = this.createChildProcessor(routeContext, true);

    // wrap
    Processor target = policy.wrap(routeContext, childProcessor);

    if (!(target instanceof Service)) {
        // wrap the target so it becomes a service and we can manage its lifecycle
        target = new WrapProcessor(target, childProcessor);
    }
    return target;
}

public void addExceptionPolicy(RouteContext routeContext, OnExceptionDefinition exceptionType) {
    if (routeContext != null) {
        // add error handler as child service so they get lifecycle handled
        Processor errorHandler = exceptionType.getErrorHandler(routeContext.getRoute().getId());
        if (errorHandler != null) {
            addChildService(errorHandler);
        }
    }

    List<Class<? extends Throwable>> list = exceptionType.getExceptionClasses();
    for (Class<? extends Throwable> clazz : list) {
        String routeId = null;
        // only get the route id, if the exception type is route scoped
        if (exceptionType.isRouteScoped()) {
            RouteDefinition route = ProcessorDefinitionHelper.getRoute(exceptionType);
            if (route != null) {
                routeId = route.getId();
            }
        }
        ExceptionPolicyKey key = new ExceptionPolicyKey(routeId, clazz, exceptionType.getOnWhen());
        exceptionPolicies.put(key, exceptionType);
    }
}

@Override
public Processor createProcessor(RouteContext routeContext) throws Exception {
    Policy policy = resolvePolicy(routeContext);
    ObjectHelper.notNull(policy, "policy", this);

    // before wrap
    policy.beforeWrap(routeContext, this);

    // create processor after the before wrap
    Processor childProcessor = this.createChildProcessor(routeContext, true);

    // wrap
    Processor target = policy.wrap(routeContext, childProcessor);

    if (!(target instanceof Service)) {
        // wrap the target so it becomes a service and we can manage its lifecycle
        target = new WrapProcessor(target, childProcessor);
    }
    return target;
}

@Override
@SuppressWarnings("unchecked")
public Processor createProcessor(RouteContext routeContext) throws Exception {
    Processor childProcessor = this.createChildProcessor(routeContext, true);

    IdempotentRepository<String> idempotentRepository =
            (IdempotentRepository<String>) resolveMessageIdRepository(routeContext);
    ObjectHelper.notNull(idempotentRepository, "idempotentRepository", this);

    Expression expression = getExpression().createExpression(routeContext);

    // these boolean should be true by default
    boolean eager = getEager() == null || getEager();
    boolean duplicate = getSkipDuplicate() == null || getSkipDuplicate();
    boolean remove = getRemoveOnFailure() == null || getRemoveOnFailure();
    // these boolean should be false by default
    boolean completionEager = getCompletionEager() != null && getCompletionEager();

    return new IdempotentConsumer(expression, idempotentRepository, eager, completionEager, duplicate, remove, childProcessor);
}

private static FlowEndProcessor createFlowEndProcessor(RouteContext routeContext) {
    CamelContext camelContext = routeContext.getCamelContext();
    FlowEndProcessor processor = beanOrNull(FlowEndProcessor.class, camelContext);

    if (processor != null) {
        return processor;
    }

    processor = new FlowEndProcessor();
    processor.setCamelContext(camelContext);
    processor.setFlowManager(ContextUtils.bean(FlowManager.class, camelContext));
    return processor;
}

@Override
public Processor createProcessor(RouteContext routeContext) throws Exception {
    FlowProcessor processor = doCreateProcessor(routeContext);
    processor
        .inType(inType)
        .outType(outType)
        .inFormat(inFormat)
        .outFormat(outFormat)
        .outConversion(outConversion)
        .renderer(renderer(routeContext, processor.getMessageRenderer()))
        .setProcessor(createChildProcessor(routeContext, false));
    return processor;
}

private PlatformMessageRenderer renderer(RouteContext routeContext, PlatformMessageRenderer defaultRenderer) {
    if (messageRenderer != null) {
        return messageRenderer;
    } else if (messageRendererBeanName != null) {
        return routeContext.lookup(messageRendererBeanName, PlatformMessageRenderer.class);
    } else {
        return defaultRenderer;
    }
}

private static Dedupe createDedupe(RouteContext routeContext) {
    CamelContext camelContext = routeContext.getCamelContext();
    Dedupe dedupe = beanOrNull(Dedupe.class, camelContext);

    if (dedupe != null) {
        return dedupe;
    }

    dedupe = new Dedupe();
    dedupe.setFlowManager(ContextUtils.bean(FlowManager.class, camelContext));
    return dedupe;
}

@Override
public FlowProcessor doCreateProcessor(RouteContext routeContext) throws Exception {
    FlowBeginProcessor processor = createFlowBeginProcessor(routeContext);
    processor
        .identifier(identifier)
        .application(application)
        .expectedAckCount(expectedAckCount)
        .replayErrorHandler(replayErrorUri)
        .register();
    return processor;
}

private static FlowErrorProcessor createFlowErrorProcessor(RouteContext routeContext) {
    CamelContext camelContext = routeContext.getCamelContext();
    FlowErrorProcessor processor = beanOrNull(FlowErrorProcessor.class, camelContext);

    if (processor != null) {
        return processor;
    }

    processor = new FlowErrorProcessor();
    processor.setCamelContext(camelContext);
    processor.setFlowManager(ContextUtils.bean(FlowManager.class, camelContext));
    return processor;
}

@Override
public Processor createProcessor(RouteContext routeContext) {
    boolean isMarkRollbackOnly = getMarkRollbackOnly() != null && getMarkRollbackOnly();
    boolean isMarkRollbackOnlyLast = getMarkRollbackOnlyLast() != null && getMarkRollbackOnlyLast();

    // validate that only either mark rollbacks is chosen and not both
    if (isMarkRollbackOnly && isMarkRollbackOnlyLast) {
        throw new IllegalArgumentException("Only either one of markRollbackOnly and markRollbackOnlyLast is possible to select as true");
    }

    RollbackProcessor answer = new RollbackProcessor(message);
    answer.setMarkRollbackOnly(isMarkRollbackOnly);
    answer.setMarkRollbackOnlyLast(isMarkRollbackOnlyLast);
    return answer;
}

protected ThreadPoolRejectedPolicy resolveRejectedPolicy(RouteContext routeContext) {
    if (getExecutorServiceRef() != null && getRejectedPolicy() == null) {
        ThreadPoolProfile threadPoolProfile = routeContext.getCamelContext().getExecutorServiceManager().getThreadPoolProfile(getExecutorServiceRef());
        if (threadPoolProfile != null) {
            return threadPoolProfile.getRejectedPolicy();
        }
    }
    return getRejectedPolicy();
}

private ServiceCallServerListStrategy configureServerListStrategy(RouteContext routeContext, ServiceCallConfigurationDefinition config) {
    ServiceCallServerListStrategy lb = config.getServerListStrategy();
    if (lb == null && config.getServerListStrategyRef() != null) {
        String ref = config.getServerListStrategyRef();
        lb = CamelContextHelper.mandatoryLookup(routeContext.getCamelContext(), ref, ServiceCallServerListStrategy.class);
    }
    return lb;
}

@Override
protected DataFormat createDataFormat(RouteContext routeContext) {
    if ("json".equals(this.driver)) {
        setProperty(routeContext.getCamelContext(), this, "dataFormatName", "json-xstream");
    }
    DataFormat answer = super.createDataFormat(routeContext);
    // need to lookup the reference for the xstreamDriver
    if (ObjectHelper.isNotEmpty(driverRef)) {
        setProperty(routeContext.getCamelContext(), answer, "xstreamDriver", CamelContextHelper.mandatoryLookup(routeContext.getCamelContext(), driverRef));
    }
    return answer;
}

@Override
public void addErrorHandlers(RouteContext routeContext, OnExceptionDefinition exception) {
    ErrorHandlerBuilder handler = handlers.get(routeContext);
    if (handler != null) {
        handler.addErrorHandlers(routeContext, exception);
    }
    super.addErrorHandlers(routeContext, exception);
}

@Override
protected DataFormat createDataFormat(RouteContext routeContext) {
    if (unmarshallClass == null && unmarshallTypeName != null) {
        try {
            unmarshallClass = routeContext.getCamelContext().getClassResolver().resolveMandatoryClass(unmarshallTypeName);
        } catch (ClassNotFoundException e) {
            throw ObjectHelper.wrapRuntimeCamelException(e);
        }
    }

    return super.createDataFormat(routeContext);
}

public Processor createErrorHandler(RouteContext routeContext, Processor processor) throws Exception {
    DefaultErrorHandler answer = new DefaultErrorHandler(routeContext.getCamelContext(), processor, getLogger(), getOnRedelivery(), 
        getRedeliveryPolicy(), getExceptionPolicyStrategy(), getRetryWhilePolicy(routeContext.getCamelContext()),
            getExecutorService(routeContext.getCamelContext()), getOnPrepareFailure(), getOnExceptionOccurred());
    // configure error handler before we can use it
    configure(routeContext, answer);
    return answer;
}

@Override
protected DataFormat createDataFormat(RouteContext routeContext) {
    DataFormat flatpack = super.createDataFormat(routeContext);

    if (ObjectHelper.isNotEmpty(parserFactoryRef)) {
        Object parserFactory = CamelContextHelper.mandatoryLookup(routeContext.getCamelContext(), parserFactoryRef);
        setProperty(routeContext.getCamelContext(), flatpack, "parserFactory", parserFactory);
    }

    return flatpack;
}

@Override
public Processor createProcessor(RouteContext routeContext) {
    if (processor == null) {
        processor = new ExchangePatternProcessor(getPattern());
    }
    return processor;
}

@Override
public Processor createProcessor(RouteContext routeContext) throws Exception {
    Processor childProcessor = this.createChildProcessor(routeContext, true);
    aggregationStrategy = createAggregationStrategy(routeContext);

    boolean isParallelProcessing = getParallelProcessing() != null && getParallelProcessing();
    boolean isStreaming = getStreaming() != null && getStreaming();
    boolean isShareUnitOfWork = getShareUnitOfWork() != null && getShareUnitOfWork();
    boolean isParallelAggregate = getParallelAggregate() != null && getParallelAggregate();
    boolean shutdownThreadPool = ProcessorDefinitionHelper.willCreateNewThreadPool(routeContext, this, isParallelProcessing);
    ExecutorService threadPool = ProcessorDefinitionHelper.getConfiguredExecutorService(routeContext, "Split", this, isParallelProcessing);

    long timeout = getTimeout() != null ? getTimeout() : 0;
    if (timeout > 0 && !isParallelProcessing) {
        throw new IllegalArgumentException("Timeout is used but ParallelProcessing has not been enabled.");
    }
    if (onPrepareRef != null) {
        onPrepare = CamelContextHelper.mandatoryLookup(routeContext.getCamelContext(), onPrepareRef, Processor.class);
    }

    Expression exp = getExpression().createExpression(routeContext);

    Splitter answer = new Splitter(routeContext.getCamelContext(), exp, childProcessor, aggregationStrategy,
                        isParallelProcessing, threadPool, shutdownThreadPool, isStreaming, isStopOnException(),
                        timeout, onPrepare, isShareUnitOfWork, isParallelAggregate);
    return answer;
}

private ErrorHandlerBuilder createErrorHandler(RouteContext routeContext) {
    ErrorHandlerBuilder handler = (ErrorHandlerBuilder)lookupErrorHandlerBuilder(routeContext, getRef());
    ObjectHelper.notNull(handler, "error handler '" + ref + "'");

    // configure if the handler support transacted
    supportTransacted = handler.supportTransacted();

    List<OnExceptionDefinition> list = getErrorHandlers(routeContext);
    if (list != null) {
        for (OnExceptionDefinition exceptionType : list) {
            handler.addErrorHandlers(routeContext, exceptionType);
        }
    }
    return handler;
}

@Override
public Processor createProcessor(RouteContext routeContext) throws Exception {
    Processor answer = this.createChildProcessor(routeContext, true);

    // force the answer as a multicast processor even if there is only one child processor in the multicast
    if (!(answer instanceof MulticastProcessor)) {
        List<Processor> list = new ArrayList<Processor>(1);
        list.add(answer);
        answer = createCompositeProcessor(routeContext, list);
    }
    return answer;
}

public Processor createErrorHandler(RouteContext routeContext, Processor processor) throws Exception {
    ErrorHandlerBuilder handler = handlers.get(routeContext);
    if (handler == null) {
        handler = createErrorHandler(routeContext);
        handlers.put(routeContext, handler);
    }
    return handler.createErrorHandler(routeContext, processor);
}

public Processor wrap(final RouteContext routeContext, final Processor processor) {
    return new Processor() {
        public void process(Exchange exchange) throws Exception {
            invoked++;
            exchange.getIn().setHeader(name, "was wrapped");
            processor.process(exchange);
        }
    };
}

/**
 * Strategy to create the unit of work to be used for the sub route
 *
 * @param routeContext the route context
 * @param processor    the processor
 * @param exchange     the exchange
 * @return the unit of work processor
 */
protected Processor createUnitOfWorkProcessor(RouteContext routeContext, Processor processor, Exchange exchange) {
    CamelInternalProcessor internal = new CamelInternalProcessor(processor);

    // 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
    UnitOfWork parent = exchange.getProperty(Exchange.PARENT_UNIT_OF_WORK, UnitOfWork.class);
    if (parent != null) {
        internal.addAdvice(new CamelInternalProcessor.ChildUnitOfWorkProcessorAdvice(routeContext, parent));
    } else {
        internal.addAdvice(new CamelInternalProcessor.UnitOfWorkProcessorAdvice(routeContext));
    }

    return internal;
}

@Override
public Processor createProcessor(RouteContext routeContext) throws Exception {
    ObjectHelper.notNull(headerName, "headerName");
    Expression expr = getExpression().createExpression(routeContext);
    Expression nameExpr = ExpressionBuilder.parseSimpleOrFallbackToConstantExpression(getHeaderName(), routeContext.getCamelContext());
    return new SetHeaderProcessor(nameExpr, expr);
}

@Override
public Processor createProcessor(RouteContext routeContext) throws Exception {
    Processor tryProcessor = createOutputsProcessor(routeContext, getOutputsWithoutCatches());
    if (tryProcessor == null) {
        throw new IllegalArgumentException("Definition has no children on " + this);
    }

    List<Processor> catchProcessors = new ArrayList<Processor>();
    if (catchClauses != null) {
        for (CatchDefinition catchClause : catchClauses) {
            catchProcessors.add(createProcessor(routeContext, catchClause));
        }
    }

    FinallyDefinition finallyDefinition = finallyClause;
    if (finallyDefinition == null) {
        finallyDefinition = new FinallyDefinition();
        finallyDefinition.setParent(this);
    }
    Processor finallyProcessor = createProcessor(routeContext, finallyDefinition);

    // must have either a catch or finally
    if (finallyClause == null && catchClauses == null) {
        throw new IllegalArgumentException("doTry must have one or more catch or finally blocks on " + this);
    }

    return new TryProcessor(tryProcessor, catchProcessors, finallyProcessor);
}