Instant Apache Camel Message Routing

For the examples in this book, we are going to use Apache Camel version 2.11 (http://camel.apache.org/) and Apache Maven version 2.2.1 or newer (http://maven.apache.org/) as a build tool. Both of these projects can be downloaded for free from their websites. The complete source code for all the examples in this book is available on github at https://github.com/bibryam/camel-message-routing-examples repository. It contains Camel routes in Spring XML and Java DSL with accompanying unit tests. The source code for this tutorial is located under the project: camel-message-routing-examples/creating-camel-project.

That's all it takes to create our first Camel application. Now, we can run it using a Java IDE or from the command line with Maven mvn exec:java.

Camel has a modular architecture; its core (camel-core dependency) contains all the functionality needed to run a Camel application—DSL for various languages, the routing engine, implementations of EIPs, a number of data converters, and core components. This is the only dependency needed to run this application. Then there are optional technology specific connector dependencies (called components) such as JMS, SOAP, JDBC, Twitter, and so on, which are not needed for this example, as the file and log components we used are all part of the camel-core.

Camel routes are created using a Domain Specific Language (DSL), specifically tailored for application integration. Camel DSLs are high-level languages that allow us to easily create routes, combining various processing steps and EIPs without going into low-level implementation details. In the Java DSL, we create a route by extending RouteBuilder and overriding the configure method. A route represents a chain of processing steps applied to a message based on some rules. The route has a beginning defined by the from endpoint, and one or more processing steps commonly called "Processors" (which implement the Processor interface).

With the Pipes and Filters pattern, a large processing task is divided into a sequence of smaller independent processing steps (Filters) that are connected by channels (Pipes). Each filter processes messages received from the inbound channel, and publishes the result to the outbound channel. In our route, the processing steps are reading the file using a polling consumer, logging it and writing the file to the target folder, all of them piped by Camel in the sequence specified in the DSL. We can visualize the individual steps in the application with the following diagram:

A route has exactly one input called consumer and identified by the keyword from. A consumer receives messages from producers or external systems, wraps them in a Camel specific format called Exchange, and starts routing them. There are two types of consumers: a polling consumer that fetches messages periodically (for example, reading files from a folder) and an event-driven consumer that listens for events and gets activated when a message arrives (for example, an HTTP server). All the other processor nodes in the route are either a type of integration pattern or producers used for sending messages to various endpoints. Producers are identified by the keyword to and they are capable of converting exchanges and delivering them to other channels using the underlying transport mechanism. In our example, the log producer logs the files using the log4J API, whereas the file producer writes them to a target folder.

The route is not enough to have a running application; it is only a template that defines the processing steps. The engine that runs and manages the routes is called Camel Context. A high level view of CamelContext looks like the following diagram:

CamelContext is a dynamic multithread route container, responsible for managing all aspects of the routing: route lifecycle, message conversions, configurations, error handling, monitoring, and so on. When CamelContext is started, it starts the components, endpoints and activates the routes. The routes are kept running until CamelContext is stopped again when it performs a graceful shutdown giving time for all the in-flight messages to complete processing. CamelContext is dynamic, it allows us to start, stop routes, add new routes, or remove running routes at runtime. In our example, after adding the MoveFileRoute, we start CamelContext and let it copy files for 10 seconds, and then the application terminates. If we check the target folder, we should see files copied from the source folder.

Camel applications can run as standalone applications or can be embedded in other containers such as Spring or Apache Karaf. To make development and deployment to various environments easy, Camel provides a number of DSLs, including Spring XML, Blueprint XML, Groovy, and Scala. Next, we will have a look at the Spring XML DSL.

<dependency>
    <groupId>org.apache.camel</groupId>
    <artifactId>camel-spring</artifactId>
    <version>${camel-version}</version>
</dependency>

<beans xmlns="http://www.springframework.org/schema/beans"
        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
        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">
<camelContext xmlns="http://camel.apache.org/schema/spring">
    <route>
        <from uri="file://source"/>
            <to uri="log://org.apache.camel.howto?showAll=true"/>
            <to uri="file://target"/>
    </route>
</camelContext>
</beans>

public static void main(String[] args) throws Exception {
    AbstractApplicationContext springContext = new ClassPathXmlApplicationContext("META-INF/spring/move-file-context.xml");
    springContext.start();
    Thread.sleep(10000);
    springContext.stop();
}

The complete source code for this tutorial is located under the following project: camel-message-routing-examples/routing-different-destinations.

We will extend the previous example by adding conditions to direct incoming files into different folders based on file names:

from("file://source")
        .choice()
        .when(simple("${in.header.CamelFileName} contains 'widget.txt'"))
            .to("file://widget")
        .when(simple("${in.header.CamelFileName} contains 'gadget.txt'"))
            .to("file://gadget")
        .otherwise()
            .to("log://org.apache.camel.howto?showAll=true");

A Context-Based Router is similar to a switch statement or to the if then-elseif statement in Java. The incoming message will be directed to one of the possible channels, depending on the condition that is evaluated, usually against the message content. In our example, there are three possible destinations for the incoming messages, and the conditions are written using Simple language. The condition has to be a predicate, that is, an expression that returns only true or false. If the result is true, the message is routed into that channel of the Context-Based Router, otherwise the next condition in order is evaluated. Similar to the default case in a Java switch statement, there is an optional otherwise element for this pattern. If the message doesn't match any of the predicates, it will be directed to the default channel of the route.

Camel expressions operate on the routed message and are used as part of various integration patterns. There are two types of expressions: one that implements the Expression interface and can produce a value from any type, and the other that implements the Predicate interface and produces a Boolean result. Expressions give access to all parts of the routed message, which in Camel world is represented by the Exchange class and has the following structure:

The Exchange class wraps the mandatory In Message representing the request data coming from an inbound channel, and an optional Out Message for the response going into an outbound channel. Each message consists of a payload called Body with Object type and Headers, which is a Map for storing key-value pairs associated with the message. The Exchange class also has Properties for storing Camel specific data, Message Exchange Pattern (MEP), ExchangeId, and Exception field for tracking exceptions if any were encountered during routing. Usually, when a processor receives Exchange, it reads the data from In Message (including Body and Headers), and depending on MEP writes the processed results to Out Message or updates In Message, to be passed to the next processor in the route. In our example, file consumer reads a file from the source folder then puts its reference to the message body, and populates various headers with information about the file (this is where the CamelFileName header gets populated). Then the expressions in the Context-Based Router retrieve the file name from the headers and do the comparison. If the file name contains the string widget.txt the message is sent to the widget folder, otherwise the next condition is evaluated which does a similar comparison with the gadget.txt string. If the message doesn't satisfy any of the conditions, it is sent to the default channel where it is logged and not written to any folder.

Next, we will have a look at the other expression languages that Camel supports, and how to make the routing even more dynamic, using Java beans for deciding which is the next endpoint for a message.

Dynamic routing

A Context-Based Router can direct a message to the correct recipient if all the recipients are known in advance. All of the outbound channels of the Context-Based Router have to be specified as part of its definition, and only then it can choose one based on the message content. This introduces a dependency of the router to all possible destinations and prevents it from choosing a destination dynamically. Dynamic Router pattern solves this problem by choosing a destination for each message at runtime, and configuring itself using control messages from each participating destination.

In Camel, a Dynamic Router pattern is implemented with the dynamicRouter statement that accepts an expression to determine where the message should go next. After routing the message, the expression is evaluated again using the updated message to determine where the message should be routed next. Evaluating the expression and routing the message continues until the expression returns null to indicate the end of routing for that message. So it is important for the expression to return null at some point, otherwise Camel will continue routing the same message endlessly. Because of this complex nature, dynamicRouter expressions usually use Bean expressions:

from("direct:start")
    .dynamicRouter(method(DestinationChooser.class, "nextEndpoint"));

Note

For an example using the Dynamic Router, have a look at http://camel.apache.org/dynamic-router.html.

The complete source code for this tutorial is located under the project camel-message-routing-examples/using-components.

Camel-core contains a dozen components which are always available with Camel. These include bean, direct, vm, file, log, mock, seda, timer, xslt, and a few more. All of the other components used for routing have to be retrieved and added additionally to the classpath.

When this route is started, it will start a Jetty web server listening on localhost port 8080. If we use a web browser and try to access http://localhost:8080, the route will log the request, make an HTTP call to http://www.google.co.uk, and return the result to the requestor. And, what we should see in our browser (if all works fine) is Google's home page. So this route is in fact an HTTP proxy that logs all requests.

Routes in Camel consist of EIPs and components. EIPs transform the messages inside the routes, whereas components usually allow messages to reach external systems. Components can be added to Camel runtime (CamelContext) in two ways. The less popular way is manually instantiating a component and adding it to the context:

camelContext.addComponent("jetty", new JettyHttpComponent());

The second way of adding components to Camel runtime is through the auto-discovery mechanism. In this mode, when a component is needed as part of the routing process and it is not available to CamelContext, Camel will scan the classpath and look at META-INF/services/org/apache/camel/component for a file matching the component name. The jetty component's file META-INF/services/org/apache/camel/component/jetty contains the following:

class=org.apache.camel.component.jetty.JettyHttpComponent

The content of the file shows the actual component class, so that Camel can instantiate it, and add it to CamelContext. A component is in essence an endpoint factory, it creates endpoints based on URLs. The Endpoint interface models the end of channel through which systems can send and receive messages. Camel endpoints are usually referenced using URLs. URLs provide a unified way for describing endpoints for various systems. The Camel URL has the following structure: schema:context path:options. For example, the Jetty URL from the previous route looks like:

jetty://http://www.google.co.uk?bridgeEndpoint=true&throwExceptionOnFailure=false

Its schema is jetty, which tells Camel what is the component class name, that the context path is http://www.google.co.uk and it has two options bridgeEndpoint=true&throwExceptionOnFailure=false. The context path is endpoint specific; it means different things for different endpoints. For example, the context path of the log endpoint used above org.apache.camel.howto represents the logging category, whereas in Jetty it is the host and port name. Options are also endpoint specific, and each endpoint has different options for the underlying technology. Once an endpoint is instantiated for a specific URL, then it is responsible for creating consumers and producers.

At a very high level, the process of using a component in a Camel route goes through the following steps: when a URI is used in a route, based on the URI schema, a component is instantiated. The component uses the context path and the options to create an endpoint. Then the endpoint creates a consumer or producer depending on what is needed in the route. When the consumer receives a message from the external system, it transforms the message to an exchange and starts routing it in Camel. The producer, on the other hand, receives an exchange as part of the routing, and delivers it to the external system using the external API.

We saw how to pass primitive types and string literals as Endpoint options. Next, we will see how Camel offers a unified access to the environment it is running on, and how to pass complex objects from its environment as options in the URLs.

SimpleRegistry registry = new SimpleRegistry();
registry.put("myHttpClient", new HttpClient());
CamelContext context = new DefaultCamelContext(registry);

.to("jetty:http://www.google.co.uk?bridgeEndpoint=true&httpClient=#myHttpClient");

<bean id="myHttpClient" class="org.eclipse.jetty.client.HttpClient"/>

The complete source code for this tutorial is located under the project, camel-message-routing-examples/connecting-routes.

Typical reasons for splitting a routing logic into multiple smaller routes are transaction boundaries, different error handling strategies for different parts of the application, code reuse, testability, and so on. In our example, we created two routes only to demonstrate the direct component usage. Direct component, as the name suggests, provides direct synchronous invocation of the consumer when a message is sent to that endpoint using a direct producer. Each direct endpoint is identified by a unique name in CamelContext, and it can have only one consumer that receives messages and one or multiple producers. In our example, route one sends messages to direct:logger using a direct producer, and route two consumes from the same endpoint.

Using direct component allows connecting two routes as if they were one route: the message flows through all processors of the first route, then the second one and returns control back to the consumer of the first route for the next message. Because it is synchronous invocation (literally direct producer invokes the direct consumer method) and uses the same thread for processing the messages in both routes, it is also possible to make both routes participate in the same transaction when a transacted route is created.

Another thing to notice is that we started using mock component. Mock endpoints collect all the exchanges they receive and allow us write tests and verify the routing logic. In a real word application, instead of having mock endpoints hardcoded in the route we would have other endpoints for further processing the messages and replace them with mock endpoints during testing. We can see how mock endpoints are used by looking at the unit tests accompanying each route, and we will also demonstrate how to use them in the later part of this book.

In addition to synchronous route connections, Camel also lets us connect routes asynchronously, connect routes which are in different CamelContexts, and also exchange messages with routes from custom Java code.

Asynchronous connection

Another option for connecting routes is to use the SEDA (Staged Event Driven Architecture) component. SEDA provides asynchronous behavior, where messages are exchanged using an inmemory BlockingQueue. When this component is used, the producer sends a message and instantly returns to process the next message while letting the second route to process the messages using its own thread pool. This decoupling of routing threads allows us to implement patterns such as Competing Consumers (by setting the concurrentConsumers option to a number greater than 1) or a Publish-Subscribe pattern (by setting the multipleConsumers option to true). With the Competing Consumers pattern, a message is delivered to only one from multiple consumers, allowing concurrent processing, whereas with Publish-Subscribe patterns each consumer receives a copy of the published messages. This component is ideal for situations when there is a long running task, and to prevent from blocking the whole route, we could create a separate route for the long running task and connect it to the main route using a SEDA component. With this approach, long running tasks will be processed by a separate thread on the subroute keeping the main route unblocked.

One important concept in Camel, which is also related to the SEDA component, is the message exchange pattern. Camel messages are either one way event messages (InOnly) or request reply (InOut). By default, many components use an InOnly exchange pattern, but we can specify the exchange pattern explicitly as part of the endpoint invocation:

from("activemq:someQueue")
    .inOnly("seda:nextRoute");

We can also specify it as a part of the route:

from("mq:someQueue")
    .setExchangePattern(ExchangePattern.InOut)
    .to("mock:result");

With one way messaging (which is also known as fire and forget) the sender doesn't expect a response from the receiver, whereas with request reply style messaging a response is expected. The Exchange object keeps track of MEP and handles one way messages with its In Message field, and uses Out Message when a response is expected.

We can see the different ways for explicitly specifying the exchange pattern at http://camel.apache.org/request-reply.html.

The SEDA component will behave differently depending on the MEP. If the MEP is InOnly, the SEDA producer puts the message to the queue and continues with the next message without waiting for it to be processed. But, if the MEP is InOut, after passing the message it will block and wait for it to be processed (by a different thread from the SEDA consumer) or timeout occurs. This default behavior can be controlled with the waitForTaskToComplete option, where setting it to Always will make it always wait forthe exchange to be processed and setting a value to Never will prevent it from waiting, regardless of the messaging pattern. We can see all the options available to the SEDA component at http://camel.apache.org/seda.html.

ProducerTemplate producerTemplate = camelContext.createProducerTemplate();
template.sendBody("direct:logger", "<hello>world!</hello>");

template.sendBody("activemq:my.queue", "<hello>world!</hello>");

ConsumerTemplate consumerTemplate = camelContext().createConsumerTemplate();
String result = consumerTemplate.receiveBody("activemq:my.queue", String.class);

The complete source code for this tutorial is located under the following project: camel-message-routing-examples/removing-unwanted-messages.

Start with a filter definition, followed by a predicate expression and the processors to which this filter applies:

from("direct:start")
    .filter(header("userStatus").isEqualTo("valid"))
        .to("mock:valid")
    .end()
    .to("mock:all");

In the Java DSL, optionally we can mark the end of the filter, specifying up to where the filter applies. If not specified it applies for the rest of the route.

This pattern is a simpler version of the Context-Based Router pattern demonstrated in the previous tutorial. It is conceptually similar to a Java if statement. If the incoming message satisfies the condition specified in the filter, the message is passed to the child processors. If the message fails to satisfy the condition, it is discarded and not passed to the child processor. In our example, we let only messages with valid userStatus header reach the mock:valid endpoint. None of the other messages will be passed to the mock:valid endpoint. Note that when a message is filtered out, it is not totally stopped from routing, it is not passed to child processors but it continues routing after the filter element. There is another command called stop that is used to completely stop a message from being processed further.

Filtering messages is a very common operation in pipelines. Next, we will have a look at a special type of filter, the Idempotent consumer, and how to use Java beans for filtering messages, and as part of routes in general.

import static org.apache.camel.processor.idempotent.MemoryIdempotentRepository.memoryIdempotentRepository;
from("direct:start")
    .idempotentConsumer(
header("messageId"), memoryIdempotentRepository())
    .to("mock:result");

filter().method("myFilterBean", "isValidRequest")

public boolean isValidRequest(@Header("userId") String userId, @Body String body, Exchange exchange)

filter().method("myFilterBean", "isValidRequest(@Header("userId"), ${body}, null)")

 from(...)
    .bean(OrderService.class, "doSomething(${body}, ${header.high})")

 from(...)  
    .to("bean:orderService?method=doSomething(${body.asXml}, ${header.high})")

The complete source code for this tutorial is located under the project camel-message-routing-examples/transforming-messages.

In addition to the standard camel-core dependencies, for this tutorial we will need also the xmljson component:

<dependency>
    <groupId>org.apache.camel</groupId>
    <artifactId>camel-xmljson</artifactId>
    <version>${camel-version}</version>
</dependency>

Camel data formats are used to transform data between a low-level presentation and a high-level presentation by providing two operations: marshall and unmarshall. XmlJson data format is marshaling from XML to JSON and unmarshaling from a JSON to XML format. There are other data formats that convert from Java to XML and JSON or vice versa, but all of them require a POJO. The difference for this data format is that it doesn't require a POJO for the conversion. The only thing we did was to instantiate a dataFormat instance with the xmljson type and configure its options for this specific transformation.

When the default options are good enough, there is no need to instantiate a data format instance. Instead, as part of the route, we only specify the operation type (marshall/unmarshall) and the data Format type:

<marshal><xmljson/></marshal>

Message translator is a very general pattern and data transformations happen in many different places in Camel. We will see now different techniques for transforming messages and a very useful pattern for normalizing messages into a common format.

from(...)
    .convertBodyTo(String.class)
    .to(...)

Document document = message.getBody(Document.class);

</transform>
    <simple>New message ${body}</simple>
</transform>

.to("bean:myConverter?method=convert(${body}, ${header.userId})")

from("...")
    .process(new Processor() {
        public void process(Exchange exchange) throws Exception {
            exchange.getIn().setBody("Changed body");
        }
    })
    .to("...");

Normalizing messages to a common format

In some occasions, semantically equivalent data is received in different formats from disparate sources and it has to be converted to a common data format for uniform processing throughout the system. In these situations using the Normalizer pattern can be a good way to go, as shown in the following diagram:

As we can see in the preceding diagram, the Normalizer uses a Context-Based Router to route each message to an appropriate message translator depending on the message format. Once the messages are transformed to a common format they all can be processed in a unified way. We can see example routes using this pattern in Spring XML or Java DSL at http://camel.apache.org/normalizer.html.

The complete source code for this tutorial is located under the project camel-message-routing-examples/splitting-message.

The only required configuration for Splitter is the criteria used to split the messages. It can be any expression language, so we have the freedom to choose from Simple language, XPath, XQuery, and many others, depending mainly on the message type and ease of splitting. In our example, to split the incoming XML we chose to use XPath expression that will be applied to the main message, and the sub messages will be sent to next endpoint(s).

There are a couple of options to customize the way newly created Exchanges are processed. With onPrepareRef it is possible to reference a Processor in the Registry that will be called just before sending the sub message to child endpoints. Setting parallelProcessing will cause new Exchanges to be processed in parallel instead of sequentially; stopOnException will interrupt further processing remaining sub messages if an Exception is thrown and return the error, whereas leaving the default false value will finish processing all sub messages and return the error at the end. Splitter also supports splitting messages using tokenizer expressions and splitting streamed messages, which is useful for chunking large files with a low memory footprint.

In addition to splitting messages, the Splitter can also aggregate the messages back into one. This turns Splitter into a very powerful and compact construct used for splitting and then aggregating messages without a need for an additional Aggregator.

Exchange aggregate(Exchange oldExchange, Exchange newExchange)

<threadPoolProfile id="defaultThreadPoolProfile" defaultProfile="true" poolSize="10" maxPoolSize="20" maxQueueSize="1000" rejectedPolicy="CallerRuns"/>

The complete source code for this tutorial is located under the project camel-message-routing-examples/aggregating-messages.

The Aggregator also has two other mandatory settings: the correlation expression and the completion condition. The first one is evaluated on each incoming message to generate a correlation key. The key is used to determine which messages should be grouped together. In our example, the key would be the invoiceId, so all invoice item messages belonging to the same invoice will be merged into one message. The completion condition is used to tell Camel when the aggregation is completed and the currently built message should be sent further down the route.

When a new message comes in, the first thing done by the Aggregator is to evaluate the correlation expression and get the correlation key. Then the correlation key is used to look up the internal repository for an existing message with the same correlation key. The Aggregator has an internal (in memory by default) repository that contains the current aggregated Exchange for each correlation key. The next step is to pass the old Exchange (if there is one) and the new Exchange to the AggregationStrategy to merge them. After the aggregation, the completeness condition is checked, and if it is satisfied the aggregated Exchange is sent to the next processor in the route. If the aggregation has not completed yet, the internal repository is updated with the aggregated Exchange, and it continues waiting for the next Exchange to complete it or for timeout to occur.

There are a number of ways for specifying the completion condition and at least one should be present: completionPredicate is a Predicate expression which has to evaluate to true in order to complete the aggregated Exchange; completionSize states how many messages should be aggregated to complete the aggregation, it can be a fixed number or dynamically retrieved from the Exchange with an expression. completionTimeout is based on an inactivity period in milliseconds of the aggregated message, and completes the aggregated Exchange if there are no further incoming messages within the specified time period; completionInterval is similar but completes all currently aggregated messages after each time interval in milliseconds. The latter two completion conditions are asynchronous conditions, because these conditions are not evaluated on each incoming message, they run in a background thread and whenever a condition is satisfied the corresponding aggregated message starts its journey on the route.

The Aggregator is usually used together with other patterns. Here are two common integration patterns that the Aggregator takes part in:

Composed message processor

This pattern is useful for maintaining the incoming message flow while processing each element of the composite messages in a separate flow.

In the preceding routing diagram, we can see how the incoming composite message is split up into sub messages and then each sub message is routed to an appropriate destination using a Context-Based Router. After each sub message has been processed differently based on its type, they are aggregated back into a single message using Aggregator. Notice that the composed message processor pattern can also be achieved without using Aggregator explicitly. If you remember from the previous tutorial that Splitter has an embedded Aggregator in the form of AggregationStrategy. Using that strategy, Splitter can aggregate the results from each sub message and have the same effect as having an explicit Aggregator.

Scatter-gather

Here, a message is broadcast to multiple channels using the recipients list pattern or the publish subscribe mechanism. The difference in this case is that there is no Splitter involved, so each recipient gets the same copy of the incoming message.

After the message has been processed by multiple recipients, Aggregator receives the results and produces one single result. In this scenario, Aggregator usually does not merge results; instead it picks up only one of them because they are identical, in the example in the preceding diagram – the "Best" Quote.

The complete source code for this tutorial is located under the project camel-message-routing-examples/reorganizing-messages.

Let's start with a route that is supposably getting random messages, and we want to put them back in order. Here is how a Resequencer definition looks in Java DSL:

from("direct:start")
    .resequence(header("message_index"))
    .batch().size(100).timeout(1000L)
    .to("mock:result");

The main piece of information Resequencer needs to know is the criteria that will be used to order the messages. This can be the message body itself, or a header or property and is specified using an expression. In addition to the header used for ordering messages, in our example there are two other options: batchSize and batchTimeout. These options control how long the Resequencer should wait and collect messages before sorting and letting them go.

In the previous example, the Resequencer will collect up to 100 messages or will wait up to one second (whichever happens first), and then if there are any messages batched, sort them by the message_index header value, and let the messages further down the pipeline. When used in the default batching mode, Resequencer can also do reverse sorting, allow duplicate messages, and reject old messages or messages with an invalid sequence number. The downside of the batch processing algorithm is that it always collects messages up to a certain number or timeout, reducing the overall throughput of the system. There is also a stream based algorithm which reorders messages continuously based on detections of gaps between messages rather than fixed batch size. In order to use stream-based Resequencer, the messages must contain a unique sequence number for which the predecessor and successor is known. For example, the message sequence 1, 2, 4 has a gap, and the Resequencer will let message 1 and 2 go instantly without any batching, but retain message 4 until 3 arrives or timeout occurs. Configuring a streaming based Resequencer is very similar to a batch based one, except this one has the timeout and capacity options:

from("direct:start")
    .resequence(header("message_index"))
    .stream().capacity(100).timeout(1000L)
    .to("mock:result");

More information about the Resequencer pattern and how to use it can be found here at http://camel.apache.org/resequencer.html.

There are two other patterns that work on the message flow without modifying the message content. Let's have a look at Throttler and Delayer patterns.

<route>
    <from uri="seda:a"/>
        <throttle maximumRequestsPerPeriod="3" timePeriodMillis="10000">
            <to uri="mock:result"/>
        </throttle>
</route>

<route>
    <from uri="direct:start"/>
    <delay asyncDelayed="true">
        <constant>1000</constant>
    </delay>
    <to uri="mock:result"/>
</route>

The complete source code for this tutorial is located under the project camel-message-routing-examples/multicasting-messages.

When using a static Recipient List, we hardcode the outbound channels as part of the route definition. We can also specify a custom AggregationStrategy, options for parallel processing, and whether to continue multicasting the message if an error occurs in any of the recipients.

from("direct:start")
    .multicast(new HighestQuoteAggregator())
        .parallelProcessing().stopOnException(false)
        .to("mock:a", "mock:b", "mock:c")
    .end()
    .to("mock:result");

When a message reaches multicast, if the parallelProcessing options are not set, by default the message will be processed sequentially from the mock:a, mock:b, and mock:c endpoints. But, notice that this is different from processing a message sequentially in a pipeline style by a Camel route. The reason is that multicast will create a copy of the original Exchange for each recipient and mock:b will not see any changes done by mock:a or mock:c. Each of the recipients will get their own copy of the original incoming Exchange. One thing to keep in mind is that Camel will not do a deep copy and if we have any objects in the message body or headers, they will be shared across all recipients and potentially mutated concurrently. For these scenarios, a good place for making proper object copies is in the onPrepareRef Processor:

.onPrepareRef(new Processor() {
    @Override
    public void process(Exchange exchange) throws Exception {
        Order body = exchange.getIn().getBody(Order.class);
        Order clone = body.deepClone();
        exchange.getIn().setBody(clone);
    }
})

Similarly to Splitter and Aggregator, we can use a custom AggregationStrategy for aggregating replies from all recipients and create the outgoing message from the multicast. The default AggregationStrategy will simply pick up the last reply from the recipients.

Also, notice that we had to specify the end of the multicast definition, otherwise all the remaining endpoints from the route would also be considered as part of the multicast. Then, there are options which are common with the Splitter pattern: executorServiceRef, for custom thread pool, stopOnException, to stop broadcasting a message to the remaining recipients in case of errors, shareUnitOfWork, and so on. Sharing a unit of work allows multicasted Exchanges to report back any failures and propagate the exceptions to the original Exchange.

There are two other Camel patterns which allow delivery of messages to multiple recipients: Recipient List which has very similar behavior to multicast but uses expression to choose the recipients dynamically; and Wire Tap which is a much simpler version of multicast and allows sending a copy of the incoming message to one additional recipient.

from("direct:a")
    .recipientList(header("recipients"));

Wire Tap

Wire Tap sends a copy of the incoming message to one separate channel while keeping the final destination of the message unchanged. By default, it will create a copy of the original message and process it using a separate thread pool in a fire-and-forget fashion.

Wire Tap cannot deliver messages to more than one additional destination or use dynamic recipients. Also, it doesn't allow propagating exceptions by sharing a unit of work or stopping when an exception is encountered. It is mainly useful for monitoring, logging, and troubleshooting purposes. It is fully documented on this page http://camel.apache.org/wire-tap.html.

The complete source code for this tutorial is located under the project camel-message-routing-examples/error-handling.

An error handler can either be global, applying to all routes in a CamelContext, or applied to individual routes. There are different types of error handlers, each with different options and behaving slightly differently. In our example, we will create a DeadLetterChannel error handler applied to one route only.

As we already know, the consumer is responsible for receiving messages from other systems, creating an Exchange and then it starts routing it. If an error occurs before the routing of a message, for example, if the consumer cannot read a file from the file system, error handling will not be triggered. It is the responsibility of the consumer to deal with errors arising before the start of the routing process. Camel error handling applies only during routing of Exchanges. When a Processor (Endpoint or EIP) throws an exception during routing, Camel catches the exception and stores it in the Exchange's exception field. The error handler kicks in only when the Exchange contains a caught exception. There are different error handlers available and one is always selected:

In our example, for the mandatory deadLetterUri option, we have specified mock:error, so if any error happens during the routing, the error handler will take the control, cleanup the exception from the Exchange, and move the message to this endpoint. Notice that deadLetterUri can also be another route with multiple steps, where we could still access the exception that caused the failure from an Exchange property:

Exception exception = exchange.getProperty(Exchange.EXCEPTION_CAUGHT, Exception.class);

The important thing to remember is that the exception field of the Exchange will be cleaned up and for the caller it looks like the Exchange has been processed successfully.

Usually in a route, we have multiple steps (processors), which are connected by channels, and each step can modify the message. When an exception is thrown from a processor, the channel responsible for that processor notices the exception and passes the control to the error handler. Depending on where the exception has occurred, it is possible that the Exchange has been modified from successfully completed processors and DeadLetterChannel will get a modified Exchange. To solve this problem, errorHandler has the useOriginalMessage option. When this option is set, DeadLetterChannel will receive an Exchange containing the original In Message Body that was passed to the route. So, in our example, if the mock:result throws an exception, even though the message body has been modified by the transform step, the DeadLetterChannel's endpoint mock:error will receive the original message passed to the route. This option is useful for situations where we want to replay the failed Exchanges again to the same route using the original message.

Certain kinds of errors, for example caused by database deadlock, network failure, and so on are likely to succeed when retried a couple of times a little later. For these kinds of scenarios, DeadLetterChannel(and also DefaultErrorHandler and TransactionErrorHandler) offers a redelivery feature which is configured with a RediliveryPolicy. For our example, we have allowed up to three retry attempts with a 1 second delay between the attempts. In addition, we have set it to exponentially back off and use backOffMultiplier of 2, so the second attempt will be done 2 seconds after the first one, and the third and final, attempt will be done 4 seconds after the second. If all of these attempts to process the request fail, only then the message will be moved to DLQ.

There are two other mechanisms in Camel for dealing with erroneous situations: Exception Clause and doTry-doCatch-doFinally construct. The first one can be used together with error handlers, whereas the latter allows catching exceptions similarly to the Java language. Also, we will have quick look at monitoring and logging tools used with Camel.

<route>
    <from uri="direct:start"/>
    <onException>
        <exception>org.camel.ValidationException</exception>
        <exception>org.camel.OrderFailedException</exception>
        <redeliveryPolicy maximumRedeliveries="1"/>
        <handled>
            <constant>true</constant>
        </handled>
        <to uri="mock:error"/>
    </onException>
    <process ref="orderValidator"/>
    <to uri="mock:result"/>
</route>

<route>
    <from uri="direct:start"/>
    <doTry>
        <process ref="orderValidator"/>
        <doCatch> 
<exception>org.camel.ValidationException</exception>
<exception>org.camel.OrderFailedException</exception>
            <to uri="mock:catch"/>
        </doCatch>
        <doFinally>
            <to uri="mock:finally"/>
        </doFinally>
    </doTry>
</route>

The complete source code for this tutorial is located under the following project: camel-message-routing-examples/testing-routes.

We will use JUnit, but there is also TestNG support, although it has fewer features.

<dependency>
    <groupId>junit</groupId>
    <artifactId>junit</artifactId>
    <version>${junit-version}</version>
    <scope>test</scope>
</dependency>

For testing applications using Java DSL we need the camel-test dependency:

<dependency>
    <groupId>org.apache.camel</groupId>
    <artifactId>camel-test</artifactId>
    <version>${camel-version}</version>
    <scope>test</scope>
</dependency>

CamelTestSupport is the base class for testing routes. During the setup stage, it will create a new CamelContext, producerTemplate, consumerTemplate, load the routes defined in the test class and start them. Then, in our test, we can access mock endpoints, set expectations and trigger events or directly send messages to the route endpoints using producerTemplate. At the end of each test method CamelTestSupport will stop the routes and CamelContext (unless it is configured to tear down everything at the end of all tests in the test class).

The base class also provides lots of other helpful methods and hooks to customize the test lifecycle. For example, in some cases it is required to set expectations or do some work before starting the routes. That is possible by overriding the isUseAdviceWith method to return true which will prevent CamelContext from starting automatically and it has to be started manually as part of tests:

@Override
public boolean isUseAdviceWith() {
    return true;
}
public void pollsFilesOnStart() throws Exception {
    getMockEndpoint("mock:result").expectedBodiesReceived("Some file content");
    camelContext.start();
    assertMockEndpointsSatisfied();
}

Mock endpoint (http://camel.apache.org/mock.html) is another helpful tool used for testing routes. It is similar to other mocking libraries such as Mockito and jMock, and represents an inmemory list that collects all Exchanges it interacts with. It has mainly two types of methods: for specifying expectations and for verifying them, but it can also simulate specific behavior:

mockReplying.whenAnyExchangeReceived(new Processor() { 
    @Override
    public void process(Exchange exchange) throws Exception {
        Message in = exchange.getIn();
        in.setBody("Mock response: " + in.getBody());
    }
});

The previous code snippet configures mockReplying to modify the message body when an Exchange is received in order to simulate the behavior of the mocked endpoint.

Testing Spring XML DSL is no different, apart from using the Spring's ApplicationContext for creating the routes. Here, we will have a look at that and a few other testing tools.

<dependency>
    <groupId>org.apache.camel</groupId>
    <artifactId>camel-test-spring</artifactId>
    <version>${camel-version}</version>
    <scope>test</scope>
</dependency>

public class SimpleChoiceRouteSpringTest extends CamelSpringTestSupport {
    @Override
    protected AbstractApplicationContext createApplicationContext() {
        return new ClassPathXmlApplicationContext("META-INF/spring/simple-choice-route-context.xml");
    }
    //the rest of the code is the same as previous
 }

@RunWith(CamelSpringJUnit4ClassRunner.class)
@DirtiesContext(classMode = DirtiesContext.ClassMode.AFTER_EACH_TEST_METHOD)
@ContextConfiguration(locations = {"classpath:/META-INF/spring/simple-choice-route-context.xml"})
public class SimpleChoiceRouteEnhancedTest {
    @Autowired
    private CamelContext camelContext;

    @Produce(uri = "direct:start")
    protected ProducerTemplate start;
    //the rest of the class is the same as the example previous
}