Java 类org.apache.camel.Service 实例源码

@Override
public Processor createProcessor(RouteContext routeContext) throws Exception {
    Policy policy = resolvePolicy(routeContext);
    ObjectHelper.notNull(policy, "policy", this);

    // before wrap
    policy.beforeWrap(routeContext, this);

    // create processor after the before wrap
    Processor childProcessor = this.createChildProcessor(routeContext, true);

    // wrap
    Processor target = policy.wrap(routeContext, childProcessor);

    if (!(target instanceof Service)) {
        // wrap the target so it becomes a service and we can manage its lifecycle
        target = new WrapProcessor(target, childProcessor);
    }
    return target;
}

@Override
public Processor createProcessor(RouteContext routeContext) throws Exception {
    Policy policy = resolvePolicy(routeContext);
    ObjectHelper.notNull(policy, "policy", this);

    // before wrap
    policy.beforeWrap(routeContext, this);

    // create processor after the before wrap
    Processor childProcessor = this.createChildProcessor(routeContext, true);

    // wrap
    Processor target = policy.wrap(routeContext, childProcessor);

    if (!(target instanceof Service)) {
        // wrap the target so it becomes a service and we can manage its lifecycle
        target = new WrapProcessor(target, childProcessor);
    }
    return target;
}

/**
 * Resumes each element of the given {@code services} if {@code services} itself is
 * not {@code null}, otherwise this method would return immediately.
 * <p/>
 * If there's any exception being thrown while resuming the elements one after the
 * other this method would rethrow the <b>first</b> such exception being thrown.
 * 
 * @see #resumeService(Object)
 */
public static void resumeServices(Collection<?> services) throws Exception {
    if (services == null) {
        return;
    }
    Exception firstException = null;
    for (Object value : services) {
        if (value instanceof Service) {
            Service service = (Service)value;
            try {
                resumeService(service);
            } catch (Exception e) {
                if (LOG.isDebugEnabled()) {
                    LOG.debug("Caught exception resuming service: " + service, e);
                }
                if (firstException == null) {
                    firstException = e;
                }
            }
        }
    }
    if (firstException != null) {
        throw firstException;
    }
}

/**
 * Suspends each element of the given {@code services} if {@code services} itself is
 * not {@code null}, otherwise this method would return immediately.
 * <p/>
 * If there's any exception being thrown while suspending the elements one after the
 * other this method would rethrow the <b>first</b> such exception being thrown.
 * 
 * @see #suspendService(Object)
 */
public static void suspendServices(Collection<?> services) throws Exception {
    if (services == null) {
        return;
    }
    Exception firstException = null;
    for (Object value : services) {
        if (value instanceof Service) {
            Service service = (Service)value;
            try {
                suspendService(service);
            } catch (Exception e) {
                if (LOG.isDebugEnabled()) {
                    LOG.debug("Caught exception suspending service: " + service, e);
                }
                if (firstException == null) {
                    firstException = e;
                }
            }
        }
    }
    if (firstException != null) {
        throw firstException;
    }
}

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

/**
 * Calculates the total number of inflight exchanges for the given route
 *
 * @param order the route
 * @return number of inflight exchanges
 */
protected static int getPendingInflightExchanges(RouteStartupOrder order) {
    int inflight = 0;

    // the consumer is the 1st service so we always get the consumer
    // the child services are EIPs in the routes which may also have pending
    // inflight exchanges (such as the aggregator)
    for (Service service : order.getServices()) {
        Set<Service> children = ServiceHelper.getChildServices(service);
        for (Service child : children) {
            if (child instanceof ShutdownAware) {
                inflight += ((ShutdownAware) child).getPendingExchangesSize();
            }
        }
    }

    return inflight;
}

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

