import org.apache.camel.util.ServiceHelper; //导入依赖的package包/类
@Before
public void setUp() throws Exception {
    camelContext = new DefaultCamelContext();
    SimpleRegistry registry = new SimpleRegistry();
    Map<String, Object> params = new HashMap<String, Object>();
    params.put("custName", "Willem");
    // bind the params
    registry.put("params", params);
    camelContext.setRegistry(registry);
    
    template = camelContext.createProducerTemplate();
    ServiceHelper.startServices(template, camelContext);

    Endpoint value = camelContext.getEndpoint(getEndpointUri());
    assertNotNull("Could not find endpoint!", value);
    assertTrue("Should be a JPA endpoint but was: " + value, value instanceof JpaEndpoint);
    endpoint = (JpaEndpoint)value;

    transactionTemplate = endpoint.createTransactionTemplate();
    entityManager = endpoint.createEntityManager();
}
 

import org.apache.camel.util.ServiceHelper; //导入依赖的package包/类
@Override
public PollingConsumer acquirePollingConsumer(Endpoint endpoint) {
    // always create a new consumer
    PollingConsumer answer;
    try {
        answer = endpoint.createPollingConsumer();
        boolean singleton = true;
        if (answer instanceof IsSingleton) {
            singleton = ((IsSingleton) answer).isSingleton();
        }
        if (getCamelContext().isStartingRoutes() && singleton) {
            // if we are currently starting a route, then add as service and enlist in JMX
            // - but do not enlist non-singletons in JMX
            // - note addService will also start the service
            getCamelContext().addService(answer);
        } else {
            // must then start service so producer is ready to be used
            ServiceHelper.startService(answer);
        }
    } catch (Exception e) {
        throw new FailedToCreateConsumerException(endpoint, e);
    }
    return answer;
}
 

import org.apache.camel.util.ServiceHelper; //导入依赖的package包/类
public void testMultithreaded() throws Exception {
    int numMessages = 100;

    Object[] bodies = new Object[numMessages];
    for (int i = 0; i < numMessages; i++) {
        bodies[i] = "msg" + i;
    }

    getMockEndpoint("mock:result").expectedBodiesReceived(bodies);
    getMockEndpoint("mock:result").setResultWaitTime(20000);

    ProducerTemplate producerTemplate = context.createProducerTemplate();
    ProducerTemplate producerTemplate2 = context.createProducerTemplate();

    ExecutorService service = context.getExecutorServiceManager().newFixedThreadPool(this, getName(), 2);

    service.execute(new Sender(producerTemplate, 0, numMessages, 2));
    service.execute(new Sender(producerTemplate2, 1, numMessages, 2));

    assertMockEndpointsSatisfied();

    ServiceHelper.stopServices(producerTemplate, producerTemplate2);
}
 

import org.apache.camel.util.ServiceHelper; //导入依赖的package包/类
@Override
protected void doStop() throws Exception {
    // note: we cannot do doForceCompletionOnStop from this doStop method
    // as this is handled in the prepareShutdown method which is also invoked when stopping a route
    // and is better suited for preparing to shutdown than this doStop method is

    if (aggregateController != null) {
        aggregateController.onStop(this);
    }

    if (recoverService != null) {
        camelContext.getExecutorServiceManager().shutdown(recoverService);
    }
    ServiceHelper.stopServices(timeoutMap, processor, deadLetterProducerTemplate);

    if (closedCorrelationKeys != null) {
        // it may be a service so stop it as well
        ServiceHelper.stopService(closedCorrelationKeys);
        closedCorrelationKeys.clear();
    }
    batchConsumerCorrelationKeys.clear();
    redeliveryState.clear();
}
 

import org.apache.camel.util.ServiceHelper; //导入依赖的package包/类
@Override
protected void doShutdown() throws Exception {
    // shutdown aggregation repository and the strategy
    ServiceHelper.stopAndShutdownServices(aggregationRepository, aggregationStrategy);

    // cleanup when shutting down
    inProgressCompleteExchanges.clear();

    if (shutdownExecutorService) {
        camelContext.getExecutorServiceManager().shutdownNow(executorService);
    }
    if (shutdownTimeoutCheckerExecutorService) {
        camelContext.getExecutorServiceManager().shutdownNow(timeoutCheckerExecutorService);
        timeoutCheckerExecutorService = null;
    }

    super.doShutdown();
}
 

