Java 类org.apache.camel.util.ServiceHelper 实例源码

@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;
}

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);
}

@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();
}

@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();
}

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();
    }
}

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);
    }
}

/**
 * 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;
}

@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();
}

@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);
}

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);
}

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);
}

@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);
    }
}

@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);
                    }
                }
            });
        }
    }
}

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);
}

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);
        }
    }
}

/**
 * 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);
                }
            }
        }
    }
}

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();
    }
}

/**
 * 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);
                }
            }
        }
    }
}

@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();
}

@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();
}

@Override
protected void doStart() throws Exception {
    ObjectHelper.notEmpty(getName(), "name", "serviceName");
    ObjectHelper.notNull(camelContext, "camelContext");
    ObjectHelper.notNull(serviceCallExpression, "serviceCallExpression");

    LOG.info("ServiceCall with service name: {} is using load balancer: {} and service discovery: {}",
        name, loadBalancer, serverListStrategy);

    processor = new SendDynamicProcessor(uri, serviceCallExpression);
    processor.setCamelContext(getCamelContext());
    if (exchangePattern != null) {
        processor.setPattern(exchangePattern);
    }

    ServiceHelper.startServices(serverListStrategy, processor);
}

@Override
protected void doStart() throws Exception {
    jsonMarshalEndpoint = getCamelContext().getEndpoint("dataformat:json-jackson:marshal");
    Objects.requireNonNull(jsonMarshalEndpoint, "jsonMarshalEndpoint");
    jsonMarshalProducer = jsonMarshalEndpoint.createProducer();
    Objects.requireNonNull(jsonMarshalProducer, "jsonMarshalProducer");
    ServiceHelper.startServices(jsonMarshalEndpoint, jsonMarshalProducer);
    super.doStart();
}

@Override
protected void doStart() throws Exception {
    jsonMarshalEndpoint = getCamelContext().getEndpoint("dataformat:json-jackson:marshal");
    Objects.requireNonNull(jsonMarshalEndpoint, "jsonMarshalEndpoint");
    jsonMarshalProducer = jsonMarshalEndpoint.createProducer();
    Objects.requireNonNull(jsonMarshalProducer, "jsonMarshalProducer");
    ServiceHelper.startServices(jsonMarshalEndpoint, jsonMarshalProducer);
    super.doStart();
}

/**
 * Releases an acquired producer back after usage.
 *
 * @param endpoint the endpoint
 * @param producer the producer to release
 * @throws Exception can be thrown if error stopping producer if that was needed.
 */
public void releaseProducer(Endpoint endpoint, Producer producer) throws Exception {
    if (producer instanceof ServicePoolAware) {
        // release back to the pool
        pool.release(endpoint, producer);
    } else if (!producer.isSingleton()) {
        // stop and shutdown non-singleton producers as we should not leak resources
        ServiceHelper.stopAndShutdownService(producer);
    }
}

@Override
protected void doStart() throws Exception {
    super.doStart();

    LOG.debug("Netty consumer binding to: {}", configuration.getAddress());

    if (nettyServerBootstrapFactory == null) {
        // setup pipeline factory
        ServerPipelineFactory pipelineFactory;
        ServerPipelineFactory factory = configuration.getServerPipelineFactory();
        if (factory != null) {
            pipelineFactory = factory.createPipelineFactory(this);
        } else {
            pipelineFactory = new DefaultServerPipelineFactory(this);
        }

        if (isTcp()) {
            if (configuration.isClientMode()) {
                nettyServerBootstrapFactory = new ClientModeTCPNettyServerBootstrapFactory();
            } else {
                nettyServerBootstrapFactory = new SingleTCPNettyServerBootstrapFactory();
            }
        } else {
            nettyServerBootstrapFactory = new SingleUDPNettyServerBootstrapFactory();
        }
        nettyServerBootstrapFactory.init(context, configuration, pipelineFactory);
    }

    ServiceHelper.startServices(nettyServerBootstrapFactory);

    LOG.info("Netty consumer bound to: " + configuration.getAddress());
}

public SubscriptionHelper getSubscriptionHelper() throws Exception {
    if (subscriptionHelper == null) {
        // lazily create subscription helper
        subscriptionHelper = new SubscriptionHelper(this);

        // also start the helper to connect to Salesforce
        ServiceHelper.startService(subscriptionHelper);
    }
    return subscriptionHelper;
}

protected void unInitReplyManager() {
    try {
        if (replyManager != null) {
            if (LOG.isDebugEnabled()) {
                LOG.debug("Stopping JmsReplyManager: {} from processing replies from: {}", replyManager,
                        endpoint.getReplyTo() != null ? endpoint.getReplyTo() : "temporary queue");
            }
            ServiceHelper.stopService(replyManager);
        }
    } catch (Exception e) {
        throw ObjectHelper.wrapRuntimeCamelException(e);
    } finally {
        started.set(false);
    }
}