private void shutdownServices(Object service) {
    // do not rethrow exception as we want to keep shutting down in case of problems

    // allow us to do custom work before delegating to service helper
    try {
        if (service instanceof Service) {
            ServiceHelper.stopAndShutdownService(service);
        } else if (service instanceof Collection) {
            ServiceHelper.stopAndShutdownServices((Collection<?>)service);
        }
    } catch (Throwable e) {
        log.warn("Error occurred while shutting down service: " + service + ". This exception will be ignored.", e);
        // fire event
        EventHelper.notifyServiceStopFailure(this, service, e);
    }
}

/**
 * Gather all the endpoints this route service uses
 * <p/>
 * This implementation finds the endpoints by searching all the child services
 * for {@link org.apache.camel.EndpointAware} processors which uses an endpoint.
 */
public Set<Endpoint> gatherEndpoints() {
    Set<Endpoint> answer = new LinkedHashSet<Endpoint>();
    for (Route route : routes) {
        Set<Service> services = gatherChildServices(route, true);
        for (Service service : services) {
            if (service instanceof EndpointAware) {
                Endpoint endpoint = ((EndpointAware) service).getEndpoint();
                if (endpoint != null) {
                    answer.add(endpoint);
                }
            }
        }
    }
    return answer;
}

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

/**
 * Gather the route scoped error handler from the given route
 */
private void doGetRouteScopedErrorHandler(Set<Service> services, Route route) {
    // only include error handlers if they are route scoped
    boolean includeErrorHandler = !routeDefinition.isContextScopedErrorHandler(route.getRouteContext().getCamelContext());
    List<Service> extra = new ArrayList<Service>();
    if (includeErrorHandler) {
        for (Service service : services) {
            if (service instanceof Channel) {
                Processor eh = ((Channel) service).getErrorHandler();
                if (eh != null && eh instanceof Service) {
                    extra.add((Service) eh);
                }
            }
        }
    }
    if (!extra.isEmpty()) {
        services.addAll(extra);
    }
}

/**
 * Gather all other kind of route scoped services from the given route, except error handler
 */
private void doGetRouteScopedServices(List<Service> services, Route route) {
    for (ProcessorDefinition<?> output : route.getRouteContext().getRoute().getOutputs()) {
        if (output instanceof OnExceptionDefinition) {
            OnExceptionDefinition onExceptionDefinition = (OnExceptionDefinition) output;
            if (onExceptionDefinition.isRouteScoped()) {
                Processor errorHandler = onExceptionDefinition.getErrorHandler(route.getId());
                if (errorHandler != null && errorHandler instanceof Service) {
                    services.add((Service) errorHandler);
                }
            }
        } else if (output instanceof OnCompletionDefinition) {
            OnCompletionDefinition onCompletionDefinition = (OnCompletionDefinition) output;
            if (onCompletionDefinition.isRouteScoped()) {
                Processor onCompletionProcessor = onCompletionDefinition.getOnCompletion(route.getId());
                if (onCompletionProcessor != null && onCompletionProcessor instanceof Service) {
                    services.add((Service) onCompletionProcessor);
                }
            }
        }
    }
}

@Override
public void onCamelContextStarted(CamelContext context, boolean alreadyStarted) throws Exception {
    // new services may be added while starting a service
    // so use a while loop to get the newly added services as well
    while (!services.isEmpty()) {
        Service service = services.iterator().next();
        try {
            ServiceHelper.startService(service);
        } catch (Exception e) {
            if (service instanceof Endpoint) {
                Endpoint endpoint = (Endpoint) service;
                throw new ResolveEndpointFailedException(endpoint.getEndpointUri(), e);
            } else {
                throw e;
            }
        } finally {
            services.remove(service);
        }
    }
}

protected CamelContext createCamelContext() throws Exception {
    setUseRouteBuilder(false);

    final AbstractXmlApplicationContext applicationContext = new ClassPathXmlApplicationContext(
            "org/apache/camel/dataformat/jibx/SpringJibxConfigurationTest.xml");

    setCamelContextService(new Service() {
        public void start() throws Exception {
            applicationContext.start();
        }

        public void stop() throws Exception {
            applicationContext.stop();
        }
    });

    return SpringCamelContext.springCamelContext(applicationContext);
}

