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

public void addComponent(String componentName, final Component component) {
    ObjectHelper.notNull(component, "component");
    synchronized (components) {
        if (components.containsKey(componentName)) {
            throw new IllegalArgumentException("Cannot add component as its already previously added: " + componentName);
        }
        component.setCamelContext(this);
        components.put(componentName, component);
        for (LifecycleStrategy strategy : lifecycleStrategies) {
            strategy.onComponentAdd(componentName, component);
        }

        // keep reference to properties component up to date
        if (component instanceof PropertiesComponent && "properties".equals(componentName)) {
            propertiesComponent = (PropertiesComponent) component;
        }
    }
}

public Component removeComponent(String componentName) {
    synchronized (components) {
        Component oldComponent = components.remove(componentName);
        if (oldComponent != null) {
            try {
                stopServices(oldComponent);
            } catch (Exception e) {
                log.warn("Error stopping component " + oldComponent + ". This exception will be ignored.", e);
            }
            for (LifecycleStrategy strategy : lifecycleStrategies) {
                strategy.onComponentRemove(componentName, oldComponent);
            }
        }
        // keep reference to properties component up to date
        if (oldComponent != null && "properties".equals(componentName)) {
            propertiesComponent = null;
        }
        return oldComponent;
    }
}

public void testService() throws Exception {
    // JMX tests dont work well on AIX CI servers (hangs them)
    if (isPlatform("aix")) {
        return;
    }

    // must enable always as CamelContext has been started
    // and we add the service manually below
    context.getManagementStrategy().getManagementAgent().setRegisterAlways(true);

    MyService service = new MyService();
    for (LifecycleStrategy strategy : context.getLifecycleStrategies()) {
        strategy.onServiceAdd(context, service, null);
    }

    MBeanServer mbeanServer = getMBeanServer();

    Set<ObjectName> set = mbeanServer.queryNames(new ObjectName("*:type=services,*"), null);
    assertEquals(9, set.size());
}

public void testNonManagedService() throws Exception {
    // JMX tests dont work well on AIX CI servers (hangs them)
    if (isPlatform("aix")) {
        return;
    }

    // must enable always as CamelContext has been started
    // and we add the service manually below
    context.getManagementStrategy().getManagementAgent().setRegisterAlways(true);

    MyNonService service = new MyNonService();
    for (LifecycleStrategy strategy : context.getLifecycleStrategies()) {
        strategy.onServiceAdd(context, service, null);
    }

    MBeanServer mbeanServer = getMBeanServer();

    Set<ObjectName> set = mbeanServer.queryNames(new ObjectName("*:type=services,*"), null);
    assertEquals(8, set.size());
}

/**
 * Wraps the given output in an error handler
 *
 * @param routeContext the route context
 * @param output the output
 * @return the output wrapped with the error handler
 * @throws Exception can be thrown if failed to create error handler builder
 */
protected Processor wrapInErrorHandler(RouteContext routeContext, Processor output) throws Exception {
    ErrorHandlerFactory builder = routeContext.getRoute().getErrorHandlerBuilder();
    // create error handler
    Processor errorHandler = builder.createErrorHandler(routeContext, output);

    // invoke lifecycles so we can manage this error handler builder
    for (LifecycleStrategy strategy : routeContext.getCamelContext().getLifecycleStrategies()) {
        strategy.onErrorHandlerAdd(routeContext, errorHandler, builder);
    }

    return errorHandler;
}

public Endpoint addEndpoint(String uri, Endpoint endpoint) throws Exception {
    Endpoint oldEndpoint;

    startService(endpoint);
    oldEndpoint = endpoints.remove(getEndpointKey(uri));
    for (LifecycleStrategy strategy : lifecycleStrategies) {
        strategy.onEndpointAdd(endpoint);
    }
    addEndpointToRegistry(uri, endpoint);
    if (oldEndpoint != null) {
        stopServices(oldEndpoint);
    }

    return oldEndpoint;
}