/**
 * Creates a new builder.
 *
 * @param context the Camel context
 */
public NotifyBuilder(CamelContext context) {
    this.context = context;
    eventNotifier = new ExchangeNotifier();
    try {
        ServiceHelper.startService(eventNotifier);
    } catch (Exception e) {
        throw ObjectHelper.wrapRuntimeCamelException(e);
    }
    context.getManagementStrategy().addEventNotifier(eventNotifier);
}

@Override
protected void doStop() throws Exception {
    if (beanStarted) {
        try {
            // Stop the bean if it implements Service interface and if cached
            // so meant to be reused
            ServiceHelper.stopService(processor.getBean());
            beanStarted = false;
        } catch (NoSuchBeanException e) {
        }
    }

    super.doStop();
}

protected void doStart() throws Exception {
    ObjectHelper.notNull(configuration, "setNettyServerBootstrapConfiguration() must be called with a NettyServerBootstrapConfiguration instance", this);

    // port must be set
    if (configuration.getPort() <= 0) {
        throw new IllegalArgumentException("Port must be configured on NettySharedHttpServerBootstrapConfiguration " + configuration);
    }
    // hostname must be set
    if (ObjectHelper.isEmpty(configuration.getHost())) {
        throw new IllegalArgumentException("Host must be configured on NettySharedHttpServerBootstrapConfiguration " + configuration);
    }

    LOG.debug("NettySharedHttpServer using configuration: {}", configuration);

    // force using tcp as the underlying transport
    configuration.setProtocol("tcp");

    channelFactory = new HttpServerMultiplexChannelHandler();
    channelFactory.init(configuration.getPort());

    ChannelPipelineFactory pipelineFactory = new HttpServerSharedPipelineFactory(configuration, channelFactory, classResolver);

    // thread factory and pattern
    String port = Matcher.quoteReplacement("" + configuration.getPort());
    String pattern = threadPattern;
    pattern = pattern.replaceFirst("#port#", port);
    ThreadFactory tf = new CamelThreadFactory(pattern, "NettySharedHttpServer", true);

    // create bootstrap factory and disable compatible check as its shared among the consumers
    bootstrapFactory = new HttpServerBootstrapFactory(channelFactory, false);
    bootstrapFactory.init(tf, configuration, pipelineFactory);

    ServiceHelper.startServices(channelFactory);

    if (startServer) {
        LOG.info("Starting NettySharedHttpServer on {}:{}", configuration.getHost(), configuration.getPort());
        ServiceHelper.startServices(bootstrapFactory);
    }
}

@Test
public void testScheduledResumeRoutePolicy() throws Exception {
    MockEndpoint success = context.getEndpoint("mock:success", MockEndpoint.class);
    success.expectedMessageCount(1);

    context.getComponent("quartz2", QuartzComponent.class).setPropertiesFile("org/apache/camel/routepolicy/quartz2/myquartz.properties");
    context.addRoutes(new RouteBuilder() {
        public void configure() {
            SimpleScheduledRoutePolicy policy = new SimpleScheduledRoutePolicy();
            long startTime = System.currentTimeMillis() + 3000L;
            policy.setRouteResumeDate(new Date(startTime));
            policy.setRouteResumeRepeatCount(1);
            policy.setRouteResumeRepeatInterval(3000);

            from("direct:start")
                .routeId("test")
                .routePolicy(policy)
                .to("mock:success");
        } 
    });
    context.start();

    ServiceHelper.suspendService(context.getRoute("test").getConsumer());
    try {
        template.sendBody("direct:start", "Ready or not, Here, I come");
        fail("Should have thrown an exception");
    } catch (CamelExecutionException e) {
        LOG.debug("Consumer successfully suspended");
    } 

    Thread.sleep(4000);
    template.sendBody("direct:start", "Ready or not, Here, I come");

    context.getComponent("quartz2", QuartzComponent.class).stop();
    success.assertIsSatisfied();
}

protected ReplyManager createReplyManager() throws Exception {
    // use a temporary queue
    ReplyManager replyManager = new TemporaryQueueReplyManager(getEndpoint().getCamelContext());
    replyManager.setEndpoint(getEndpoint());

    String name = "RabbitMQReplyManagerTimeoutChecker[" + getEndpoint().getExchangeName() + "]";
    ScheduledExecutorService replyManagerExecutorService = getEndpoint().getCamelContext().getExecutorServiceManager().newSingleThreadScheduledExecutor(name, name);
    replyManager.setScheduledExecutorService(replyManagerExecutorService);
    log.info("Starting reply manager service " + name);
    ServiceHelper.startService(replyManager);

    return replyManager;
}

protected void doStart() throws Exception {
    // optimize to only get (create) a processor if really needed
    if (beanHolder.supportProcessor() && allowProcessor(method, beanHolder.getBeanInfo())) {
        processor = beanHolder.getProcessor();
        ServiceHelper.startService(processor);
    } else if (beanHolder instanceof ConstantBeanHolder) {
        try {
            // Start the bean if it implements Service interface and if cached
            // so meant to be reused
            ServiceHelper.startService(beanHolder.getBean());
        } catch (NoSuchBeanException e) {
            // ignore
        }
    }
}

@Before
public void setUp() throws Exception {
    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();
}

@Override
protected void doStart() throws Exception {
    super.doStart();

    final SalesforceEndpointConfig config = endpoint.getConfiguration();

    // is a query configured in the endpoint?
    if (config.getSObjectQuery() != null) {
        // Note that we don't lookup topic if the query is not specified
        // create REST client for PushTopic operations
        SalesforceComponent component = endpoint.getComponent();
        RestClient restClient = new DefaultRestClient(component.getConfig().getHttpClient(),
                endpoint.getConfiguration().getApiVersion(), endpoint.getConfiguration().getFormat(), component.getSession());
        // don't forget to start the client
        ServiceHelper.startService(restClient);

        try {
            PushTopicHelper helper = new PushTopicHelper(config, topicName, restClient);
            helper.createOrUpdateTopic();
        } finally {
            // don't forget to stop the client
            ServiceHelper.stopService(restClient);
        }
    }

    // subscribe to topic
    subscriptionHelper.subscribe(topicName, this);
    subscribed = true;
}

public void setUp() throws Exception {
    super.setUp();
    context.setUseMDCLogging(true);

    context2 = new DefaultCamelContext();
    context2.setUseMDCLogging(true);
    template2 = context2.createProducerTemplate();
    ServiceHelper.startServices(template2, context2);

    // add routes after CamelContext has been started
    context2.addRoutes(createRouteBuilder2());
}

@Override
public void releasePollingConsumer(Endpoint endpoint, PollingConsumer pollingConsumer) {
    // stop and shutdown the consumer as its not cache or reused
    try {
        ServiceHelper.stopAndShutdownService(pollingConsumer);
    } catch (Exception e) {
        throw ObjectHelper.wrapRuntimeCamelException(e);
    }
}

@Test
public void testScheduledResumeRoutePolicy() throws Exception {
    MockEndpoint success = context.getEndpoint("mock:success", MockEndpoint.class);
    success.expectedMessageCount(1);

    context.getComponent("quartz2", QuartzComponent.class).setPropertiesFile("org/apache/camel/routepolicy/quartz2/myquartz.properties");
    context.addRoutes(new RouteBuilder() {
        public void configure() {
            CronScheduledRoutePolicy policy = new CronScheduledRoutePolicy();
            policy.setRouteResumeTime("*/3 * * * * ?");

            from("direct:start")
                .routeId("test")
                .routePolicy(policy)
                .to("mock:success");
        } 
    });
    context.start();

    ServiceHelper.suspendService(context.getRoute("test").getConsumer());

    Thread.sleep(5000);
    assertTrue(context.getRouteStatus("test") == ServiceStatus.Started);

    template.sendBody("direct:start", "Ready or not, Here, I come");

    success.assertIsSatisfied();
}

protected void doStart() throws Exception {
    if (producerCache == null) {
        if (cacheSize < 0) {
            producerCache = new EmptyProducerCache(this, camelContext);
            LOG.debug("DynamicSendTo {} is not using ProducerCache", this);
        } else if (cacheSize == 0) {
            producerCache = new ProducerCache(this, camelContext);
            LOG.debug("DynamicSendTo {} using ProducerCache with default cache size", this);
        } else {
            producerCache = new ProducerCache(this, camelContext, cacheSize);
            LOG.debug("DynamicSendTo {} using ProducerCache with cacheSize={}", this, cacheSize);
        }
    }
    ServiceHelper.startService(producerCache);
}

protected void startChildService(Route route, List<Service> services) throws Exception {
    for (Service service : services) {
        LOG.debug("Starting child service on route: {} -> {}", route.getId(), service);
        for (LifecycleStrategy strategy : camelContext.getLifecycleStrategies()) {
            strategy.onServiceAdd(camelContext, service, route);
        }
        ServiceHelper.startService(service);
        addChildService(service);
    }
}