protected CamelContext createCamelContext() throws Exception {
    setUseRouteBuilder(false);
    final AbstractXmlApplicationContext applicationContext = new ClassPathXmlApplicationContext("org/apache/camel/jaxb/CamelJaxbNoNamespaceSchemaLocationTest.xml");
    setCamelContextService(new Service() {
        public void start() throws Exception {
            applicationContext.start();
        }

        public void stop() throws Exception {
            applicationContext.stop();
        }
    });

    return SpringCamelContext.springCamelContext(applicationContext);

}

protected CamelContext createCamelContext() throws Exception {
    setUseRouteBuilder(false);
    final AbstractXmlApplicationContext applicationContext = new ClassPathXmlApplicationContext("org/apache/camel/jaxb/CamelJaxbTest.xml");
    setCamelContextService(new Service() {
        public void start() throws Exception {
            applicationContext.start();
        }

        public void stop() throws Exception {
            applicationContext.stop();
        }
    });

    return SpringCamelContext.springCamelContext(applicationContext);

}

private static void doStopCamelContext(CamelContext context,
                                       Service camelContextService) throws Exception {
    if (camelContextService != null) {
        if (camelContextService == threadService.get()) {
            threadService.remove();
        }
        camelContextService.stop();
        camelContextService = null;
    } else {
        if (context != null) {
            if (context == threadCamelContext.get()) {
                threadCamelContext.remove();
            }
            context.stop();
            context = null;
        }
    }
}

protected CamelContext createCamelContext() throws Exception {
    setUseRouteBuilder(false);

    final AbstractXmlApplicationContext applicationContext = new ClassPathXmlApplicationContext("org/apache/camel/dataformat/xstream/SpringMarshalDomainObjectJSONTest.xml");
    setCamelContextService(new Service() {
        public void start() throws Exception {
            applicationContext.start();

        }

        public void stop() throws Exception {
            applicationContext.stop();
        }
    });

    return SpringCamelContext.springCamelContext(applicationContext);        
}

protected CamelContext createCamelContext() throws Exception {
    setUseRouteBuilder(false);

    final AbstractXmlApplicationContext applicationContext = new ClassPathXmlApplicationContext(
        "org/apache/camel/dataformat/xstream/SpringXStreamConfigurationTest.xml");

    setCamelContextService(new Service() {
        public void start() throws Exception {
            applicationContext.start();
        }

        public void stop() throws Exception {
            applicationContext.stop();
        }
    });

    return SpringCamelContext.springCamelContext(applicationContext);
}

protected CamelContext createCamelContext() throws Exception {
    setUseRouteBuilder(false);

    final AbstractXmlApplicationContext applicationContext = new ClassPathXmlApplicationContext("org/apache/camel/dataformat/xstream/SpringMarshalListTest.xml");
    setCamelContextService(new Service() {
        public void start() throws Exception {
            applicationContext.start();

        }

        public void stop() throws Exception {
            applicationContext.stop();
        }
    });

    return SpringCamelContext.springCamelContext(applicationContext);        
}

protected CamelContext createCamelContext() throws Exception {
    setUseRouteBuilder(false);

    final AbstractXmlApplicationContext applicationContext = new ClassPathXmlApplicationContext(
        "org/apache/camel/dataformat/xstream/SpringMarshalOmitFieldsTest.xml");

    setCamelContextService(new Service() {
        public void start() throws Exception {
            applicationContext.start();
        }

        public void stop() throws Exception {
            applicationContext.stop();
        }
    });

    return SpringCamelContext.springCamelContext(applicationContext);
}

public static CamelContext createSpringCamelContext(ContextTestSupport test, String classpathUri) throws Exception {
    test.setUseRouteBuilder(false);

    final AbstractXmlApplicationContext applicationContext = new ClassPathXmlApplicationContext(classpathUri);
    test.setCamelContextService(new Service() {
        public void start() throws Exception {
            applicationContext.start();
        }

        public void stop() throws Exception {
            applicationContext.stop();
        }
    });

    return SpringCamelContext.springCamelContext(applicationContext);
}