import org.apache.camel.util.ServiceHelper; //导入依赖的package包/类
protected void doStop() throws Exception {
    // when stopping we intend to shutdown
    ServiceHelper.stopAndShutdownService(statistics);
    if (stopServicePool) {
        ServiceHelper.stopAndShutdownService(pool);
    }
    try {
        ServiceHelper.stopAndShutdownServices(producers.values());
    } finally {
        // ensure producers are removed, and also from JMX
        for (Producer producer : producers.values()) {
            getCamelContext().removeService(producer);
        }
    }
    producers.clear();
    if (statistics != null) {
        statistics.clear();
    }
}
 

import org.apache.camel.util.ServiceHelper; //导入依赖的package包/类
public void activate() {
    LOG.debug("CamelDestination activate().... ");
    ObjectHelper.notNull(camelContext, "CamelContext", this);
    try {
        LOG.debug("establishing Camel connection");
        destinationEndpoint = getCamelContext().getEndpoint(camelDestinationUri);
        if (destinationEndpoint == null) {
            throw new NoSuchEndpointException(camelDestinationUri);
        }
        consumer = destinationEndpoint.createConsumer(new ConsumerProcessor());
        ServiceHelper.startService(consumer);
    } catch (NoSuchEndpointException nex) {
        throw nex;
    } catch (Exception ex) {
        if (destinationEndpoint == null) {
            throw new FailedToCreateConsumerException(camelDestinationUri, ex);
        }
        throw new FailedToCreateConsumerException(destinationEndpoint, ex);
    }
}
 

import org.apache.camel.util.ServiceHelper; //导入依赖的package包/类
/**
 * Gather all child services
 */
private Set<Service> gatherChildServices(Route route, boolean includeErrorHandler) {
    // gather list of services to stop as we need to start child services as well
    List<Service> services = new ArrayList<Service>();
    services.addAll(route.getServices());
    // also get route scoped services
    doGetRouteScopedServices(services, route);
    Set<Service> list = new LinkedHashSet<Service>();
    for (Service service : services) {
        list.addAll(ServiceHelper.getChildServices(service));
    }
    if (includeErrorHandler) {
        // also get route scoped error handler (which must be done last)
        doGetRouteScopedErrorHandler(list, route);
    }
    Set<Service> answer = new LinkedHashSet<Service>();
    answer.addAll(list);
    return answer;
}
 

import org.apache.camel.util.ServiceHelper; //导入依赖的package包/类
@Override
protected void doStart() throws Exception {
    ObjectHelper.notNull(executorService, "executorService", this);
    ObjectHelper.notNull(scheduledExecutorService, "scheduledExecutorService", this);
    ObjectHelper.notNull(endpoint, "endpoint", this);

    // timeout map to use for purging messages which have timed out, while waiting for an expected reply
    // when doing request/reply over JMS
    log.trace("Using timeout checker interval with {} millis", endpoint.getRequestTimeoutCheckerInterval());
    correlation = new CorrelationTimeoutMap(scheduledExecutorService, endpoint.getRequestTimeoutCheckerInterval(), executorService);
    ServiceHelper.startService(correlation);

    // create JMS listener and start it
    listenerContainer = createListenerContainer();
    listenerContainer.afterPropertiesSet();
    log.debug("Starting reply listener container on endpoint: {}", endpoint);

    endpoint.onListenerContainerStarting(listenerContainer);
    listenerContainer.start();
}
 

import org.apache.camel.util.ServiceHelper; //导入依赖的package包/类
@Override
protected void doStop() throws Exception {
    // stop event notifier
    camelContext.getManagementStrategy().removeEventNotifier(eventNotifier);
    ServiceHelper.stopService(eventNotifier);

    // stop and close collector
    ServiceHelper.stopAndShutdownService(spanCollector);
    if (spanCollector instanceof Closeable) {
        IOHelper.close((Closeable) spanCollector);
    }
    // clear braves
    braves.clear();
    // remove route policy
    camelContext.getRoutePolicyFactories().remove(this);
}
 

