Chapter 2. Basic Principles of Route Building

In Apache Camel, pipelining is the dominant paradigm for connecting nodes in a route definition. The pipeline concept is probably most familiar to users of the UNIX operating system, where it is used to join operating system commands. For example, ls | more is an example of a command that pipes a directory listing, ls, to the page-scrolling utility, more. The basic idea of a pipeline is that the output of one command is fed into the input of the next. The natural analogy in the case of a route is for the Out message from one processor to be copied to the In message of the next processor.

Every node in a route, except for the initial endpoint, is a processor, in the sense that they inherit from the org.apache.camel.Processor interface. In other words, processors make up the basic building blocks of a DSL route. For example, DSL commands such as filter(), delayer(), setBody(), setHeader(), and to() all represent processors. When considering how processors connect together to build up a route, it is important to distinguish two different processing approaches.

The first approach is where the processor simply modifies the exchange’s In message, as shown in Figure 2.1, “Processor Modifying an In Message”. The exchange’s Out message remains null in this case.

Figure 2.1. Processor Modifying an In Message

The following route shows a setHeader() command that modifies the current In message by adding (or modifying) the BillingSystem heading:

from("activemq:orderQueue")
    .setHeader("BillingSystem", xpath("/order/billingSystem"))
    .to("activemq:billingQueue");

The second approach is where the processor creates an Out message to represent the result of the processing, as shown in Figure 2.2, “Processor Creating an Out Message”.

Figure 2.2. Processor Creating an Out Message

The following route shows a transform() command that creates an Out message with a message body containing the string, DummyBody:

from("activemq:orderQueue")
    .transform(constant("DummyBody"))
    .to("activemq:billingQueue");

where constant("DummyBody") represents a constant expression. You cannot pass the string, DummyBody, directly, because the argument to transform() must be an expression type.

Figure 2.3, “Sample Pipeline for InOnly Exchanges” shows an example of a processor pipeline for InOnly exchanges. Processor A acts by modifying the In message, while processors B and C create an Out message. The route builder links the processors together as shown. In particular, processors B and C are linked together in the form of a pipeline: that is, processor B’s Out message is moved to the In message before feeding the exchange into processor C, and processor C’s Out message is moved to the In message before feeding the exchange into the producer endpoint. Thus the processors' outputs and inputs are joined into a continuous pipeline, as shown in Figure 2.3, “Sample Pipeline for InOnly Exchanges”.

Figure 2.3. Sample Pipeline for InOnly Exchanges

Apache Camel employs the pipeline pattern by default, so you do not need to use any special syntax to create a pipeline in your routes. For example, the following route pulls messages from a userdataQueue queue, pipes the message through a Velocity template (to produce a customer address in text format), and then sends the resulting text address to the queue, envelopeAddressQueue:

from("activemq:userdataQueue")
    .to(ExchangePattern.InOut, "velocity:file:AdressTemplate.vm")
    .to("activemq:envelopeAddresses");

Where the Velocity endpoint, velocity:file:AdressTemplate.vm, specifies the location of a Velocity template file, file:AdressTemplate.vm, in the file system. The to() command changes the exchange pattern to InOut before sending the exchange to the Velocity endpoint and then changes it back to InOnly afterwards. For more details of the Velocity endpoint, see Velocity in the Apache Camel Component Reference Guide.

Figure 2.4, “Sample Pipeline for InOut Exchanges” shows an example of a processor pipeline for InOut exchanges, which you typically use to support remote procedure call (RPC) semantics. Processors A, B, and C are linked together in the form of a pipeline, with the output of each processor being fed into the input of the next. The final Out message produced by the producer endpoint is sent all the way back to the consumer endpoint, where it provides the reply to the original request.

Figure 2.4. Sample Pipeline for InOut Exchanges

Note that in order to support the InOut exchange pattern, it is essential that the last node in the route (whether it is a producer endpoint or some other kind of processor) creates an Out message. Otherwise, any client that connects to the consumer endpoint would hang and wait indefinitely for a reply message. You should be aware that not all producer endpoints create Out messages.

Consider the following route that processes payment requests, by processing incoming HTTP requests:

from("jetty:http://localhost:8080/foo")
    .to("cxf:bean:addAccountDetails")
    .to("cxf:bean:getCreditRating")
    .to("cxf:bean:processTransaction");

Where the incoming payment request is processed by passing it through a pipeline of Web services, cxf:bean:addAccountDetails, cxf:bean:getCreditRating, and cxf:bean:processTransaction. The final Web service, processTransaction, generates a response (Out message) that is sent back through the JETTY endpoint.

When the pipeline consists of just a sequence of endpoints, it is also possible to use the following alternative syntax:

from("jetty:http://localhost:8080/foo")
    .pipeline("cxf:bean:addAccountDetails", "cxf:bean:getCreditRating", "cxf:bean:processTransaction");

The pipeline for InOptionalOut exchanges is essentially the same as the pipeline in Figure 2.4, “Sample Pipeline for InOut Exchanges”. The difference between InOut and InOptionalOut is that an exchange with the InOptionalOut exchange pattern is allowed to have a null Out message as a reply. That is, in the case of an InOptionalOut exchange, a nullOut message is copied to the In message of the next node in the pipeline. By contrast, in the case of an InOut exchange, a nullOut message is discarded and the original In message from the current node would be copied to the In message of the next node instead.

A standard route takes its input from just a single endpoint, using the from(EndpointURL) syntax in the Java DSL. But what if you need to define multiple inputs for your route? Apache Camel provides several alternatives for specifying multiple inputs to a route. The approach to take depends on whether you want the exchanges to be processed independently of each other or whether you want the exchanges from different inputs to be combined in some way (in which case, you should use the the section called “Content enricher pattern”).

The simplest way to specify multiple inputs is using the multi-argument form of the from() DSL command, for example:

from("URI1", "URI2", "URI3").to("DestinationUri");

Or you can use the following equivalent syntax:

from("URI1").from("URI2").from("URI3").to("DestinationUri");

In both of these examples, exchanges from each of the input endpoints, URI1, URI2, and URI3, are processed independently of each other and in separate threads. In fact, you can think of the preceding route as being equivalent to the following three separate routes:

from("URI1").to("DestinationUri");
from("URI2").to("DestinationUri");
from("URI3").to("DestinationUri");

For example, you might want to merge incoming messages from two different messaging systems and process them using the same route. In most cases, you can deal with multiple inputs by dividing your route into segments, as shown in Figure 2.5, “Processing Multiple Inputs with Segmented Routes”.

Figure 2.5. Processing Multiple Inputs with Segmented Routes

The initial segments of the route take their inputs from some external queues — for example, activemq:Nyse and activemq:Nasdaq — and send the incoming exchanges to an internal endpoint, InternalUrl. The second route segment merges the incoming exchanges, taking them from the internal endpoint and sending them to the destination queue, activemq:USTxn. The InternalUrl is the URL for an endpoint that is intended only for use within a router application. The following types of endpoints are suitable for internal use:

The main purpose of these endpoints is to enable you to glue together different segments of a route. They all provide an effective way of merging multiple inputs into a single route.

The direct component provides the simplest mechanism for linking together routes. The event model for the direct component is synchronous, so that subsequent segments of the route run in the same thread as the first segment. The general format of a direct URL is direct:EndpointID, where the endpoint ID, EndpointID, is simply a unique alphanumeric string that identifies the endpoint instance.

For example, if you want to take the input from two message queues, activemq:Nyse and activemq:Nasdaq, and merge them into a single message queue, activemq:USTxn, you can do this by defining the following set of routes:

from("activemq:Nyse").to("direct:mergeTxns");
from("activemq:Nasdaq").to("direct:mergeTxns");

from("direct:mergeTxns").to("activemq:USTxn");

Where the first two routes take the input from the message queues, Nyse and Nasdaq, and send them to the endpoint, direct:mergeTxns. The last queue combines the inputs from the previous two queues and sends the combined message stream to the activemq:USTxn queue.

The implementation of the direct endpoint behaves as follows: whenever an exchange arrives at a producer endpoint (for example, to("direct:mergeTxns")), the direct endpoint passes the exchange directly to all of the consumers endpoints that have the same endpoint ID (for example, from("direct:mergeTxns")). Direct endpoints can only be used to communicate between routes that belong to the same CamelContext in the same Java virtual machine (JVM) instance.

The SEDA component provides an alternative mechanism for linking together routes. You can use it in a similar way to the direct component, but it has a different underlying event and threading model, as follows:

One of the main advantages of using a SEDA endpoint is that the routes can be more responsive, owing to the built-in consumer thread pool. The stock transactions example can be re-written to use SEDA endpoints instead of direct endpoints, as follows:

from("activemq:Nyse").to("seda:mergeTxns");
from("activemq:Nasdaq").to("seda:mergeTxns");

from("seda:mergeTxns").to("activemq:USTxn");

The main difference between this example and the direct example is that when using SEDA, the second route segment (from seda:mergeTxns to activemq:USTxn) is processed by a pool of five threads.

The VM component is very similar to the SEDA endpoint. The only difference is that, whereas the SEDA component is limited to linking together route segments from within the same CamelContext, the VM component enables you to link together routes from distinct Apache Camel applications, as long as they are running within the same Java virtual machine.

The stock transactions example can be re-written to use VM endpoints instead of SEDA endpoints, as follows:

from("activemq:Nyse").to("vm:mergeTxns");
from("activemq:Nasdaq").to("vm:mergeTxns");

And in a separate router application (running in the same Java VM), you can define the second segment of the route as follows:

from("vm:mergeTxns").to("activemq:USTxn");

The content enricher pattern defines a fundamentally different way of dealing with multiple inputs to a route. When an exchange enters the enricher processor, the enricher contacts an external resource to retrieve information, which is then added to the original message. In this pattern, the external resource effectively represents a second input to the message.

For example, suppose you are writing an application that processes credit requests. Before processing a credit request, you need to augment it with the data that assigns a credit rating to the customer, where the ratings data is stored in a file in the directory, src/data/ratings. You can combine the incoming credit request with data from the ratings file using the pollEnrich() pattern and a GroupedExchangeAggregationStrategy aggregation strategy, as follows:

from("jms:queue:creditRequests")
    .pollEnrich("file:src/data/ratings?noop=true", new GroupedExchangeAggregationStrategy())
    .bean(new MergeCreditRequestAndRatings(), "merge")
    .to("jms:queue:reformattedRequests");

Where the GroupedExchangeAggregationStrategy class is a standard aggregation strategy from the org.apache.camel.processor.aggregate package that adds each new exchange to a java.util.List instance and stores the resulting list in the Exchange.GROUPED_EXCHANGE exchange property. In this case, the list contains two elements: the original exchange (from the creditRequests JMS queue); and the enricher exchange (from the file endpoint).

To access the grouped exchange, you can use code like the following:

public class MergeCreditRequestAndRatings {
    public void merge(Exchange ex) {
        // Obtain the grouped exchange
        List<Exchange> list = ex.getProperty(Exchange.GROUPED_EXCHANGE, List.class);

        // Get the exchanges from the grouped exchange
        Exchange originalEx = list.get(0);
        Exchange ratingsEx  = list.get(1);

        // Merge the exchanges
        ...
    }
}

An alternative approach to this application would be to put the merge code directly into the implementation of the custom aggregation strategy class.

For more details about the content enricher pattern, see Section 10.1, “Content Enricher”.

Bean integration provides a general purpose mechanism for processing messages using arbitrary Java objects. By inserting a bean reference into a route, you can call an arbitrary method on a Java object, which can then access and modify the incoming exchange. The mechanism that maps an exchange’s contents to the parameters and return values of a bean method is known as parameter binding. Parameter binding can use any combination of the following approaches in order to initialize a method’s parameters:

Beans are made accessible through a bean registry, which is a service that enables you to look up beans using either the class name or the bean ID as a key. The way that you create an entry in the bean registry depends on the underlying framework — for example, plain Java, Spring, Guice, or Blueprint. Registry entries are usually created implicitly (for example, when you instantiate a Spring bean in a Spring XML file).

Apache Camel implements a plug-in strategy for the bean registry, defining an integration layer for accessing beans which makes the underlying registry implementation transparent. Hence, it is possible to integrate Apache Camel applications with a variety of different bean registries, as shown in Table 2.2, “Registry Plug-Ins”.

Normally, you do not have to worry about configuring bean registries, because the relevant bean registry is automatically installed for you. For example, if you are using the Spring framework to define your routes, the Spring ApplicationContextRegistry plug-in is automatically installed in the current CamelContext instance.

Deployment in an OSGi container is a special case. When an Apache Camel route is deployed into the OSGi container, the CamelContext automatically sets up a registry chain for resolving bean instances: the registry chain consists of the OSGi registry, followed by the Blueprint (or Spring) registry.

To process exchange objects using a Java bean (which is a plain old Java object or POJO), use the bean() processor, which binds the inbound exchange to a method on the Java object. For example, to process inbound exchanges using the class, MyBeanProcessor, define a route like the following:

from("file:data/inbound")
    .bean(MyBeanProcessor.class, "processBody")
    .to("file:data/outbound");

Where the bean() processor creates an instance of MyBeanProcessor type and invokes the processBody() method to process inbound exchanges. This approach is adequate if you only want to access the MyBeanProcessor instance from a single route. However, if you want to access the same MyBeanProcessor instance from multiple routes, use the variant of bean() that takes the Object type as its first argument. For example:

MyBeanProcessor myBean = new MyBeanProcessor();

from("file:data/inbound")
    .bean(myBean, "processBody")
    .to("file:data/outbound");
from("activemq:inboundData")
    .bean(myBean, "processBody")
    .to("activemq:outboundData");

If a bean defines overloaded methods, you can choose which of the overloaded methods to invoke by specifying the method name along with its parameter types. For example, if the MyBeanBrocessor class has two overloaded methods, processBody(String) and processBody(String,String), you can invoke the latter overloaded method as follows:

from("file:data/inbound")
  .bean(MyBeanProcessor.class, "processBody(String,String)")
  .to("file:data/outbound");

Alternatively, if you want to identify a method by the number of parameters it takes, rather than specifying the type of each parameter explicitly, you can use the wildcard character, \*. For example, to invoke a method named processBody that takes two parameters, irrespective of the exact type of the parameters, invoke the bean() processor as follows:

from("file:data/inbound")
.bean(MyBeanProcessor.class, "processBody(*,*)")
.to("file:data/outbound");

When specifying the method, you can use either a simple unqualified type name—for example, processBody(Exchange)—or a fully qualified type name—for example, processBody(org.apache.camel.Exchange).

You can specify parameter values explicitly, when you call the bean method. The following simple type values can be passed:

The following example shows how you can mix explicit parameter values with type specifiers in the same method invocation:

from("file:data/inbound")
  .bean(MyBeanProcessor.class, "processBody(String, 'Sample string value', true, 7)")
  .to("file:data/outbound");

In the preceding example, the value of the first parameter would presumably be determined by a parameter binding annotation (see the section called “Basic annotations”).

In addition to the simple type values, you can also specify parameter values using the Simple language (Chapter 30, The Simple Language). This means that the full power of the Simple language is available when specifying parameter values. For example, to pass the message body and the value of the title header to a bean method:

from("file:data/inbound")
  .bean(MyBeanProcessor.class, "processBodyAndHeader(${body},${header.title})")
  .to("file:data/outbound");

You can also pass the entire header hash map as a parameter. For example, in the following example, the second method parameter must be declared to be of type java.util.Map:

from("file:data/inbound")
  .bean(MyBeanProcessor.class, "processBodyAndAllHeaders(${body},${header})")
  .to("file:data/outbound");

To bind exchanges to a bean method, you can define a method signature that conforms to certain conventions. In particular, there are two basic conventions for method signatures:

If you want to implement a bean method that accesses or modifies the incoming message body, you must define a method signature that takes a single String argument and returns a String value. For example:

// Java
package com.acme;

public class MyBeanProcessor {
    public String processBody(String body) {
        // Do whatever you like to 'body'...
        return newBody;
    }
}

For greater flexibility, you can implement a bean method that accesses the incoming exchange. This enables you to access or modify all headers, bodies, and exchange properties. For processing exchanges, the method signature takes a single org.apache.camel.Exchange parameter and returns void. For example:

// Java
package com.acme;

public class MyBeanProcessor {
    public void processExchange(Exchange exchange) {
        // Do whatever you like to 'exchange'...
        exchange.getIn().setBody("Here is a new message body!");
    }
}

Instead of creating a bean instance in Java, you can create an instance using Spring XML. In fact, this is the only feasible approach if you are defining your routes in XML. To define a bean in XML, use the standard Spring bean element. The following example shows how to create an instance of MyBeanProcessor:

<beans ...>
    ...
    <bean id="myBeanId" class="com.acme.MyBeanProcessor"/>