public static CamelContext createSpringCamelContext(CamelTestSupport test, String classpathUri) throws Exception {
    test.setUseRouteBuilder(false);

    final AbstractXmlApplicationContext applicationContext = new ClassPathXmlApplicationContext(classpathUri);
    test.setCamelContextService(new Service() {
        public void start() throws Exception {
            applicationContext.start();
        }

        public void stop() throws Exception {
            applicationContext.stop();
        }
    });

    return SpringCamelContext.springCamelContext(applicationContext);
}

public void onServiceAdd(CamelContext context, Service service, Route route) {
    if (!initialized) {
        // pre register so we can register later when we have been initialized
        PreRegisterService pre = new PreRegisterService();
        pre.onServiceAdd(context, service, route);
        preServices.add(pre);
        return;
    }

    // services can by any kind of misc type but also processors
    // so we have special logic when its a processor

    if (!shouldRegister(service, route)) {
        // avoid registering if not needed
        return;
    }

    Object managedObject = getManagedObjectForService(context, service, route);
    if (managedObject == null) {
        // service should not be managed
        return;
    }

    // skip already managed services, for example if a route has been restarted
    if (getManagementStrategy().isManaged(managedObject, null)) {
        LOG.trace("The service is already managed: {}", service);
        return;
    }

    try {
        manageObject(managedObject);
    } catch (Exception e) {
        LOG.warn("Could not register service: " + service + " as Service MBean.", e);
    }
}

public void onServiceRemove(CamelContext context, Service service, Route route) {
    // the agent hasn't been started
    if (!initialized) {
        return;
    }

    Object managedObject = getManagedObjectForService(context, service, route);
    if (managedObject != null) {
        try {
            unmanageObject(managedObject);
        } catch (Exception e) {
            LOG.warn("Could not unregister service: " + service + " as Service MBean.", e);
        }
    }
}

public ObjectName getObjectNameForService(CamelContext context, Service service) throws MalformedObjectNameException {
    StringBuilder buffer = new StringBuilder();
    buffer.append(domainName).append(":");
    buffer.append(KEY_CONTEXT + "=").append(getContextId(context)).append(",");
    buffer.append(KEY_TYPE + "=" + TYPE_SERVICE + ",");
    buffer.append(KEY_NAME + "=").append(service.getClass().getSimpleName());
    if (!(service instanceof StaticService)) {
        buffer.append("(").append(ObjectHelper.getIdentityHashCode(service)).append(")");
    }
    return createObjectName(buffer);
}

/**
 * Stops the given {@code value}, rethrowing the first exception caught.
 * <p/>
 * Calling this method has no effect if {@code value} is {@code null}.
 * 
 * @see Service#stop()
 * @see #stopServices(Collection)
 */
public static void stopService(Object value) throws Exception {
    if (isStopped(value)) {
        // only stop service if not already stopped
        LOG.trace("Service already stopped: {}", value);
        return;
    }
    if (value instanceof Service) {
        Service service = (Service)value;
        LOG.trace("Stopping service {}", value);
        service.stop();
    } else if (value instanceof Collection) {
        stopServices((Collection<?>)value);
    }
}

@Override
public void onServiceRemove(CamelContext context, Service service, Route route) {
    super.onServiceRemove(context, service, route);

    // if its a consumer then de-register it from the rest registry
    if (service instanceof Consumer) {
        removeRestService((Consumer) service);
    }
}

public Component getComponent(String name, boolean autoCreateComponents, boolean autoStart) {
    // synchronize the look up and auto create so that 2 threads can't
    // concurrently auto create the same component.
    synchronized (components) {
        Component component = components.get(name);
        if (component == null && autoCreateComponents) {
            try {
                if (log.isDebugEnabled()) {
                    log.debug("Using ComponentResolver: {} to resolve component with name: {}", getComponentResolver(), name);
                }
                component = getComponentResolver().resolveComponent(name, this);
                if (component != null) {
                    addComponent(name, component);
                    if (autoStart && (isStarted() || isStarting())) {
                        // If the component is looked up after the context is started, lets start it up.
                        if (component instanceof Service) {
                            startService((Service)component);
                        }
                    }
                }
            } catch (Exception e) {
                throw new RuntimeCamelException("Cannot auto create component: " + name, e);
            }
        }
        log.trace("getComponent({}) -> {}", name, component);
        return component;
    }
}

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

