/**
 * Stops the streamer.
 *
 * @throws IgniteException In cases if failed to stop the streamer.
 */
public void stop() throws IgniteException {
    // If the Camel Context is stopping or stopped, reject this call to stop.
    if (camelCtx.getStatus() == ServiceStatus.Stopped || camelCtx.getStatus() == ServiceStatus.Stopping)
        throw new IgniteException("Failed to stop Camel streamer (CamelContext already stopped or stopping).");

    // Stop Camel services.
    try {
        ServiceHelper.stopAndShutdownServices(camelCtx, endpoint, consumer);
    }
    catch (Exception e) {
        throw new IgniteException("Failed to stop Camel streamer [errMsg=" + e.getMessage() + ']');
    }

    U.log(log, "Stopped Camel streamer, formerly consuming from endpoint URI: " + endpointUri);
}
 

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

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

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

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

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

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

@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 doShutdown() throws Exception {
    ServiceHelper.stopAndShutdownServices(processors, errorHandlers, aggregationStrategy);
    // only clear error handlers when shutting down
    errorHandlers.clear();

    if (shutdownExecutorService && executorService != null) {
        getCamelContext().getExecutorServiceManager().shutdownNow(executorService);
    }
    if (aggregateExecutorService != null) {
        getCamelContext().getExecutorServiceManager().shutdownNow(aggregateExecutorService);
    }
}
 

@Override
protected void doShutdown() throws Exception {
    ServiceHelper.stopAndShutdownServices(processors, errorHandlers, aggregationStrategy);
    // only clear error handlers when shutting down
    errorHandlers.clear();

    if (shutdownExecutorService && executorService != null) {
        getCamelContext().getExecutorServiceManager().shutdownNow(executorService);
    }
    if (aggregateExecutorService != null) {
        getCamelContext().getExecutorServiceManager().shutdownNow(aggregateExecutorService);
    }
}
 

@Override
protected void doShutdown() throws Exception {
    ServiceHelper.stopAndShutdownServices(processors, errorHandlers, aggregationStrategy);
    // only clear error handlers when shutting down
    errorHandlers.clear();

    if (shutdownExecutorService && executorService != null) {
        getCamelContext().getExecutorServiceManager().shutdownNow(executorService);
    }
    if (aggregateExecutorService != null) {
        getCamelContext().getExecutorServiceManager().shutdownNow(aggregateExecutorService);
    }
}
 

@Override
protected void doShutdown() throws Exception {
    ServiceHelper.stopAndShutdownServices(processors, errorHandlers, aggregationStrategy);
    // only clear error handlers when shutting down
    errorHandlers.clear();

    if (shutdownExecutorService && executorService != null) {
        getCamelContext().getExecutorServiceManager().shutdownNow(executorService);
    }
    if (aggregateExecutorService != null) {
        getCamelContext().getExecutorServiceManager().shutdownNow(aggregateExecutorService);
    }
}
 

@Override
protected void doShutdown() throws Exception {
    ServiceHelper.stopAndShutdownServices(processors, errorHandlers, aggregationStrategy);
    // only clear error handlers when shutting down
    errorHandlers.clear();

    if (shutdownExecutorService && executorService != null) {
        getCamelContext().getExecutorServiceManager().shutdownNow(executorService);
    }
    if (aggregateExecutorService != null) {
        getCamelContext().getExecutorServiceManager().shutdownNow(aggregateExecutorService);
    }
}
 

@Override
protected void doShutdown() throws Exception {
    ServiceHelper.stopAndShutdownServices(processors, errorHandlers, aggregationStrategy);
    // only clear error handlers when shutting down
    errorHandlers.clear();

    if (shutdownExecutorService && executorService != null) {
        getCamelContext().getExecutorServiceManager().shutdownNow(executorService);
    }
    if (aggregateExecutorService != null) {
        getCamelContext().getExecutorServiceManager().shutdownNow(aggregateExecutorService);
    }
}
 

@Override
protected void doStop() throws Exception {
    ServiceHelper.stopServices(jsonMarshalProducer, jsonMarshalEndpoint);
    super.doStop();
}
 