public Collection<Endpoint> removeEndpoints(String uri) throws Exception {
    Collection<Endpoint> answer = new ArrayList<Endpoint>();
    Endpoint oldEndpoint = endpoints.remove(getEndpointKey(uri));
    if (oldEndpoint != null) {
        answer.add(oldEndpoint);
        stopServices(oldEndpoint);
    } else {
        for (Map.Entry<EndpointKey, Endpoint> entry : endpoints.entrySet()) {
            oldEndpoint = entry.getValue();
            if (EndpointHelper.matchEndpoint(this, oldEndpoint.getEndpointUri(), uri)) {
                try {
                    stopServices(oldEndpoint);
                } catch (Exception e) {
                    log.warn("Error stopping endpoint " + oldEndpoint + ". This exception will be ignored.", e);
                }
                answer.add(oldEndpoint);
                endpoints.remove(entry.getKey());
            }
        }
    }

    // notify lifecycle its being removed
    for (Endpoint endpoint : answer) {
        for (LifecycleStrategy strategy : lifecycleStrategies) {
            strategy.onEndpointRemove(endpoint);
        }
    }

    return answer;
}

public boolean removeService(Object object) throws Exception {
    if (object instanceof Endpoint) {
        removeEndpoint((Endpoint) object);
        return true;
    }
    if (object instanceof Service) {
        Service service = (Service) object;
        for (LifecycleStrategy strategy : lifecycleStrategies) {
            strategy.onServiceRemove(this, service, null);
        }
        return servicesToStop.remove(service);
    }
    return false;
}

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

private List<Runnable> doShutdownNow(ExecutorService executorService, boolean failSafe) {
    ObjectHelper.notNull(executorService, "executorService");

    List<Runnable> answer = null;
    if (!executorService.isShutdown()) {
        if (failSafe) {
            // log as warn, as we shutdown as fail-safe, so end user should see more details in the log.
            LOG.warn("Forcing shutdown of ExecutorService: {}", executorService);
        } else {
            LOG.debug("Forcing shutdown of ExecutorService: {}", executorService);
        }
        answer = executorService.shutdownNow();
        if (LOG.isTraceEnabled()) {
            LOG.trace("Shutdown of ExecutorService: {} is shutdown: {} and terminated: {}.",
                executorService, executorService.isShutdown(), executorService.isTerminated());
        }
    }

    // let lifecycle strategy be notified as well which can let it be managed in JMX as well
    ThreadPoolExecutor threadPool = null;
    if (executorService instanceof ThreadPoolExecutor) {
        threadPool = (ThreadPoolExecutor) executorService;
    } else if (executorService instanceof SizedScheduledExecutorService) {
        threadPool = ((SizedScheduledExecutorService) executorService).getScheduledThreadPoolExecutor();
    }
    if (threadPool != null) {
        for (LifecycleStrategy lifecycle : camelContext.getLifecycleStrategies()) {
            lifecycle.onThreadPoolRemove(camelContext, threadPool);
        }
    }

    // remove reference as its shutdown (do not remove if fail-safe)
    if (!failSafe) {
        executorServices.remove(executorService);
    }

    return answer;
}

private void doAddService(Object object, boolean stopOnShutdown, boolean forceStart) throws Exception {

        // inject CamelContext
        if (object instanceof CamelContextAware) {
            CamelContextAware aware = (CamelContextAware) object;
            aware.setCamelContext(this);
        }

        if (object instanceof Service) {
            Service service = (Service) object;

            for (LifecycleStrategy strategy : lifecycleStrategies) {
                if (service instanceof Endpoint) {
                    // use specialized endpoint add
                    strategy.onEndpointAdd((Endpoint) service);
                } else {
                    strategy.onServiceAdd(this, service, null);
                }
            }

            if (!forceStart) {
                // now start the service (and defer starting if CamelContext is starting up itself)
                deferStartService(object, stopOnShutdown);
            } else {
                // 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);
                    }
                }
                ServiceHelper.startService(service);
            }
        }
    }

public List<LifecycleStrategy> getLifecycleStrategies() {
    return lifecycleStrategies;
}

public void setLifecycleStrategies(List<LifecycleStrategy> lifecycleStrategies) {
    this.lifecycleStrategies = lifecycleStrategies;
}

public void addLifecycleStrategy(LifecycleStrategy lifecycleStrategy) {
    this.lifecycleStrategies.add(lifecycleStrategy);
}