public boolean hasService(Object object) {
    if (object instanceof Service) {
        Service service = (Service) object;
        return servicesToStop.contains(service);
    }
    return false;
}

@Override
public <T> T hasService(Class<T> type) {
    for (Service service : servicesToStop) {
        if (type.isInstance(service)) {
            return type.cast(service);
        }
    }
    return null;
}

public Language resolveLanguage(String language) {
    Language answer;
    synchronized (languages) {
        answer = languages.get(language);

        // check if the language is singleton, if so return the shared instance
        if (answer instanceof IsSingleton) {
            boolean singleton = ((IsSingleton) answer).isSingleton();
            if (singleton) {
                return answer;
            }
        }

        // language not known or not singleton, then use resolver
        answer = getLanguageResolver().resolveLanguage(language, this);

        // inject CamelContext if aware
        if (answer != null) {
            if (answer instanceof CamelContextAware) {
                ((CamelContextAware) answer).setCamelContext(this);
            }
            if (answer instanceof Service) {
                try {
                    startService((Service) answer);
                } catch (Exception e) {
                    throw ObjectHelper.wrapRuntimeCamelException(e);
                }
            }

            languages.put(language, answer);
        }
    }

    return answer;
}

private void startService(Service service) throws Exception {
    // and register startup aware so they can be notified when
    // camel context has been started
    if (service instanceof StartupListener) {
        StartupListener listener = (StartupListener) service;
        addStartupListener(listener);
    }

    if (service instanceof CamelContextAware) {
        CamelContextAware aware = (CamelContextAware) service;
        aware.setCamelContext(this);
    }

    service.start();
}

private void startServices(Collection<?> services) throws Exception {
    for (Object element : services) {
        if (element instanceof Service) {
            startService((Service)element);
        }
    }
}

/**
 * Stats the given service
 */
protected void startService(Service service, CamelContext camelContext, Object bean, String beanName) throws Exception {
    // defer starting the service until CamelContext has started all its initial services
    if (camelContext != null) {
        camelContext.deferStartService(service, true);
    } else {
        // mo CamelContext then start service manually
        ServiceHelper.startService(service);
    }

    boolean singleton = isSingleton(bean, beanName);
    if (!singleton) {
        LOG.debug("Service is not singleton so you must remember to stop it manually {}", service);
    }
}

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

public List<Service> getServices() {
    List<Service> answer = new ArrayList<Service>();
    Collection<Route> routes = routeService.getRoutes();
    for (Route route : routes) {
        answer.addAll(route.getServices());
    }
    return answer;
}

private static void doStopCamelContext(CamelContext context, Service camelContextService) throws Exception {
    if (camelContextService != null) {
        if (camelContextService == threadService.get()) {
            threadService.remove();
        }
        camelContextService.stop();
    } else {
        if (context != null) {
            if (context == threadCamelContext.get()) {
                threadCamelContext.remove();
            }
            context.stop();
        }
    }
}

@Test(expected = HostUnavailableException.class)
public void testHostUnavailableException() throws Throwable {
    // cm-sms://sgw01.cm.nl/gateway.ashx?defaultFrom=MyBusiness&defaultMaxNumberOfParts=8&productToken=ea723fd7-da81-4826-89bc-fa7144e71c40&testConnectionOnStartup=true
    String schemedUri = "cm-sms://dummy.sgw01.cm.nl/gateway.ashx?defaultFrom=MyBusiness&defaultMaxNumberOfParts=8&productToken=ea723fd7-da81-4826-89bc-fa7144e71c40&testConnectionOnStartup=true";
    Service service = camelContext.getEndpoint(schemedUri).createProducer();
    service.start();
}