protected Processor createErrorHandler(RouteContext routeContext, Exchange exchange, Processor processor) {
    Processor answer;

    boolean tryBlock = exchange.getProperty(Exchange.TRY_ROUTE_BLOCK, false, boolean.class);

    // do not wrap in error handler if we are inside a try block
    if (!tryBlock && routeContext != null) {
        // wrap the producer in error handler so we have fine grained error handling on
        // the output side instead of the input side
        // this is needed to support redelivery on that output alone and not doing redelivery
        // for the entire multicast block again which will start from scratch again

        // create key for cache
        final PreparedErrorHandler key = new PreparedErrorHandler(routeContext, processor);

        // lookup cached first to reuse and preserve memory
        answer = errorHandlers.get(key);
        if (answer != null) {
            LOG.trace("Using existing error handler for: {}", processor);
            return answer;
        }

        LOG.trace("Creating error handler for: {}", processor);
        ErrorHandlerFactory builder = routeContext.getRoute().getErrorHandlerBuilder();
        // create error handler (create error handler directly to keep it light weight,
        // instead of using ProcessorDefinition.wrapInErrorHandler)
        try {
            processor = builder.createErrorHandler(routeContext, processor);

            // and wrap in unit of work processor so the copy exchange also can run under UoW
            answer = createUnitOfWorkProcessor(routeContext, processor, exchange);

            boolean child = exchange.getProperty(Exchange.PARENT_UNIT_OF_WORK, UnitOfWork.class) != null;

            // must start the error handler
            ServiceHelper.startServices(answer);

            // here we don't cache the child unit of work
            if (!child) {
                // add to cache
                errorHandlers.putIfAbsent(key, answer);
            }

        } catch (Exception e) {
            throw ObjectHelper.wrapRuntimeCamelException(e);
        }
    } else {
        // and wrap in unit of work processor so the copy exchange also can run under UoW
        answer = createUnitOfWorkProcessor(routeContext, processor, exchange);
    }

    return answer;
}
 

@Override
protected void doStop() throws Exception {
    ServiceHelper.stopServices(processors, errorHandlers, aggregationStrategy);
}
 

protected Processor createErrorHandler(RouteContext routeContext, Exchange exchange, Processor processor) {
    Processor answer;

    boolean tryBlock = exchange.getProperty(Exchange.TRY_ROUTE_BLOCK, false, boolean.class);

    // do not wrap in error handler if we are inside a try block
    if (!tryBlock && routeContext != null) {
        // wrap the producer in error handler so we have fine grained error handling on
        // the output side instead of the input side
        // this is needed to support redelivery on that output alone and not doing redelivery
        // for the entire multicast block again which will start from scratch again

        // create key for cache
        final PreparedErrorHandler key = new PreparedErrorHandler(routeContext, processor);

        // lookup cached first to reuse and preserve memory
        answer = errorHandlers.get(key);
        if (answer != null) {
            LOG.trace("Using existing error handler for: {}", processor);
            return answer;
        }

        LOG.trace("Creating error handler for: {}", processor);
        ErrorHandlerFactory builder = routeContext.getRoute().getErrorHandlerBuilder();
        // create error handler (create error handler directly to keep it light weight,
        // instead of using ProcessorDefinition.wrapInErrorHandler)
        try {
            processor = builder.createErrorHandler(routeContext, processor);

            // and wrap in unit of work processor so the copy exchange also can run under UoW
            answer = createUnitOfWorkProcessor(routeContext, processor, exchange);

            boolean child = exchange.getProperty(Exchange.PARENT_UNIT_OF_WORK, UnitOfWork.class) != null;

            // must start the error handler
            ServiceHelper.startServices(answer);

            // here we don't cache the child unit of work
            if (!child) {
                // add to cache
                errorHandlers.putIfAbsent(key, answer);
            }

        } catch (Exception e) {
            throw ObjectHelper.wrapRuntimeCamelException(e);
        }
    } else {
        // and wrap in unit of work processor so the copy exchange also can run under UoW
        answer = createUnitOfWorkProcessor(routeContext, processor, exchange);
    }

    return answer;
}
 

@Override
protected void doStop() throws Exception {
    ServiceHelper.stopServices(processors, errorHandlers, aggregationStrategy);
}
 