protected void doStop() throws Exception {

        // if we are stopping CamelContext then we are shutting down
        boolean isShutdownCamelContext = camelContext.isStopping();

        if (isShutdownCamelContext || isRemovingRoutes()) {
            // need to call onRoutesRemove when the CamelContext is shutting down or Route is shutdown
            for (LifecycleStrategy strategy : camelContext.getLifecycleStrategies()) {
                strategy.onRoutesRemove(routes);
            }
        }

        for (Route route : routes) {
            LOG.debug("Stopping services on route: {}", route.getId());

            // gather list of services to stop as we need to start child services as well
            Set<Service> services = gatherChildServices(route, true);

            // stop services
            stopChildService(route, services, isShutdownCamelContext);

            // stop the route itself
            if (isShutdownCamelContext) {
                ServiceHelper.stopAndShutdownServices(route);
            } else {
                ServiceHelper.stopServices(route);
            }

            // invoke callbacks on route policy
            if (route.getRouteContext().getRoutePolicyList() != null) {
                for (RoutePolicy routePolicy : route.getRouteContext().getRoutePolicyList()) {
                    routePolicy.onStop(route);
                }
            }
            // fire event
            EventHelper.notifyRouteStopped(camelContext, route);
        }
        if (isRemovingRoutes()) {
            camelContext.removeRouteCollection(routes);
        }
        // need to warm up again
        warmUpDone.set(false);
    }

private boolean doShutdown(ExecutorService executorService, long shutdownAwaitTermination, boolean failSafe) {
    if (executorService == null) {
        return false;
    }

    boolean warned = false;

    // shutting down a thread pool is a 2 step process. First we try graceful, and if that fails, then we go more aggressively
    // and try shutting down again. In both cases we wait at most the given shutdown timeout value given
    // (total wait could then be 2 x shutdownAwaitTermination, but when we shutdown the 2nd time we are aggressive and thus
    // we ought to shutdown much faster)
    if (!executorService.isShutdown()) {
        StopWatch watch = new StopWatch();

        LOG.trace("Shutdown of ExecutorService: {} with await termination: {} millis", executorService, shutdownAwaitTermination);
        executorService.shutdown();

        if (shutdownAwaitTermination > 0) {
            try {
                if (!awaitTermination(executorService, shutdownAwaitTermination)) {
                    warned = true;
                    LOG.warn("Forcing shutdown of ExecutorService: {} due first await termination elapsed.", executorService);
                    executorService.shutdownNow();
                    // we are now shutting down aggressively, so wait to see if we can completely shutdown or not
                    if (!awaitTermination(executorService, shutdownAwaitTermination)) {
                        LOG.warn("Cannot completely force shutdown of ExecutorService: {} due second await termination elapsed.", executorService);
                    }
                }
            } catch (InterruptedException e) {
                warned = true;
                LOG.warn("Forcing shutdown of ExecutorService: {} due interrupted.", executorService);
                // we were interrupted during shutdown, so force shutdown
                executorService.shutdownNow();
            }
        }

        // if we logged at WARN level, then report at INFO level when we are complete so the end user can see this in the log
        if (warned) {
            LOG.info("Shutdown of ExecutorService: {} is shutdown: {} and terminated: {} took: {}.",
                executorService, executorService.isShutdown(), executorService.isTerminated(), TimeUtils.printDuration(watch.taken()));
        } else if (LOG.isDebugEnabled()) {
            LOG.debug("Shutdown of ExecutorService: {} is shutdown: {} and terminated: {} took: {}.",
                executorService, executorService.isShutdown(), executorService.isTerminated(), TimeUtils.printDuration(watch.taken()));
        }
    }

    // let lifecycle strategy be notified as well which can let it be managed in JMX as well
    ThreadPoolExecutor threadPool = null;
    if (executorService instanceof ThreadPoolExecutor) {
        threadPool = (ThreadPoolExecutor) executorService;
    } else if (executorService instanceof SizedScheduledExecutorService) {
        threadPool = ((SizedScheduledExecutorService) executorService).getScheduledThreadPoolExecutor();
    }
    if (threadPool != null) {
        for (LifecycleStrategy lifecycle : camelContext.getLifecycleStrategies()) {
            lifecycle.onThreadPoolRemove(camelContext, threadPool);
        }
    }

    // remove reference as its shutdown (do not remove if fail-safe)
    if (!failSafe) {
        executorServices.remove(executorService);
    }

    return warned;
}