</beans>

It is also possible to pass data to the bean’s constructor arguments using Spring syntax. For full details of how to use the Spring bean element, see The IoC Container from the Spring reference guide.

Where the beanRef() processor invokes the MyBeanProcessor.processBody() method on the specified bean instance. You can also invoke the bean from within a Spring XML route, using the Camel schema’s bean element. For example:

<camelContext id="CamelContextID" xmlns="http://camel.apache.org/schema/spring">
  <route>
    <from uri="file:data/inbound"/>
    <bean ref="myBeanId" method="processBody"/>
    <to uri="file:data/outbound"/>
  </route>
</camelContext>

For a slight efficiency gain, you can set the cache option to true, which avoids looking up the registry every time a bean is used. For example, to enable caching, you can set the cache attribute on the bean element as follows:

<bean ref="myBeanId" method="processBody" cache="true"/>

When you create an object instance using the Spring bean element, you can reference it from Java using the bean’s ID (the value of the bean element’s id attribute). For example, given the bean element with ID equal to myBeanId, you can reference the bean in a Java DSL route using the beanRef() processor, as follows:

from("file:data/inbound").beanRef("myBeanId", "processBody").to("file:data/outbound");

Alternatively, you can reference the Spring bean by injection, using the @BeanInject annotation as follows:

// Java
import org.apache.camel.@BeanInject;
...
public class MyRouteBuilder extends RouteBuilder {

   @BeanInject("myBeanId")
   com.acme.MyBeanProcessor bean;

   public void configure() throws Exception {
     ..
   }
}

If you omit the bean ID from the @BeanInject annotation, Camel looks up the registry by type, but this only works if there is just a single bean of the given type. For example, to look up and inject the bean of com.acme.MyBeanProcessor type:

@BeanInject
com.acme.MyBeanProcessor bean;

For the beans used by a Camel context, the correct shutdown order is usually:

If this shutdown order is reversed, then it could happen that the Camel context tries to access a bean that is already destroyed (either leading directly to an error; or the Camel context tries to create the missing bean while it is being destroyed, which also causes an error). The default shutdown order in Spring XML depends on the order in which the beans and the camelContext appear in the Spring XML file. In order to avoid random errors due to incorrect shutdown order, therefore, the camelContext is configured to shut down before any of the other beans in the Spring XML file. This is the default behaviour since Apache Camel 2.13.0.

If you need to change this behaviour (so that the Camel context is not forced to shut down before the other beans), you can set the shutdownEager attribute on the camelContext element to false. In this case, you could potentially exercise more fine-grained control over shutdown order using the Spring depends-on attribute.

The basic parameter bindings described in the section called “Basic method signatures” might not always be convenient to use. For example, if you have a legacy Java class that performs some data manipulation, you might want to extract data from an inbound exchange and map it to the arguments of an existing method signature. For this kind of parameter binding, Apache Camel provides the following kinds of Java annotation:

Table 2.3, “Basic Bean Annotations” shows the annotations from the org.apache.camel Java package that you can use to inject message data into the arguments of a bean method.

For example, the following class shows you how to use basic annotations to inject message data into the processExchange() method arguments.

// Java
import org.apache.camel.*;

public class MyBeanProcessor {
    public void processExchange(
        @Header(name="user") String user,
        @Body String body,
        Exchange exchange
    ) {
        // Do whatever you like to 'exchange'...
        exchange.getIn().setBody(body + "UserName = " + user);
    }
}

Notice how you are able to mix the annotations with the default conventions. As well as injecting the annotated arguments, the parameter binding also automatically injects the exchange object into the org.apache.camel.Exchange argument.

The expression language annotations provide a powerful mechanism for injecting message data into a bean method’s arguments. Using these annotations, you can invoke an arbitrary script, written in the scripting language of your choice, to extract data from an inbound exchange and inject the data into a method argument. Table 2.4, “Expression Language Annotations” shows the annotations from the org.apache.camel.language package (and sub-packages, for the non-core annotations) that you can use to inject message data into the arguments of a bean method.

For example, the following class shows you how to use the @XPath annotation to extract a username and a password from the body of an incoming message in XML format:

// Java
import org.apache.camel.language.*;

public class MyBeanProcessor {
    public void checkCredentials(
        @XPath("/credentials/username/text()") String user,
        @XPath("/credentials/password/text()") String pass
    ) {
        // Check the user/pass credentials...
        ...
    }
}

The @Bean annotation is a special case, because it enables you to inject the result of invoking a registered bean. For example, to inject a correlation ID into a method argument, you can use the @Bean annotation to invoke an ID generator class, as follows:

// Java
import org.apache.camel.language.*;

public class MyBeanProcessor {
    public void processCorrelatedMsg(
        @Bean("myCorrIdGenerator") String corrId,
        @Body String body
    ) {
        // Check the user/pass credentials...
        ...
    }
}

Where the string, myCorrIdGenerator, is the bean ID of the ID generator instance. The ID generator class can be instantiated using the spring bean element, as follows:

<beans ...>
    ...
    <bean id="myCorrIdGenerator" class="com.acme.MyIdGenerator"/>
</beans>

Where the MyIdGenerator class could be defined as follows:

// Java
package com.acme;

public class MyIdGenerator {

    private UserManager userManager;

    public String generate(
        @Header(name = "user") String user,
        @Body String payload
    ) throws Exception {
       User user = userManager.lookupUser(user);
       String userId = user.getPrimaryId();
       String id = userId + generateHashCodeForPayload(payload);
       return id;
   }
}

Notice that you can also use annotations in the referenced bean class, MyIdGenerator. The only restriction on the generate() method signature is that it must return the correct type to inject into the argument annotated by @Bean. Because the @Bean annotation does not let you specify a method name, the injection mechanism simply invokes the first method in the referenced bean that has the matching return type.

Parameter binding annotations can be inherited from an interface or from a superclass. For example, if you define a Java interface with a Header annotation and a Body annotation, as follows:

// Java
import org.apache.camel.*;

public interface MyBeanProcessorIntf {
    void processExchange(
        @Header(name="user") String user,
        @Body String body,
        Exchange exchange
    );
}

The overloaded methods defined in the implementation class, MyBeanProcessor, now inherit the annotations defined in the base interface, as follows:

// Java
import org.apache.camel.*;

public class MyBeanProcessor implements MyBeanProcessorIntf {
    public void processExchange(
        String user,  // Inherits Header annotation
        String body,  // Inherits Body annotation
        Exchange exchange
    ) {
        ...
    }
}

The class that implements a Java interface is often protected, private or in package-only scope. If you try to invoke a method on an implementation class that is restricted in this way, the bean binding falls back to invoking the corresponding interface method, which is publicly accessible.

For example, consider the following public BeanIntf interface:

// Java
public interface BeanIntf {
    void processBodyAndHeader(String body, String title);
}

Where the BeanIntf interface is implemented by the following protected BeanIntfImpl class:

// Java
protected class BeanIntfImpl implements BeanIntf {
    void processBodyAndHeader(String body, String title) {
        ...
    }
}

The following bean invocation would fall back to invoking the public BeanIntf.processBodyAndHeader method:

from("file:data/inbound")
  .bean(BeanIntfImpl.class, "processBodyAndHeader(${body}, ${header.title})")
  .to("file:data/outbound");

Bean integration has the capability to invoke static methods without creating an instance of the associated class. For example, consider the following Java class that defines the static method, changeSomething():

// Java
...
public final class MyStaticClass {
    private MyStaticClass() {
    }

    public static String changeSomething(String s) {
        if ("Hello World".equals(s)) {
            return "Bye World";
        }
        return null;
    }

    public void doSomething() {
        // noop
    }
}

You can use bean integration to invoke the static changeSomething method, as follows:

from("direct:a")
  *.bean(MyStaticClass.class, "changeSomething")*
  .to("mock:a");

Note that, although this syntax looks identical to the invocation of an ordinary function, bean integration exploits Java reflection to identify the method as static and proceeds to invoke the method without instantiating MyStaticClass.

In the special case where a route is deployed into a Red Hat JBoss Fuse container, it is possible to invoke an OSGi service directly using bean integration. For example, assuming that one of the bundles in the OSGi container has exported the service, org.fusesource.example.HelloWorldOsgiService, you could invoke the sayHello method using the following bean integration code:

from("file:data/inbound")
  .bean(org.fusesource.example.HelloWorldOsgiService.class, "sayHello")
  .to("file:data/outbound");

You could also invoke the OSGi service from within a Spring or blueprint XML file, using the bean component, as follows:

<to uri="bean:org.fusesource.example.HelloWorldOsgiService?method=sayHello"/>

The way this works is that Apache Camel sets up a chain of registries when it is deployed in the OSGi container. First of all, it looks up the specified class name in the OSGi service registry; if this lookup fails, it then falls back to the local Spring DM or blueprint registry.

When processing messages with Java code (for example, in a bean class or in a processor class), it is often necessary to create a fresh exchange instance. If you need to create an Exchange object, the easiest approach is to invoke the methods of the ExchangeBuilder class, as described here.

The fully qualified name of the ExchangeBuilder class is as follows:

org.apache.camel.builder.ExchangeBuilder

The ExchangeBuilder exposes the static method, anExchange, which you can use to start building an exchange object.

For example, the following code creates a new exchange object containing the message body string, Hello World!, and with headers containing username and password credentials:

// Java
import org.apache.camel.Exchange;
import org.apache.camel.builder.ExchangeBuilder;
...
Exchange exch = ExchangeBuilder.anExchange(camelCtx)
                    .withBody("Hello World!")
                    .withHeader("username", "jdoe")
                    .withHeader("password", "pass")
                    .build();

The ExchangeBuilder class supports the following methods:

The property placeholders feature can be used to substitute strings into various contexts (such as endpoint URIs and attributes in XML DSL elements), where the placeholder settings are stored in Java properties files. This feature can be useful, if you want to share settings between different Apache Camel applications or if you want to centralize certain configuration settings.

For example, the following route sends requests to a Web server, whose host and port are substituted by the placeholders, {{remote.host}} and {{remote.port}}:

from("direct:start").to("http://{{remote.host}}:{{remote.port}}");

The placeholder values are defined in a Java properties file, as follows:

# Java properties file
remote.host=myserver.com
remote.port=8080

Apache Camel using the PropertyPlaceholders support the following:

Property settings are stored in one or more Java properties files and must conform to the standard Java properties file format. Each property setting appears on its own line, in the format Key=Value. Lines with # or ! as the first non-blank character are treated as comments.

For example, a property file could have content as shown in Example 2.4, “Sample Property File”.

# Property placeholder settings
# (in Java properties file format)
cool.end=mock:result
cool.result=result
cool.concat=mock:{{cool.result}}
cool.start=direct:cool
cool.showid=true

cheese.end=mock:cheese
cheese.quote=Camel rocks
cheese.type=Gouda

bean.foo=foo
bean.bar=bar

The properties component must be configured with the locations of one or more property files before you can start using it in route definitions. You must provide the property values using one of the following resolvers:

For example, to specify the com/fusesource/cheese.properties property file and the com/fusesource/bar.properties property file, both located on the classpath, you would use the following location string:

com/fusesource/cheese.properties,com/fusesource/bar.properties

You can embed Java system properties and O/S environment variables in a location PathName.

Java system properties can be embedded in a location resolver using the syntax, ${PropertyName}. For example, if the root directory of Red Hat JBoss Fuse is stored in the Java system property, karaf.home, you could embed that directory value in a file location, as follows:

file:${karaf.home}/etc/foo.properties

O/S environment variables can be embedded in a location resolver using the syntax, ${env:VarName}. For example, if the root directory of JBoss Fuse is stored in the environment variable, SMX_HOME, you could embed that directory value in a file location, as follows:

file:${env:SMX_HOME}/etc/foo.properties

Before you can start using property placeholders, you must configure the properties component, specifying the locations of one or more property files.

In the Java DSL, you can configure the properties component with the property file locations, as follows:

// Java
import org.apache.camel.component.properties.PropertiesComponent;
...
PropertiesComponent pc = new PropertiesComponent();
pc.setLocation("com/fusesource/cheese.properties,com/fusesource/bar.properties");
context.addComponent("properties", pc);

As shown in the addComponent() call, the name of the properties component must be set to properties.

In the XML DSL, you can configure the properties component using the dedicated propertyPlacholder element, as follows:

<camelContext ...>
   <propertyPlaceholder
      id="properties"
      location="com/fusesource/cheese.properties,com/fusesource/bar.properties"
   />
</camelContext>

If you want the properties component to ignore any missing .properties files when it is being initialized, you can set the ignoreMissingLocation option to true (normally, a missing .properties file would result in an error being raised).

Additionally, if you want the properties component to ignore any missing locations that are specified using Java system properties or O/S environment variables, you can set the ignoreMissingLocation option to true.

After it is configured, the property component automatically substitutes placeholders (in the appropriate contexts). The syntax of a placeholder depends on the context, as follows:

Wherever an endpoint URI string appears in a route, the first step in parsing the endpoint URI is to apply the property placeholder parser. The placeholder parser automatically substitutes any property names appearing between double braces, {{Key}}. For example, given the property settings shown in Example 2.4, “Sample Property File”, you could define a route as follows:

from("{{cool.start}}")
    .to("log:{{cool.start}}?showBodyType=false&showExchangeId={{cool.showid}}")
    .to("mock:{{cool.result}}");

By default, the placeholder parser looks up the properties bean ID in the registry to find the property component. If you prefer, you can explicitly specify the scheme in the endpoint URIs. For example, by prefixing properties: to each of the endpoint URIs, you can define the following equivalent route:

from("properties:{{cool.start}}")
    .to("properties:log:{{cool.start}}?showBodyType=false&showExchangeId={{cool.showid}}")
    .to("properties:mock:{{cool.result}}");

When specifying the scheme explicitly, you also have the option of specifying options to the properties component. For example, to override the property file location, you could set the location option as follows:

from("direct:start").to("properties:{{bar.end}}?location=com/mycompany/bar.properties");

You can also use property placeholders in the XML DSL, for setting various attributes of the DSL elements. In this context, the placholder syntax also uses double braces, {{Key}}. For example, you could define a jmxAgent element using property placeholders, as follows:

<camelContext id="camel" xmlns="http://camel.apache.org/schema/spring">
    <propertyPlaceholder id="properties" location="org/apache/camel/spring/jmx.properties"/>

    <!-- we can use property placeholders when we define the JMX agent -->
    <jmxAgent id="agent" registryPort="{{myjmx.port}}"
              usePlatformMBeanServer="{{myjmx.usePlatform}}"
              createConnector="true"
              statisticsLevel="RoutesOnly"
            />

    <route>
        <from uri="seda:start"/>
        <to uri="mock:result"/>
    </route>
</camelContext>

You can use the regular placeholder syntax for specifying attribute values of xs:string type — for example, <jmxAgent registryPort="{{myjmx.port}}" …​>. But for attributes of any other type (for example, xs:int or xs:boolean), you must use the special syntax, prop:AttributeName="Key".

For example, given that a property file defines the stop.flag property to have the value, true, you can use this property to set the stopOnException boolean attribute, as follows:

<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:prop="http://camel.apache.org/schema/placeholder"
       ... >

    <bean id="illegal" class="java.lang.IllegalArgumentException">
        <constructor-arg index="0" value="Good grief!"/>
    </bean>

    <camelContext xmlns="http://camel.apache.org/schema/spring">

        <propertyPlaceholder id="properties"
                             location="classpath:org/apache/camel/component/properties/myprop.properties"
                             xmlns="http://camel.apache.org/schema/spring"/>

        <route>
            <from uri="direct:start"/>
            <multicast prop:stopOnException="stop.flag">
                <to uri="mock:a"/>
                <throwException ref="damn"/>
                <to uri="mock:b"/>
            </multicast>
        </route>

    </camelContext>

</beans>

When invoking an EIP command in the Java DSL, you can set any EIP option using the value of a property placeholder, by adding a sub-clause of the form, placeholder("OptionName", "Key").

For example, given that a property file defines the stop.flag property to have the value, true, you can use this property to set the stopOnException option of the multicast EIP, as follows:

from("direct:start")
    .multicast().placeholder("stopOnException", "stop.flag")
        .to("mock:a").throwException(new IllegalAccessException("Damn")).to("mock:b");

You can also substitute property placeholders in Simple language expressions, but in this case the syntax of the placeholder is ${properties:Key}. For example, you can substitute the cheese.quote placeholder inside a Simple expression, as follows:

from("direct:start")
    .transform().simple("Hi ${body} do you think ${properties:cheese.quote}?");

You can specify a default value for the property, using the syntax, ${properties:Key:DefaultVal}. For example:

from("direct:start")
    .transform().simple("Hi ${body} do you think ${properties:cheese.quote:cheese is good}?");