protected Processor createErrorHandler(RouteContext routeContext, Exchange exchange, Processor processor) {
    Processor answer;

    boolean tryBlock = exchange.getProperty(Exchange.TRY_ROUTE_BLOCK, false, boolean.class);

    // do not wrap in error handler if we are inside a try block
    if (!tryBlock && routeContext != null) {
        // wrap the producer in error handler so we have fine grained error handling on
        // the output side instead of the input side
        // this is needed to support redelivery on that output alone and not doing redelivery
        // for the entire multicast block again which will start from scratch again

        // create key for cache
        final PreparedErrorHandler key = new PreparedErrorHandler(routeContext, processor);

        // lookup cached first to reuse and preserve memory
        answer = errorHandlers.get(key);
        if (answer != null) {
            LOG.trace("Using existing error handler for: {}", processor);
            return answer;
        }

        LOG.trace("Creating error handler for: {}", processor);
        ErrorHandlerFactory builder = routeContext.getRoute().getErrorHandlerBuilder();
        // create error handler (create error handler directly to keep it light weight,
        // instead of using ProcessorDefinition.wrapInErrorHandler)
        try {
            processor = builder.createErrorHandler(routeContext, processor);

            // and wrap in unit of work processor so the copy exchange also can run under UoW
            answer = createUnitOfWorkProcessor(routeContext, processor, exchange);

            boolean child = exchange.getProperty(Exchange.PARENT_UNIT_OF_WORK, UnitOfWork.class) != null;

            // must start the error handler
            ServiceHelper.startServices(answer);

            // here we don't cache the child unit of work
            if (!child) {
                // add to cache
                errorHandlers.putIfAbsent(key, answer);
            }

        } catch (Exception e) {
            throw ObjectHelper.wrapRuntimeCamelException(e);
        }
    } else {
        // and wrap in unit of work processor so the copy exchange also can run under UoW
        answer = createUnitOfWorkProcessor(routeContext, processor, exchange);
    }

    return answer;
}
 

@Override
protected void doStop() throws Exception {
    ServiceHelper.stopServices(processors, errorHandlers, aggregationStrategy);
}
 

protected Processor createErrorHandler(RouteContext routeContext, Exchange exchange, Processor processor) {
    Processor answer;

    boolean tryBlock = exchange.getProperty(Exchange.TRY_ROUTE_BLOCK, false, boolean.class);

    // do not wrap in error handler if we are inside a try block
    if (!tryBlock && routeContext != null) {
        // wrap the producer in error handler so we have fine grained error handling on
        // the output side instead of the input side
        // this is needed to support redelivery on that output alone and not doing redelivery
        // for the entire multicast block again which will start from scratch again

        // create key for cache
        final PreparedErrorHandler key = new PreparedErrorHandler(routeContext, processor);

        // lookup cached first to reuse and preserve memory
        answer = errorHandlers.get(key);
        if (answer != null) {
            LOG.trace("Using existing error handler for: {}", processor);
            return answer;
        }

        LOG.trace("Creating error handler for: {}", processor);
        ErrorHandlerFactory builder = routeContext.getRoute().getErrorHandlerBuilder();
        // create error handler (create error handler directly to keep it light weight,
        // instead of using ProcessorDefinition.wrapInErrorHandler)
        try {
            processor = builder.createErrorHandler(routeContext, processor);

            // and wrap in unit of work processor so the copy exchange also can run under UoW
            answer = createUnitOfWorkProcessor(routeContext, processor, exchange);

            boolean child = exchange.getProperty(Exchange.PARENT_UNIT_OF_WORK, UnitOfWork.class) != null;

            // must start the error handler
            ServiceHelper.startServices(answer);

            // here we don't cache the child unit of work
            if (!child) {
                // add to cache
                errorHandlers.putIfAbsent(key, answer);
            }

        } catch (Exception e) {
            throw ObjectHelper.wrapRuntimeCamelException(e);
        }
    } else {
        // and wrap in unit of work processor so the copy exchange also can run under UoW
        answer = createUnitOfWorkProcessor(routeContext, processor, exchange);
    }

    return answer;
}
 

@Override
protected void doStop() throws Exception {
    ServiceHelper.stopServices(processors, errorHandlers, aggregationStrategy);
}
 

