import org.apache.camel.spi.RouteContext; //导入依赖的package包/类
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;
}
 

import org.apache.camel.spi.RouteContext; //导入依赖的package包/类
@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());
}
 

import org.apache.camel.spi.RouteContext; //导入依赖的package包/类
@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;
}
 

import org.apache.camel.spi.RouteContext; //导入依赖的package包/类
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;
}
 

import org.apache.camel.spi.RouteContext; //导入依赖的package包/类
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);
}
 

import org.apache.camel.spi.RouteContext; //导入依赖的package包/类
/**
 * 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;
}
 

import org.apache.camel.spi.RouteContext; //导入依赖的package包/类
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;
}
 

import org.apache.camel.spi.RouteContext; //导入依赖的package包/类
@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;
}
 

import org.apache.camel.spi.RouteContext; //导入依赖的package包/类
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;
}
 

import org.apache.camel.spi.RouteContext; //导入依赖的package包/类
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);
    }
}
 

import org.apache.camel.spi.RouteContext; //导入依赖的package包/类
/**
 * 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;
}
 

import org.apache.camel.spi.RouteContext; //导入依赖的package包/类
/**
 * 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;
}
 

import org.apache.camel.spi.RouteContext; //导入依赖的package包/类
@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;
}
 

import org.apache.camel.spi.RouteContext; //导入依赖的package包/类
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);
    }
}
 

import org.apache.camel.spi.RouteContext; //导入依赖的package包/类
@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);
}
 

import org.apache.camel.spi.RouteContext; //导入依赖的package包/类
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;
}
 

import org.apache.camel.spi.RouteContext; //导入依赖的package包/类
@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;
}
 

import org.apache.camel.spi.RouteContext; //导入依赖的package包/类
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;
    }
}
 

import org.apache.camel.spi.RouteContext; //导入依赖的package包/类
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;
}
 

import org.apache.camel.spi.RouteContext; //导入依赖的package包/类
@Override
public FlowProcessor doCreateProcessor(RouteContext routeContext) throws Exception {
    FlowBeginProcessor processor = createFlowBeginProcessor(routeContext);
    processor
        .identifier(identifier)
        .application(application)
        .expectedAckCount(expectedAckCount)
        .replayErrorHandler(replayErrorUri)
        .register();
    return processor;
}