It is also possible to override the location of the property file using the syntax, ${properties-location:Location:Key}. For example, to substitute the bar.quote placeholder using the settings from the com/mycompany/bar.properties property file, you can define a Simple expression as follows:

from("direct:start")
    .transform().simple("Hi ${body}. ${properties-location:com/mycompany/bar.properties:bar.quote}.");

In older releases, the xs:string type attributes were used to support placeholders in the XML DSL. For example, the timeout attribute would be a xs:int type. Therefore, you cannot set a string value as the placeholder key.

From Apache Camel 2.7 onwards, this is now possible by using a special placeholder namespace. The following example illustrates the prop prefix for the namespace. It enables you to use the prop prefix in the attributes in the XML DSLs.

stop=true

<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:prop="http://camel.apache.org/schema/placeholder"
       xsi:schemaLocation="
       http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
       http://camel.apache.org/schema/spring http://camel.apache.org/schema/spring/camel-spring.xsd
    ">

    <!-- Notice in the declaration above, we have defined the prop prefix as the Camel placeholder namespace -->

    <bean id="damn" class="java.lang.IllegalArgumentException">
        <constructor-arg index="0" value="Damn"/>
    </bean>

    <camelContext xmlns="http://camel.apache.org/schema/spring">

        <propertyPlaceholder id="properties"
                             location="classpath:org/apache/camel/component/properties/myprop.properties"
                             xmlns="http://camel.apache.org/schema/spring"/>

        <route>
            <from uri="direct:start"/>
            <!-- use prop namespace, to define a property placeholder, which maps to
                 option stopOnException={{stop}} -->
            <multicast prop:stopOnException="stop">
                <to uri="mock:a"/>
                <throwException ref="damn"/>
                <to uri="mock:b"/>
            </multicast>
        </route>

    </camelContext>

</beans>

If you deploy your route into the Red Hat JBoss Fuse OSGi container, you can integrate the Apache Camel property placeholder mechanism with JBoss Fuse’s blueprint property placeholder mechanism (in fact, the integration is enabled by default). There are two basic approaches to setting up the integration, as follows:

If you define a camelContext element inside an OSGi blueprint file, the Apache Camel property placeholder mechanism automatically integrates with the blueprint property placeholder mechanism. That is, placeholders obeying the Apache Camel syntax (for example, {{cool.end}}) that appear within the scope of camelContext are implicitly resolved by looking up the blueprint property placeholder mechanism.

For example, consider the following route defined in an OSGi blueprint file, where the last endpoint in the route is defined by the property placeholder, {{result}}:

<blueprint xmlns="http://www.osgi.org/xmlns/blueprint/v1.0.0"
           xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
           xmlns:cm="http://aries.apache.org/blueprint/xmlns/blueprint-cm/v1.0.0"
           xsi:schemaLocation="
           http://www.osgi.org/xmlns/blueprint/v1.0.0 https://www.osgi.org/xmlns/blueprint/v1.0.0/blueprint.xsd">

    <!-- OSGI blueprint property placeholder -->
    <cm:property-placeholder id="myblueprint.placeholder" persistent-id="camel.blueprint">
        <!-- list some properties for this test -->
        <cm:default-properties>
            <cm:property name="result" value="mock:result"/>
        </cm:default-properties>
    </cm:property-placeholder>

    <camelContext xmlns="http://camel.apache.org/schema/blueprint">
        <!-- in the route we can use {{ }} placeholders which will look up in blueprint,
             as Camel will auto detect the OSGi blueprint property placeholder and use it -->
        <route>
            <from uri="direct:start"/>
            <to uri="mock:foo"/>
            <to uri="{{result}}"/>
        </route>
    </camelContext>

</blueprint>

The blueprint property placeholder mechanism is initialized by creating a cm:property-placeholder bean. In the preceding example, the cm:property-placeholder bean is associated with the camel.blueprint persistent ID, where a persistent ID is the standard way of referencing a group of related properties from the OSGi Configuration Admin service. In other words, the cm:property-placeholder bean provides access to all of the properties defined under the camel.blueprint persistent ID. It is also possible to specify default values for some of the properties (using the nested cm:property elements).

In the context of blueprint, the Apache Camel placeholder mechanism searches for an instance of cm:property-placeholder in the bean registry. If it finds such an instance, it automatically integrates the Apache Camel placeholder mechanism, so that placeholders like, {{result}}, are resolved by looking up the key in the blueprint property placeholder mechanism (in this example, through the myblueprint.placeholder bean).

If you want to have more control over where the Apache Camel property placeholder mechanism finds its properties, you can define a propertyPlaceholder element and specify the resolver locations explicitly.

For example, consider the following blueprint configuration, which differs from the previous example in that it creates an explicit propertyPlaceholder instance:

<blueprint xmlns="http://www.osgi.org/xmlns/blueprint/v1.0.0"
           xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
           xmlns:cm="http://aries.apache.org/blueprint/xmlns/blueprint-cm/v1.0.0"
           xsi:schemaLocation="
           http://www.osgi.org/xmlns/blueprint/v1.0.0 ">https://www.osgi.org/xmlns/blueprint/v1.0.0/blueprint.xsd">

    <!-- OSGI blueprint property placeholder -->
    <cm:property-placeholder id="myblueprint.placeholder" persistent-id="camel.blueprint">
        <!-- list some properties for this test -->
        <cm:default-properties>
            <cm:property name="result" value="mock:result"/>
        </cm:default-properties>
    </cm:property-placeholder>

    <camelContext xmlns="http://camel.apache.org/schema/blueprint">

        <!-- using Camel properties component and refer to the blueprint property placeholder by its id -->
        <propertyPlaceholder id="properties" location="blueprint:myblueprint.placeholder"/>

        <!-- in the route we can use {{ }} placeholders which will lookup in blueprint -->
        <route>
            <from uri="direct:start"/>
            <to uri="mock:foo"/>
            <to uri="{{result}}"/>
        </route>

    </camelContext>

</blueprint>

In the preceding example, the propertyPlaceholder element specifies explicitly which cm:property-placeholder bean to use by setting the location to blueprint:myblueprint.placeholder. That is, the blueprint: resolver explicitly references the ID, myblueprint.placeholder, of the cm:property-placeholder bean.

This style of configuration is useful, if there is more than one cm:property-placeholder bean defined in the blueprint file and you need to specify which one to use. It also makes it possible to source properties from multiple locations, by specifying a comma-separated list of locations. For example, if you wanted to look up properties both from the cm:property-placeholder bean and from the properties file, myproperties.properties, on the classpath, you could define the propertyPlaceholder element as follows:

<propertyPlaceholder id="properties"
  location="blueprint:myblueprint.placeholder,classpath:myproperties.properties"/>

If you define your Apache Camel application using XML DSL in a Spring XML file, you can integrate the Apache Camel property placeholder mechanism with Spring property placeholder mechanism by declaring a Spring bean of type, org.apache.camel.spring.spi.BridgePropertyPlaceholderConfigurer.

Define a BridgePropertyPlaceholderConfigurer, which replaces both Apache Camel’s propertyPlaceholder element and Spring’s ctx:property-placeholder element in the Spring XML file. You can then refer to the configured properties using either the Spring ${PropName} syntax or the Apache Camel {{PropName}} syntax.

For example, defining a bridge property placeholder that reads its property settings from the cheese.properties file:

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xmlns:ctx="http://www.springframework.org/schema/context"
    xsi:schemaLocation="
        http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
        http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd">

  <!-- Bridge Spring property placeholder with Camel -->
  <!-- Do not use <ctx:property-placeholder ... > at the same time -->
  <bean id="bridgePropertyPlaceholder"
        class="org.apache.camel.spring.spi.BridgePropertyPlaceholderConfigurer">
    <property name="location"
              value="classpath:org/apache/camel/component/properties/cheese.properties"/>
  </bean>

  <!-- A bean that uses Spring property placeholder -->
  <!-- The ${hi} is a spring property placeholder -->
  <bean id="hello" class="org.apache.camel.component.properties.HelloBean">
    <property name="greeting" value="${hi}"/>
  </bean>

  <camelContext xmlns="http://camel.apache.org/schema/spring">
    <!-- Use Camel's property placeholder {{ }} style -->
    <route>
      <from uri="direct:{{cool.bar}}"/>
      <bean ref="hello"/>
      <to uri="{{cool.end}}"/>
    </route>
  </camelContext>

</beans>