import org.apache.camel.util.ServiceHelper; //导入依赖的package包/类
protected void doStart() throws Exception {
    if (producerCache == null) {
        if (maximumCacheSize > 0) {
            producerCache = new ProducerCache(this, camelContext, maximumCacheSize);
        } else {
            producerCache = new ProducerCache(this, camelContext);
        }
        producerCache.setEventNotifierEnabled(isEventNotifierEnabled());
    }

    // need to lookup default endpoint as it may have been intercepted
    if (defaultEndpoint != null) {
        defaultEndpoint = camelContext.getEndpoint(defaultEndpoint.getEndpointUri());
    }

    ServiceHelper.startService(producerCache);
}
 

import org.apache.camel.util.ServiceHelper; //导入依赖的package包/类
protected void doStart() throws Exception {
    if (aggregationStrategy == null) {
        aggregationStrategy = defaultAggregationStrategy();
    }

    if (producerCache == null) {
        if (cacheSize < 0) {
            producerCache = new EmptyProducerCache(this, camelContext);
            LOG.debug("Enricher {} is not using ProducerCache", this);
        } else if (cacheSize == 0) {
            producerCache = new ProducerCache(this, camelContext);
            LOG.debug("Enricher {} using ProducerCache with default cache size", this);
        } else {
            producerCache = new ProducerCache(this, camelContext, cacheSize);
            LOG.debug("Enricher {} using ProducerCache with cacheSize={}", this, cacheSize);
        }
    }

    ServiceHelper.startServices(producerCache, aggregationStrategy);
}
 

import org.apache.camel.util.ServiceHelper; //导入依赖的package包/类
@Override
@Before
protected void setUp() throws Exception {
    super.setUp();

    context2 = new DefaultCamelContext();
    template2 = context2.createProducerTemplate();

    ServiceHelper.startServices(template2, context2);

    // add routes after CamelContext has been started
    RouteBuilder routeBuilder = createRouteBuilderForSecondContext();
    if (routeBuilder != null) {
        context2.addRoutes(routeBuilder);
    }
}
 

import org.apache.camel.util.ServiceHelper; //导入依赖的package包/类
@Override
public void unsubscribe() {
    if (unsubscribed.compareAndSet(false, true)) {
        if (consumer != null) {
            // must stop the consumer from the worker pool as we should not stop ourself from a thread from ourself
            workerPool.submit(new Runnable() {
                @Override
                public void run() {
                    try {
                        ServiceHelper.stopServices(consumer);
                    } catch (Exception e) {
                        LOG.warn("Error stopping consumer: " + consumer + " due " + e.getMessage() + ". This exception is ignored.", e);
                    }
                }
            });
        }
    }
}
 

import org.apache.camel.util.ServiceHelper; //导入依赖的package包/类
protected void doStart() throws Exception {
    if (isParallelProcessing() && executorService == null) {
        throw new IllegalArgumentException("ParallelProcessing is enabled but ExecutorService has not been set");
    }
    if (timeout > 0 && !isParallelProcessing()) {
        throw new IllegalArgumentException("Timeout is used but ParallelProcessing has not been enabled");
    }
    if (isParallelProcessing() && aggregateExecutorService == null) {
        // use unbounded thread pool so we ensure the aggregate on-the-fly task always will have assigned a thread
        // and run the tasks when the task is submitted. If not then the aggregate task may not be able to run
        // and signal completion during processing, which would lead to what would appear as a dead-lock or a slow processing
        String name = getClass().getSimpleName() + "-AggregateTask";
        aggregateExecutorService = createAggregateExecutorService(name);
    }
    ServiceHelper.startServices(aggregationStrategy, processors);
}
 

import org.apache.camel.util.ServiceHelper; //导入依赖的package包/类
public void deferStartService(Object object, boolean stopOnShutdown) throws Exception {
    if (object instanceof Service) {
        Service service = (Service) object;

        // only add to services to close if its a singleton
        // otherwise we could for example end up with a lot of prototype scope endpoints
        boolean singleton = true; // assume singleton by default
        if (object instanceof IsSingleton) {
            singleton = ((IsSingleton) service).isSingleton();
        }
        // do not add endpoints as they have their own list
        if (singleton && !(service instanceof Endpoint)) {
            // only add to list of services to stop if its not already there
            if (stopOnShutdown && !hasService(service)) {
                servicesToStop.add(service);
            }
        }
        // are we already started?
        if (isStarted()) {
            ServiceHelper.startService(service);
        } else {
            deferStartupListener.addService(service);
        }
    }
}
 

import org.apache.camel.util.ServiceHelper; //导入依赖的package包/类
/**
 * Releases an acquired producer back after usage.
 *
 * @param endpoint the endpoint
 * @param pollingConsumer the pollingConsumer to release
 */
public void releasePollingConsumer(Endpoint endpoint, PollingConsumer pollingConsumer) {
    if (pollingConsumer instanceof ServicePoolAware) {
        // release back to the pool
        pool.release(endpoint, pollingConsumer);
    } else {
        boolean singleton = false;
        if (pollingConsumer instanceof IsSingleton) {
            singleton = ((IsSingleton) pollingConsumer).isSingleton();
        }
        if (!singleton) {
            try {
                // stop and shutdown non-singleton producers as we should not leak resources
                ServiceHelper.stopAndShutdownService(pollingConsumer);
            } catch (Exception ex) {
                if (ex instanceof RuntimeCamelException) {
                    throw (RuntimeCamelException)ex;
                } else {
                    throw new RuntimeCamelException(ex);
                }
            }
        }
    }
}
 

import org.apache.camel.util.ServiceHelper; //导入依赖的package包/类
protected void doStop() throws Exception {
    // when stopping we intend to shutdown
    ServiceHelper.stopAndShutdownServices(statistics, pool);
    try {
        ServiceHelper.stopAndShutdownServices(consumers.values());
    } finally {
        // ensure consumers are removed, and also from JMX
        for (PollingConsumer consumer : consumers.values()) {
            getCamelContext().removeService(consumer);
        }
    }
    consumers.clear();
    if (statistics != null) {
        statistics.clear();
    }
}
 

import org.apache.camel.util.ServiceHelper; //导入依赖的package包/类
/**
 * Prepares the services for shutdown, by invoking the {@link ShutdownPrepared#prepareShutdown(boolean, boolean)} method
 * on the service if it implement this interface.
 *
 * @param service the service
 * @param forced  whether to force shutdown
 * @param includeChildren whether to prepare the child of the service as well
 */
private static void prepareShutdown(Service service, boolean suspendOnly, boolean forced, boolean includeChildren, boolean suppressLogging) {
    Set<Service> list;
    if (includeChildren) {
        // include error handlers as we want to prepare them for shutdown as well
        list = ServiceHelper.getChildServices(service, true);
    } else {
        list = new LinkedHashSet<Service>(1);
        list.add(service);
    }

    for (Service child : list) {
        if (child instanceof ShutdownPrepared) {
            try {
                LOG.trace("Preparing {} shutdown on {}", forced ? "forced" : "", child);
                ((ShutdownPrepared) child).prepareShutdown(suspendOnly, forced);
            } catch (Exception e) {
                if (suppressLogging) {
                    LOG.trace("Error during prepare shutdown on " + child + ". This exception will be ignored.", e);
                } else {
                    LOG.warn("Error during prepare shutdown on " + child + ". This exception will be ignored.", e);
                }
            }
        }
    }
}
 

import org.apache.camel.util.ServiceHelper; //导入依赖的package包/类
@Override
protected void doStart() throws Exception {
    if (dataFormat == null && name != null) {
        dataFormat = getCamelContext().resolveDataFormat(name);
    }
    if (operation.equals("marshal")) {
        marshal = new MarshalProcessor(dataFormat);
        marshal.setCamelContext(getCamelContext());
    } else {
        unmarshal = new UnmarshalProcessor(dataFormat);
        unmarshal.setCamelContext(getCamelContext());
    }

    ServiceHelper.startServices(dataFormat, marshal, unmarshal);
    super.doStart();
}