/**
 * Invoked when a new thread pool is created.
 * This implementation will invoke the {@link LifecycleStrategy#onThreadPoolAdd(org.apache.camel.CamelContext,
 * java.util.concurrent.ThreadPoolExecutor, String, String, String, String) LifecycleStrategy.onThreadPoolAdd} method,
 * which for example will enlist the thread pool in JMX management.
 *
 * @param executorService the thread pool
 * @param source          the source to use the thread pool
 * @param threadPoolProfileId profile id, if the thread pool was created from a thread pool profile
 */
private void onThreadPoolCreated(ExecutorService executorService, Object source, String threadPoolProfileId) {
    // add to internal list of thread pools
    executorServices.add(executorService);

    String id;
    String sourceId = null;
    String routeId = null;

    // extract id from source
    if (source instanceof NamedNode) {
        id = ((OptionalIdentifiedDefinition<?>) source).idOrCreate(this.camelContext.getNodeIdFactory());
        // and let source be the short name of the pattern
        sourceId = ((NamedNode) source).getShortName();
    } else if (source instanceof String) {
        id = (String) source;
    } else if (source != null) {
        if (source instanceof StaticService) {
            // the source is static service so its name would be unique
            id = source.getClass().getSimpleName();
        } else {
            // fallback and use the simple class name with hashcode for the id so its unique for this given source
            id = source.getClass().getSimpleName() + "(" + ObjectHelper.getIdentityHashCode(source) + ")";
        }
    } else {
        // no source, so fallback and use the simple class name from thread pool and its hashcode identity so its unique
        id = executorService.getClass().getSimpleName() + "(" + ObjectHelper.getIdentityHashCode(executorService) + ")";
    }

    // id is mandatory
    ObjectHelper.notEmpty(id, "id for thread pool " + executorService);

    // extract route id if possible
    if (source instanceof ProcessorDefinition) {
        RouteDefinition route = ProcessorDefinitionHelper.getRoute((ProcessorDefinition<?>) source);
        if (route != null) {
            routeId = route.idOrCreate(this.camelContext.getNodeIdFactory());
        }
    }

    // let lifecycle strategy be notified as well which can let it be managed in JMX as well
    ThreadPoolExecutor threadPool = null;
    if (executorService instanceof ThreadPoolExecutor) {
        threadPool = (ThreadPoolExecutor) executorService;
    } else if (executorService instanceof SizedScheduledExecutorService) {
        threadPool = ((SizedScheduledExecutorService) executorService).getScheduledThreadPoolExecutor();
    }
    if (threadPool != null) {
        for (LifecycleStrategy lifecycle : camelContext.getLifecycleStrategies()) {
            lifecycle.onThreadPoolAdd(camelContext, threadPool, id, sourceId, routeId, threadPoolProfileId);
        }
    }

    // now call strategy to allow custom logic
    onNewExecutorService(executorService);
}

@Override
@Inject(optional = true)
public void setLifecycleStrategies(List<LifecycleStrategy> lifecycleStrategies) {
    super.setLifecycleStrategies(lifecycleStrategies);
}

@Override
public List<LifecycleStrategy> getLifecycleStrategies() {
  return context.getLifecycleStrategies();
}

@Override
public void addLifecycleStrategy(LifecycleStrategy lifecycleStrategy) {
  context.addLifecycleStrategy(lifecycleStrategy);
}

@Override
public List<LifecycleStrategy> getLifecycleStrategies() {
    return context.getLifecycleStrategies();
}

@Override
public void addLifecycleStrategy(LifecycleStrategy lifecycleStrategy) {
    context.addLifecycleStrategy(lifecycleStrategy);
}

/**
 * Returns the lifecycle strategies used to handle lifecycle notifications
 *
 * @return the lifecycle strategies
 */
List<LifecycleStrategy> getLifecycleStrategies();

/**
 * Adds the given lifecycle strategy to be used.
 *
 * @param lifecycleStrategy the strategy
 */
void addLifecycleStrategy(LifecycleStrategy lifecycleStrategy);