The Apache Camel threading model is based on the powerful Java concurrency API, Package java.util.concurrent, that first became available in Sun’s JDK 1.5. The key interface in this API is the ExecutorService interface, which represents a thread pool. Using the concurrency API, you can create many different kinds of thread pool, covering a wide range of scenarios.

The Apache Camel thread pool API builds on the Java concurrency API by providing a central factory (of org.apache.camel.spi.ExecutorServiceManager type) for all of the thread pools in your Apache Camel application. Centralising the creation of thread pools in this way provides several advantages, including:

Some Apache Camel components — such as SEDA, JMS, and Jetty — are inherently multi-threaded. These components have all been implemented using the Apache Camel threading model and thread pool API.

If you are planning to implement your own Apache Camel component, it is recommended that you integrate your threading code with the Apache Camel threading model. For example, if your component needs a thread pool, it is recommended that you create it using the CamelContext’s ExecutorServiceManager object.

Some of the standard processors in Apache Camel create their own thread pool by default. These threading-aware processors are also integrated with the Apache Camel threading model and they provide various options that enable you to customize the thread pools that they use.

Table 2.8, “Processor Threading Options” shows the various options for controlling and setting thread pools on the threading-aware processors built-in to Apache Camel.

parallelProcessing()
executorService()
executorServiceRef()

@parallelProcessing
@executorServiceRef

parallelProcessing()
executorService()
executorServiceRef()

@parallelProcessing
@executorServiceRef

parallelProcessing()
executorService()
executorServiceRef()

@parallelProcessing
@executorServiceRef

parallelProcessing()
executorService()
executorServiceRef()

@parallelProcessing
@executorServiceRef

executorService()
executorServiceRef()
poolSize()
maxPoolSize()
keepAliveTime()
timeUnit()
maxQueueSize()
rejectedPolicy()

@executorServiceRef
@poolSize
@maxPoolSize
@keepAliveTime
@timeUnit
@maxQueueSize
@rejectedPolicy

wireTap(String uri, ExecutorService executorService)
wireTap(String uri, String executorServiceRef)

@executorServiceRef

The threads processor is a general-purpose DSL command, which you can use to introduce a thread pool into a route. It supports the following options to customize the thread pool:

To create a default thread pool for one of the threading-aware processors, enable the parallelProcessing option, using the parallelProcessing() sub-clause, in the Java DSL, or the parallelProcessing attribute, in the XML DSL.

For example, in the Java DSL, you can invoke the multicast processor with a default thread pool (where the thread pool is used to process the multicast destinations concurrently) as follows:

from("direct:start")
  .multicast().parallelProcessing()
    .to("mock:first")
    .to("mock:second")
    .to("mock:third");

You can define the same route in XML DSL as follows

<camelContext id="camel" xmlns="http://camel.apache.org/schema/spring">
    <route>
        <from uri="direct:start"/>
        <multicast parallelProcessing="true">
            <to uri="mock:first"/>
            <to uri="mock:second"/>
            <to uri="mock:third"/>
        </multicast>
    </route>
</camelContext>

The default thread pools are automatically created by a thread factory that takes its settings from the default thread pool profile. The default thread pool profile has the settings shown in Table 2.9, “Default Thread Pool Profile Settings” (assuming that these settings have not been modified by the application code).

It is possible to change the default thread pool profile settings, so that all subsequent default thread pools will be created with the custom settings. You can change the profile either in Java or in Spring XML.

For example, in the Java DSL, you can customize the poolSize option and the maxQueueSize option in the default thread pool profile, as follows:

// Java
import org.apache.camel.spi.ExecutorServiceManager;
import org.apache.camel.spi.ThreadPoolProfile;
...
ExecutorServiceManager manager = context.getExecutorServiceManager();
ThreadPoolProfile defaultProfile = manager.getDefaultThreadPoolProfile();

// Now, customize the profile settings.
defaultProfile.setPoolSize(3);
defaultProfile.setMaxQueueSize(100);
...

In the XML DSL, you can customize the default thread pool profile, as follows:

<camelContext id="camel" xmlns="http://camel.apache.org/schema/spring">
    <threadPoolProfile
        id="changedProfile"
        defaultProfile="true"
        poolSize="3"
        maxQueueSize="100"/>
    ...
</camelContext>

Note that it is essential to set the defaultProfile attribute to true in the preceding XML DSL example, otherwise the thread pool profile would be treated like a custom thread pool profile (see the section called “Creating a custom thread pool profile”), instead of replacing the default thread pool profile.

It is also possible to specify the thread pool for a threading-aware processor more directly, using either the executorService or executorServiceRef options (where these options are used instead of the parallelProcessing option). There are two approaches you can use to customize a processor’s thread pool, as follows:

When you pass a bean ID to the executorServiceRef option, the threading-aware processor first tries to find a custom thread pool with that ID in the registry. If no thread pool is registered with that ID, the processor then attempts to look up a custom thread pool profile in the registry and uses the custom thread pool profile to instantiate a custom thread pool.

A custom thread pool can be any thread pool of java.util.concurrent.ExecutorService type. The following approaches to creating a thread pool instance are recommended in Apache Camel:

Ultimately, there is not much difference between the two approaches, because the ThreadPoolBuilder is actually defined using the ExecutorServiceManager instance. Normally, the ThreadPoolBuilder is preferred, because it offers a simpler approach. But there is at least one kind of thread (the ScheduledExecutorService) that can only be created by accessing the ExecutorServiceManager instance directory.

Table 2.10, “Thread Pool Builder Options” shows the options supported by the ThreadPoolBuilder class, which you can set when defining a new custom thread pool.

In Java DSL, you can define a custom thread pool using the ThreadPoolBuilder, as follows:

// Java
import org.apache.camel.builder.ThreadPoolBuilder;
import java.util.concurrent.ExecutorService;
...
ThreadPoolBuilder poolBuilder = new ThreadPoolBuilder(context);
ExecutorService customPool = poolBuilder.poolSize(5).maxPoolSize(5).maxQueueSize(100).build("customPool");
...

from("direct:start")
  .multicast().executorService(customPool)
    .to("mock:first")
    .to("mock:second")
    .to("mock:third");

Instead of passing the object reference, customPool, directly to the executorService() option, you can look up the thread pool in the registry, by passing its bean ID to the executorServiceRef() option, as follows:

// Java
from("direct:start")
  .multicast().executorServiceRef("customPool")
    .to("mock:first")
    .to("mock:second")
    .to("mock:third");

In XML DSL, you access the ThreadPoolBuilder using the threadPool element. You can then reference the custom thread pool using the executorServiceRef attribute to look up the thread pool by ID in the Spring registry, as follows:

<camelContext id="camel" xmlns="http://camel.apache.org/schema/spring">
    <threadPool id="customPool"
                poolSize="5"
                maxPoolSize="5"
                maxQueueSize="100" />

    <route>
        <from uri="direct:start"/>
        <multicast executorServiceRef="customPool">
            <to uri="mock:first"/>
            <to uri="mock:second"/>
            <to uri="mock:third"/>
        </multicast>
    </route>
</camelContext>

If you have many custom thread pool instances to create, you might find it more convenient to define a custom thread pool profile, which acts as a factory for thread pools. Whenever you reference a thread pool profile from a threading-aware processor, the processor automatically uses the profile to create a new thread pool instance. You can define a custom thread pool profile either in Java DSL or in XML DSL.

For example, in Java DSL you can create a custom thread pool profile with the bean ID, customProfile, and reference it from within a route, as follows:

// Java
import org.apache.camel.spi.ThreadPoolProfile;
import org.apache.camel.impl.ThreadPoolProfileSupport;
...
// Create the custom thread pool profile
ThreadPoolProfile customProfile = new ThreadPoolProfileSupport("customProfile");
customProfile.setPoolSize(5);
customProfile.setMaxPoolSize(5);
customProfile.setMaxQueueSize(100);
context.getExecutorServiceManager().registerThreadPoolProfile(customProfile);
...
// Reference the custom thread pool profile in a route
from("direct:start")
  .multicast().executorServiceRef("customProfile")
    .to("mock:first")
    .to("mock:second")
    .to("mock:third");

In XML DSL, use the threadPoolProfile element to create a custom pool profile (where you let the defaultProfile option default to false, because this is not a default thread pool profile). You can create a custom thread pool profile with the bean ID, customProfile, and reference it from within a route, as follows:

<camelContext id="camel" xmlns="http://camel.apache.org/schema/spring">
    <threadPoolProfile
                id="customProfile"
                poolSize="5"
                maxPoolSize="5"
                maxQueueSize="100" />

    <route>
        <from uri="direct:start"/>
        <multicast executorServiceRef="customProfile">
            <to uri="mock:first"/>
            <to uri="mock:second"/>
            <to uri="mock:third"/>
        </multicast>
    </route>