protected Processor createErrorHandler(RouteContext routeContext, Exchange exchange, Processor processor) {
    Processor answer;

    boolean tryBlock = exchange.getProperty(Exchange.TRY_ROUTE_BLOCK, false, boolean.class);

    // do not wrap in error handler if we are inside a try block
    if (!tryBlock && routeContext != null) {
        // wrap the producer in error handler so we have fine grained error handling on
        // the output side instead of the input side
        // this is needed to support redelivery on that output alone and not doing redelivery
        // for the entire multicast block again which will start from scratch again

        // create key for cache
        final PreparedErrorHandler key = new PreparedErrorHandler(routeContext, processor);

        // lookup cached first to reuse and preserve memory
        answer = errorHandlers.get(key);
        if (answer != null) {
            LOG.trace("Using existing error handler for: {}", processor);
            return answer;
        }

        LOG.trace("Creating error handler for: {}", processor);
        ErrorHandlerFactory builder = routeContext.getRoute().getErrorHandlerBuilder();
        // create error handler (create error handler directly to keep it light weight,
        // instead of using ProcessorDefinition.wrapInErrorHandler)
        try {
            processor = builder.createErrorHandler(routeContext, processor);

            // and wrap in unit of work processor so the copy exchange also can run under UoW
            answer = createUnitOfWorkProcessor(routeContext, processor, exchange);

            boolean child = exchange.getProperty(Exchange.PARENT_UNIT_OF_WORK, UnitOfWork.class) != null;

            // must start the error handler
            ServiceHelper.startServices(answer);

            // here we don't cache the child unit of work
            if (!child) {
                // add to cache
                errorHandlers.putIfAbsent(key, answer);
            }

        } catch (Exception e) {
            throw ObjectHelper.wrapRuntimeCamelException(e);
        }
    } else {
        // and wrap in unit of work processor so the copy exchange also can run under UoW
        answer = createUnitOfWorkProcessor(routeContext, processor, exchange);
    }

    return answer;
}
 

@Override
protected void doStop() throws Exception {
    ServiceHelper.stopServices(processors, errorHandlers, aggregationStrategy);
}
 

protected Processor createErrorHandler(RouteContext routeContext, Exchange exchange, Processor processor) {
    Processor answer;

    boolean tryBlock = exchange.getProperty(Exchange.TRY_ROUTE_BLOCK, false, boolean.class);

    // do not wrap in error handler if we are inside a try block
    if (!tryBlock && routeContext != null) {
        // wrap the producer in error handler so we have fine grained error handling on
        // the output side instead of the input side
        // this is needed to support redelivery on that output alone and not doing redelivery
        // for the entire multicast block again which will start from scratch again

        // create key for cache
        final PreparedErrorHandler key = new PreparedErrorHandler(routeContext, processor);

        // lookup cached first to reuse and preserve memory
        answer = errorHandlers.get(key);
        if (answer != null) {
            LOG.trace("Using existing error handler for: {}", processor);
            return answer;
        }

        LOG.trace("Creating error handler for: {}", processor);
        ErrorHandlerFactory builder = routeContext.getRoute().getErrorHandlerBuilder();
        // create error handler (create error handler directly to keep it light weight,
        // instead of using ProcessorDefinition.wrapInErrorHandler)
        try {
            processor = builder.createErrorHandler(routeContext, processor);

            // and wrap in unit of work processor so the copy exchange also can run under UoW
            answer = createUnitOfWorkProcessor(routeContext, processor, exchange);

            boolean child = exchange.getProperty(Exchange.PARENT_UNIT_OF_WORK, UnitOfWork.class) != null;

            // must start the error handler
            ServiceHelper.startServices(answer);

            // here we don't cache the child unit of work
            if (!child) {
                // add to cache
                errorHandlers.putIfAbsent(key, answer);
            }

        } catch (Exception e) {
            throw ObjectHelper.wrapRuntimeCamelException(e);
        }
    } else {
        // and wrap in unit of work processor so the copy exchange also can run under UoW
        answer = createUnitOfWorkProcessor(routeContext, processor, exchange);
    }

    return answer;
}
 

@Override
protected void doStop() throws Exception {
    ServiceHelper.stopServices(processors, errorHandlers, aggregationStrategy);
}