</camelContext>

Some of the standard poll-based components — such as File and FTP — allow you to specify the thread pool to use. This makes it possible for different components to share the same thread pool, reducing the overall number of threads in the JVM.

For example, the see File2 in the Apache Camel Component Reference Guide. and the Ftp2 in the Apache Camel Component Reference Guide both expose the scheduledExecutorService property, which you can use to specify the component’s ExecutorService object.

To make the application logs more readable, it is often a good idea to customize the thread names (which are used to identify threads in the log). To customize thread names, you can configure the thread name pattern by calling the setThreadNamePattern method on the ExecutorServiceStrategy class or the ExecutorServiceManager class. Alternatively, an easier way to set the thread name pattern is to set the threadNamePattern property on the CamelContext object.

The following placeholders can be used in a thread name pattern:

The following is a typical example of a thread name pattern:

Camel (#camelId#) thread #counter# - #name#

The following example shows how to set the threadNamePattern attribute on a Camel context using XML DSL:

<camelContext xmlns="http://camel.apache.org/schema/spring"
              threadNamePattern="Riding the thread #counter#" >
  <route>
    <from uri="seda:start"/>
    <to uri="log:result"/>
    <to uri="mock:result"/>
  </route>
</camelContext>

By default, routes are automatically started when your Apache Camel application (as represented by the CamelContext instance) starts up and routes are automatically shut down when your Apache Camel application shuts down. For non-critical deployments, the details of the shutdown sequence are usually not very important. But in a production environment, it is often crucial that existing tasks should run to completion during shutdown, in order to avoid data loss. You typically also want to control the order in which routes shut down, so that dependencies are not violated (which would prevent existing tasks from running to completion).

For this reason, Apache Camel provides a set of features to support graceful shutdown of applications. Graceful shutdown gives you full control over the stopping and starting of routes, enabling you to control the shutdown order of routes and enabling current tasks to run to completion.

It is good practice to assign a route ID to each of your routes. As well as making logging messages and management features more informative, the use of route IDs enables you to apply greater control over the stopping and starting of routes.

For example, in the Java DSL, you can assign the route ID, myCustomerRouteId, to a route by invoking the routeId() command as follows:

from("SourceURI").routeId("myCustomRouteId").process(...).to(TargetURI);

In the XML DSL, set the route element’s id attribute, as follows:

<camelContext id="CamelContextID" xmlns="http://camel.apache.org/schema/spring">
  <route id="myCustomRouteId" >
    <from uri="SourceURI"/>
    <process ref="someProcessorId"/>
    <to uri="TargetURI"/>
  </route>
</camelContext>

By default, all of the routes that the CamelContext knows about at start time will be started automatically. If you want to control the start-up of a particular route manually, however, you might prefer to disable automatic start-up for that route.

To control whether a Java DSL route starts up automatically, invoke the autoStartup command, either with a boolean argument (true or false) or a String argument (true or false). For example, you can disable automatic start-up of a route in the Java DSL, as follows:

from("SourceURI")
  .routeId("nonAuto")
  .autoStartup(false)
  .to(TargetURI);

You can disable automatic start-up of a route in the XML DSL by setting the autoStartup attribute to false on the route element, as follows:

<camelContext id="CamelContextID" xmlns="http://camel.apache.org/schema/spring">
  <route id="nonAuto" autoStartup="false">
    <from uri="SourceURI"/>
    <to uri="TargetURI"/>
  </route>
</camelContext>

You can manually start or stop a route at any time in Java by invoking the startRoute() and stopRoute() methods on the CamelContext instance. For example, to start the route having the route ID, nonAuto, invoke the startRoute() method on the CamelContext instance, context, as follows:

// Java
context.startRoute("nonAuto");

To stop the route having the route ID, nonAuto, invoke the stopRoute() method on the CamelContext instance, context, as follows:

// Java
context.stopRoute("nonAuto");

By default, Apache Camel starts up routes in a non-deterministic order. In some applications, however, it can be important to control the startup order. To control the startup order in the Java DSL, use the startupOrder() command, which takes a positive integer value as its argument. The route with the lowest integer value starts first, followed by the routes with successively higher startup order values.

For example, the first two routes in the following example are linked together through the seda:buffer endpoint. You can ensure that the first route segment starts after the second route segment by assigning startup orders (2 and 1 respectively), as follows:

from("jetty:http://fooserver:8080")
    .routeId("first")
    .startupOrder(2)
    .to("seda:buffer");

from("seda:buffer")
    .routeId("second")
    .startupOrder(1)
    .to("mock:result");

// This route's startup order is unspecified
from("jms:queue:foo").to("jms:queue:bar");

Or in Spring XML, you can achieve the same effect by setting the route element’s startupOrder attribute, as follows:

<route id="first" startupOrder="2">
    <from uri="jetty:http://fooserver:8080"/>
    <to uri="seda:buffer"/>
</route>

<route id="second" startupOrder="1">
    <from uri="seda:buffer"/>
    <to uri="mock:result"/>
</route>

<!-- This route's startup order is unspecified -->
<route>
    <from uri="jms:queue:foo"/>
    <to uri="jms:queue:bar"/>
</route>

Each route must be assigned a unique startup order value. You can choose any positive integer value that is less than 1000. Values of 1000 and over are reserved for Apache Camel, which automatically assigns these values to routes without an explicit startup value. For example, the last route in the preceding example would automatically be assigned the startup value, 1000 (so it starts up after the first two routes).

When a CamelContext instance is shutting down, Apache Camel controls the shutdown sequence using a pluggable shutdown strategy. The default shutdown strategy implements the following shutdown sequence:

Routes are shut down in the reverse of the start-up order. That is, when a start-up order is defined using the startupOrder() command (in Java DSL) or startupOrder attribute (in XML DSL), the first route to shut down is the route with the highest integer value assigned by the start-up order and the last route to shut down is the route with the lowest integer value assigned by the start-up order.

For example, in Example 2.5, “Startup Order in Java DSL”, the first route segment to be shut down is the route with the ID, first, and the second route segment to be shut down is the route with the ID, second. This example illustrates a general rule, which you should observe when shutting down routes: the routes that expose externally-accessible consumer endpoints should be shut down first, because this helps to throttle the flow of messages through the rest of the route graph.

If a route is still processing messages when the shutdown starts, the shutdown strategy normally waits until the currently active exchange has finished processing before shutting down the route. This behavior can be configured on each route using the shutdownRunningTask option, which can take either of the following values:

For example, to shut down a File consumer endpoint gracefully, you should specify the CompleteAllTasks option, as shown in the following Java DSL fragment:

// Java
public void configure() throws Exception {
    from("file:target/pending")
        .routeId("first").startupOrder(2)
        .shutdownRunningTask(ShutdownRunningTask.CompleteAllTasks)
        .delay(1000).to("seda:foo");

    from("seda:foo")
        .routeId("second").startupOrder(1)
        .to("mock:bar");
}

The same route can be defined in the XML DSL as follows:

<camelContext id="camel" xmlns="http://camel.apache.org/schema/spring">
    <!-- let this route complete all its pending messages when asked to shut down -->
    <route id="first"
           startupOrder="2"
           shutdownRunningTask="CompleteAllTasks">
        <from uri="file:target/pending"/>
        <delay><constant>1000</constant></delay>
        <to uri="seda:foo"/>
    </route>

    <route id="second" startupOrder="1">
        <from uri="seda:foo"/>
        <to uri="mock:bar"/>
    </route>
</camelContext>

The shutdown timeout has a default value of 300 seconds. You can change the value of the timeout by invoking the setTimeout() method on the shutdown strategy. For example, you can change the timeout value to 600 seconds, as follows:

// Java
// context = CamelContext instance
context.getShutdownStrategy().setTimeout(600);

If you are implementing a custom Apache Camel component (which also inherits from the org.apache.camel.Service interface), you can ensure that your custom code receives a shutdown notification by implementing the org.apache.camel.spi.ShutdownPrepared interface. This gives the component an opportunity execute custom code in preparation for shutdown.

The OnCompletion DSL name is used to define an action that is to take place when a Unit of Work is completed. A Unit of Work is a Camel concept that encompasses an entire exchange. See Section 34.1, “Exchanges”. The onCompletion command has the following features:

When an onCompletion DSL is specified on an exchange, Camel spins off a new thread. This allows the original thread to continue without interference from the onCompletion task. A route will only support one onCompletion. In the following example, the onCompletion is triggered whether the exchange completes with success or failure. This is the default action.

from("direct:start")
     .onCompletion()
         // This route is invoked when the original route is complete.
         // This is similar to a completion callback.
         .to("log:sync")
         .to("mock:sync")
     // Must use end to denote the end of the onCompletion route.
     .end()
     // here the original route contiues
     .process(new MyProcessor())
     .to("mock:result");

For XML the format is as follows:

<route>
    <from uri="direct:start"/>
    <!-- This onCompletion block is executed when the exchange is done being routed. -->
    <!-- This callback is always triggered even if the exchange fails. -->
    <onCompletion>
        <!-- This is similar to an after completion callback. -->
        <to uri="log:sync"/>
        <to uri="mock:sync"/>
    </onCompletion>
    <process ref="myProcessor"/>
    <to uri="mock:result"/>
</route>

To trigger the onCompletion on failure, the onFailureOnly parameter can be used. Similarly, to trigger the onCompletion on success, use the onCompleteOnly parameter.

from("direct:start")
     // Here onCompletion is qualified to invoke only when the exchange fails (exception or FAULT body).
     .onCompletion().onFailureOnly()
         .to("log:sync")
         .to("mock:sync")
     // Must use end to denote the end of the onCompletion route.
     .end()
     // here the original route continues
     .process(new MyProcessor())
     .to("mock:result");

For XML, onFailureOnly and onCompleteOnly are expressed as booleans on the onCompletion tag:

<route>
    <from uri="direct:start"/>
    <!-- this onCompletion block will only be executed when the exchange is done being routed -->
    <!-- this callback is only triggered when the exchange failed, as we have onFailure=true -->
    <onCompletion onFailureOnly="true">
        <to uri="log:sync"/>
        <to uri="mock:sync"/>
    </onCompletion>
    <process ref="myProcessor"/>
    <to uri="mock:result"/>
</route>

To define onCompletion for more than just one route:

// define a global on completion that is invoked when the exchange is complete
 onCompletion().to("log:global").to("mock:sync");

 from("direct:start")
     .process(new MyProcessor())
     .to("mock:result");

To trigger the onCompletion under certain circumstances, use the onWhen predicate. The following example will trigger the onCompletion when the body of the message contains the word Hello:

/from("direct:start")
     .onCompletion().onWhen(body().contains("Hello"))
         // this route is only invoked when the original route is complete as a kind
         // of completion callback. And also only if the onWhen predicate is true
         .to("log:sync")
         .to("mock:sync")
     // must use end to denote the end of the onCompletion route
     .end()
     // here the original route contiues
     .to("log:original")
     .to("mock:result");

As of Camel 2.14, onCompletion will not use a thread pool by default. To force the use of a thread pool, either set an executorService or set parallelProcessing to true. For example, in Java DSL, use the following format:

onCompletion().parallelProcessing()
     .to("mock:before")
     .delay(1000)
     .setBody(simple("OnComplete:${body}"));

For XML the format is:

<onCompletion parallelProcessing="true">
   <to uri="before"/>
   <delay><constant>1000</constant></delay>
   <setBody><simple>OnComplete:${body}<simple></setBody>
 </onCompletion>

Use the executorServiceRef option to refer to a specific thread pool:

<onCompletion executorServiceRef="myThreadPool"
   <to uri="before"/>
   <delay><constant>1000</constant></delay>
   <setBody><simple>OnComplete:${body}</simple></setBody>
 </onCompletion>>

onCompletion can be run in two modes:

For example, to add a created by header to the response, use modeBeforeConsumer() as shown below:

.onCompletion().modeBeforeConsumer()
     .setHeader("createdBy", constant("Someone"))
 .end()

For XML, set the mode attribute to BeforeConsumer:

<onCompletion mode="BeforeConsumer">
   <setHeader headerName="createdBy">
     <constant>Someone</constant>
   </setHeader>
 </onCompletion>

Available as of Camel 2.14

While Camel provides a lot of existing metrics integration with Codahale metrics has been added for Camel routes. This allows end users to seamless feed Camel routing information together with existing data they are gathering using Codahale metrics.

To use the Codahale metrics you will need to:

Note that performance metrics are only usable if you have a way of displaying them; any kind of monitoring tooling which can integrate with JMX can be used, as the metrics are available over JMX. In addition, the actual data is 100% Codehale JSON.

Obtaining Codahale metrics for a single route can be accomplished by defining a MetricsRoutePolicy on a per route basis.

From Java create an instance of MetricsRoutePolicy to be assigned as the route’s policy. This is shown below:

from("file:src/data?noop=true").routePolicy(new MetricsRoutePolicy()).to("jms:incomingOrders");

From XML DSL you define a <bean> which is specified as the route’s policy; for example:

<bean id="policy" class="org.apache.camel.component.metrics.routepolicy.MetricsRoutePolicy"/>

<camelContext xmlns="http://camel.apache.org/schema/spring">
  <route routePolicyRef="policy">
    <from uri="file:src/data?noop=true"/>
[...]

This factory allows one to add a RoutePolicy for each route which exposes route utilization statistics using Codahale metrics. This factory can be used in Java and XML as the examples below demonstrate.

From Java you just add the factory to the CamelContext as shown below:

context.addRoutePolicyFactory(new MetricsRoutePolicyFactory());

And from XML DSL you define a <bean> as follows:

<!-- use camel-metrics route policy to gather metrics for all routes -->
<bean id="metricsRoutePolicyFactory" class="org.apache.camel.component.metrics.routepolicy.MetricsRoutePolicyFactory"/>

From Java code you can get hold of the com.codahale.metrics.MetricRegistry from the org.apache.camel.component.metrics.routepolicy.MetricsRegistryService as shown below:

MetricRegistryService registryService = context.hasService(MetricsRegistryService.class);
if (registryService != null) {
  MetricsRegistry registry = registryService.getMetricsRegistry();
  ...
}

The MetricsRoutePolicyFactory and MetricsRoutePolicy supports the following options:

Apache Camel allows you to customize the name of a CamelContext bean as it appears in JMX, by defining a management name pattern for it. For example, you can customize the name pattern of an XML CamelContext instance, as follows:

<camelContext id="myCamel" managementNamePattern="#name#">
    ...
</camelContext>

If you do not explicitly set a name pattern for the CamelContext bean, Apache Camel reverts to a default naming strategy.

By default, the JMX name of a CamelContext bean deployed in an OSGi bundle is equal to the OSGi symbolic name of the bundle. For example, if the OSGi symbolic name is MyCamelBundle, the JMX name would be MyCamelBundle. In cases where there is more than one CamelContext in the bundle, the JMX name is disambiguated by adding a counter value as a suffix. For example, if there are multiple Camel contexts in the MyCamelBundle bundle, the corresponding JMX MBeans are named as follows:

MyCamelBundle-1
MyCamelBundle-2
MyCamelBundle-3
...

One drawback of the default naming strategy is that you cannot guarantee that a given CamelContext bean will have the same JMX name between runs. If you want to have greater consistency between runs, you can control the JMX name more precisely by defining a JMX name pattern for the CamelContext instances.

To specify a name pattern on a CamelContext in Java, call the setNamePattern method, as follows:

// Java
context.getManagementNameStrategy().setNamePattern("#name#");

To specify a name pattern on a CamelContext in XML, set the managementNamePattern attribute on the camelContext element, as follows:

<camelContext id="myCamel" managementNamePattern="#name#">

You can construct a JMX name pattern by mixing literal text with any of the following tokens:

Here are some examples of JMX name patterns you could define using the supported tokens:

<camelContext id="fooContext" managementNamePattern="FooApplication-#name#">
    ...
</camelContext>
<camelContext id="myCamel" managementNamePattern="#bundleID#-#symbolicName#-#name#">
    ...
</camelContext>

Because the customised naming pattern overrides the default naming strategy, it is possible to define ambiguous JMX MBean names using this approach. For example:

<camelContext id="foo" managementNamePattern="SameOldSameOld"> ... </camelContext>
...
<camelContext id="bar" managementNamePattern="SameOldSameOld"> ... </camelContext>

In this case, Apache Camel would fail on start-up and report an MBean already exists exception. You should, therefore, take extra care to ensure that you do not define ambiguous name patterns.

The allowUseOriginalMessage option default setting is false, to cut down on copies being made of the original message when they are not needed. To enable the allowUseOriginalMessage option use the following commands: