Message Channel

Camel supports the Message Channel from the EIP patterns. The Message Channel is an internal implementation detail of the Endpoint interface and all interactions with the Message Channel are via the Endpoint interfaces.

Example

In JMS, Message Channels are represented by topics and queues such as the following

jms:queue:foo

This message channel can be then used within the JMS component

Using the Fluent Builders

to("jms:queue:foo")

Using the Spring XML Extensions

<to uri="jms:queue:foo"/>

For more details see

• Message
• Message Endpoint

Message

Camel supports the Message from the EIP patterns using the Message interface.

To support various message exchange patterns like one way Event Message and Request Reply messages Camel uses an Exchange interface which has a pattern property which can be set to InOnly for an Event Message which has a single inbound Message, or InOut for a Request Reply where there is an inbound and outbound message.

Here is a basic example of sending a Message to a route in InOnly and InOut modes

Requestor Code

//InOnly
getContext().createProducerTemplate().sendBody("direct:startInOnly", "Hello World");

//InOut
String result = (String) getContext().createProducerTemplate().requestBody("direct:startInOut", "Hello World");

Route Using the Fluent Builders

from("direct:startInOnly").inOnly("bean:process");

from("direct:startInOut").inOut("bean:process");

Route Using the Spring XML Extensions

<route>
  <from uri="direct:startInOnly"/>
  <inOnly uri="bean:process"/>
</route>

<route>
  <from uri="direct:startInOut"/>
  <inOut uri="bean:process"/>
</route>

Pipes and Filters

Camel supports the Pipes and Filters from the EIP patterns in various ways.

With Camel you can split your processing across multiple independent Endpoint instances which can then be chained together.

Using Routing Logic

You can create pipelines of logic using multiple Endpoint or Message Translator instances as follows{snippet:id=example|lang=java|url=camel/trunk/camel-core/src/test/java/org/apache/camel/processor/PipelineTest.java}Though pipeline is the default mode of operation when you specify multiple outputs in Camel. The opposite to pipeline is multicast; which fires the same message into each of its outputs. (See the example below).

In Spring XML you can use the <pipeline/> element

In the above the pipeline element is actually unnecessary, you could use this:

which is a bit more explicit.

However if you wish to use <multicast/> to avoid a pipeline - to send the same message into multiple pipelines - then the <pipeline/> element comes into its own:

In the above example we are routing from a single Endpoint to a list of different endpoints specified using URIs. If you find the above a bit confusing, try reading about the Architecture or try the Examples

Using This Pattern

Message Router

The Message Router from the EIP patterns allows you to consume from an input destination, evaluate some predicate then choose the right output destination.

The following example shows how to route a request from an input queue:a endpoint to either queue:b, queue:c or queue:d depending on the evaluation of various Predicate expressions

Using the Fluent Builders

Using the Spring XML Extensions

Choice without otherwise

If you use a choice without adding an otherwise, any unmatched exchanges will be dropped by default.

Message Translator

Camel supports the Message Translator from the EIP patterns by using an arbitrary Processor in the routing logic, by using a bean to perform the transformation, or by using transform() in the DSL. You can also use a Data Format to marshal and unmarshal messages in different encodings.

Using the Fluent Builders

You can transform a message using Camel's Bean Integration to call any method on a bean in your Registry such as your Spring XML configuration file as follows

Where the "myTransformerBean" would be defined in a Spring XML file or defined in JNDI etc. You can omit the method name parameter from beanRef() and the Bean Integration will try to deduce the method to invoke from the message exchange.

or you can add your own explicit Processor to do the transformation

or you can use the DSL to explicitly configure the transformation

Use Spring XML

You can also use Spring XML Extensions to do a transformation. Basically any Expression language can be substituted inside the transform element as shown below

Or you can use the Bean Integration to invoke a bean

You can also use Templating to consume a message from one destination, transform it with something like Velocity or XQuery and then send it on to another destination. For example using InOnly (one way messaging)

If you want to use InOut (request-reply) semantics to process requests on the My.Queue queue on ActiveMQ with a template generated response, then sending responses back to the JMSReplyTo Destination you could use this.

• Content Enricher
• Using getIn or getOut methods on Exchange

Message Endpoint

Camel supports the Message Endpoint from the EIP patterns using the Endpoint interface.

When using the DSL to create Routes you typically refer to Message Endpoints by their URIs rather than directly using the Endpoint interface. Its then a responsibility of the CamelContext to create and activate the necessary Endpoint instances using the available Component implementations.

Example

The following example route demonstrates the use of a File Consumer Endpoint and JMS Producer Endpoint

Using the Fluent Builders

from("file://local/router/messages/foo")
	.to("jms:queue:foo");

Using the Spring XML Extensions

<route>
	<from uri="file://local/router/messages/foo"/>
	<to uri="jms:queue:foo"/>
</route>

Dynamic To

Available as of Camel 2.16

There is a new <toD> that allows to send a message to a dynamic computed Endpoint using one or more Expression that are concat together. By default the Simple language is used to compute the endpoint. For example to send a message to a endpoint defined by a header you can do

<route>
  <from uri="direct:start"/>
  <toD uri="${header.foo}"/>
</route>

And in Java DSL

from("direct:start")
  .toD("${header.foo}");

You can also prefix the uri with a value because by default the uri is evaluated using the Simple language

<route>
  <from uri="direct:start"/>
  <toD uri="mock:${header.foo}"/>
</route>

And in Java DSL

from("direct:start")
  .toD("mock:${header.foo}");

In the example above we compute an endpoint that has prefix "mock:" and then the header foo is appended. So for example if the header foo has value order, then the endpoint is computed as "mock:order".

You can also use other languages than Simple such as XPath - this requires to prefix with language: as shown below (simple language is the default language). If you do not specify language: then the endpoint is a component name. And in some cases there is both a component and language with the same name such as xquery.

<route>
  <from uri="direct:start"/>
  <toD uri="language:xpath:/order/@uri"/>
</route>

This is done by specifying the name of the language followed by a colon.

from("direct:start")
  .toD("language:xpath:/order/@uri");

You can also concat multiple Language(s) together using the plus sign + such as shown below:

<route>
  <from uri="direct:start"/>
  <toD uri="jms:${header.base}+language:xpath:/order/@id"/>
</route>

In the example above the uri is a combination of Simple language and XPath where the first part is simple (simple is default language). And then the plus sign separate to another language, where we specify the language name followed by a colon

from("direct:start")
  .toD("jms:${header.base}+language:xpath:/order/@id");

You can concat as many languages as you want, just separate them with the plus sign

The Dynamic To has a few options you can configure

For more details see

• Recipient List
• Message
• Wire Tap
  
  

Point to Point Channel

Camel supports the Point to Point Channel from the EIP patterns using the following components

• SEDA for in-VM seda based messaging
• JMS for working with JMS Queues for high performance, clustering and load balancing
• JPA for using a database as a simple message queue
• XMPP for point-to-point communication over XMPP (Jabber)
• and others

The following example demonstrates point to point messaging using the JMS component 

Using the Fluent Builders

from("direct:start")
	.to("jms:queue:foo");

Using the Spring XML Extensions

<route>
	<from uri="direct:start"/>
	<to uri="jms:queue:foo"/>
</route>

Publish Subscribe Channel

Camel supports the Publish Subscribe Channel from the EIP patterns using for example the following components:

• JMS for working with JMS Topics for high performance, clustering and load balancing
• XMPP when using rooms for group communication
• SEDA for working with SEDA in the same CamelContext which can work in pub-sub, but allowing multiple consumers.
• VM as SEDA but for intra-JVM.

Using Routing Logic

Another option is to explicitly list the publish-subscribe relationship in your routing logic; this keeps the producer and consumer decoupled but lets you control the fine grained routing configuration using the DSL or Xml Configuration.

Using the Fluent Builders

Using the Spring XML Extensions

Dead Letter Channel

Camel supports the Dead Letter Channel from the EIP patterns using the DeadLetterChannel processor which is an Error Handler.

Redelivery

It is common for a temporary outage or database deadlock to cause a message to fail to process; but the chances are if its tried a few more times with some time delay then it will complete fine. So we typically wish to use some kind of redelivery policy to decide how many times to try redeliver a message and how long to wait before redelivery attempts.

The RedeliveryPolicy defines how the message is to be redelivered. You can customize things like

• The number of times a message is attempted to be redelivered before it is considered a failure and sent to the dead letter channel.
• The initial redelivery timeout.
• Whether or not exponential backoff is used, i.e., the time between retries increases using a backoff multiplier.
• Whether to use collision avoidance to add some randomness to the timings.
• Delay pattern (see below for details).
• Camel 2.11: Whether to allow redelivery during stopping/shutdown.

Once all attempts at redelivering the message fails then the message is forwarded to the dead letter queue.

About Moving Exchange to Dead Letter Queue and Using handled()

handled() on Dead Letter Channel

When all attempts of redelivery have failed the Exchange is moved to the dead letter queue (the dead letter endpoint). The exchange is then complete and from the client point of view it was processed. As such the Dead Letter Channel have handled the Exchange.

For instance configuring the dead letter channel as:

Using the Fluent Builders

Using the Spring XML Extensions

The Dead Letter Channel above will clear the caused exception setException(null), by moving the caused exception to a property on the Exchange, with the key Exchange.EXCEPTION_CAUGHT. Then the Exchange is moved to the jms:queue:dead destination and the client will not notice the failure.

About Moving Exchange to Dead Letter Queue and Using the Original Message

The option useOriginalMessage is used for routing the original input message instead of the current message that potentially is modified during routing.

For instance if you have this route:

The route listen for JMS messages and validates, transforms and handle it. During this the Exchange payload is transformed/modified. So in case something goes wrong and we want to move the message to another JMS destination, then we can configure our Dead Letter Channel with the useOriginalMessage option. But when we move the Exchange to this destination we do not know in which state the message is in. Did the error happen in before the transformOrder or after? So to be sure we want to move the original input message we received from jms:queue:order:input. So we can do this by enabling the useOriginalMessage option as shown below:

Then the messages routed to the jms:queue:dead is the original input. If we want to manually retry we can move the JMS message from the failed to the input queue, with no problem as the message is the same as the original we received.

OnRedelivery

When Dead Letter Channel is doing redeliver its possible to configure a Processor that is executed just before every redelivery attempt. This can be used for the situations where you need to alter the message before its redelivered. See below for sample.

Redelivery Default Values

Redelivery is disabled by default.

The default redeliver policy will use the following values:

• maximumRedeliveries=0
• redeliverDelay=1000L (1 second)
• maximumRedeliveryDelay = 60 * 1000L (60 seconds)
• backOffMultiplier and useExponentialBackOff are ignored.
• retriesExhaustedLogLevel=LoggingLevel.ERROR
• retryAttemptedLogLevel=LoggingLevel.DEBUG
• Stack traces are logged for exhausted messages, from Camel 2.2.
• Handled exceptions are not logged, from Camel 2.3.
• logExhaustedMessageHistory is true for default error handler, and false for dead letter channel.
• logExhaustedMessageBody Camel 2.17: is disabled by default to avoid logging sensitive message body/header details. If this option is true, then logExhaustedMessageHistory must also be true.

The maximum redeliver delay ensures that a delay is never longer than the value, default 1 minute. This can happen when useExponentialBackOff=true.

The maximumRedeliveries is the number of re-delivery attempts. By default Camel will try to process the exchange 1 + 5 times. 1 time for the normal attempt and then 5 attempts as redeliveries.
Setting the maximumRedeliveries=-1 (or < -1) will then always redelivery (unlimited).
Setting the maximumRedeliveries=0 will disable re-delivery.

Camel will log delivery failures at the DEBUG logging level by default. You can change this by specifying retriesExhaustedLogLevel and/or retryAttemptedLogLevel. See ExceptionBuilderWithRetryLoggingLevelSetTest for an example.

You can turn logging of stack traces on/off. If turned off Camel will still log the redelivery attempt. It's just much less verbose.

Redeliver Delay Pattern

Delay pattern is used as a single option to set a range pattern for delays. When a delay pattern is in use the following options no longer apply:

• delay
• backOffMultiplier
• useExponentialBackOff
• useCollisionAvoidance
• maximumRedeliveryDelay

The idea is to set groups of ranges using the following syntax: limit:delay;limit 2:delay 2;limit 3:delay 3;...;limit N:delay N

Each group has two values separated with colon:

• limit = upper limit
• delay = delay in milliseconds
  And the groups is again separated with semi-colon. The rule of thumb is that the next groups should have a higher limit than the previous group.

Lets clarify this with an example:
delayPattern=5:1000;10:5000;20:20000

That gives us three groups:

• 5:1000
• 10:5000
• 20:20000

Resulting in these delays between redelivery attempts:

• Redelivery attempt number 1..4 = 0ms (as the first group start with 5)
• Redelivery attempt number 5..9 = 1000ms (the first group)
• Redelivery attempt number 10..19 = 5000ms (the second group)
• Redelivery attempt number 20.. = 20000ms (the last group)

Note: The first redelivery attempt is 1, so the first group should start with 1 or higher.

You can start a group with limit 1 to e.g., have a starting delay: delayPattern=1:1000;5:5000

• Redelivery attempt number 1..4 = 1000ms (the first group)
• Redelivery attempt number 5.. = 5000ms (the last group)

There is no requirement that the next delay should be higher than the previous. You can use any delay value you like. For example with delayPattern=1:5000;3:1000 we start with 5 sec delay and then later reduce that to 1 second.

Redelivery header

When a message is redelivered the DeadLetterChannel will append a customizable header to the message to indicate how many times its been redelivered.
Before Camel 2.6: The header is CamelRedeliveryCounter, which is also defined on the Exchange.REDELIVERY_COUNTER.
From Camel 2.6: The header CamelRedeliveryMaxCounter, which is also defined on the Exchange.REDELIVERY_MAX_COUNTER, contains the maximum redelivery setting. This header is absent if you use retryWhile or have unlimited maximum redelivery configured.

And a boolean flag whether it is being redelivered or not (first attempt). The header CamelRedelivered contains a boolean if the message is redelivered or not, which is also defined on the Exchange.REDELIVERED.

Dynamically Calculated Delay From the Exchange

In Camel 2.9 and 2.8.2: The header is CamelRedeliveryDelay, which is also defined on the Exchange.REDELIVERY_DELAY. If this header is absent, normal redelivery rules apply.

Which Endpoint Failed

Available as of Camel 2.1

When Camel routes messages it will decorate the Exchange with a property that contains the last endpoint Camel send the Exchange to:

The Exchange.TO_ENDPOINT have the constant value CamelToEndpoint. This information is updated when Camel sends a message to any endpoint. So if it exists its the last endpoint which Camel send the Exchange to.

When for example processing the Exchange at a given Endpoint and the message is to be moved into the dead letter queue, then Camel also decorates the Exchange with another property that contains that last endpoint:

The Exchange.FAILURE_ENDPOINT have the constant value CamelFailureEndpoint.

This allows for example you to fetch this information in your dead letter queue and use that for error reporting. This is usable if the Camel route is a bit dynamic such as the dynamic Recipient List so you know which endpoints failed.

Note: this information is retained on the Exchange even if the message is subsequently processed successfully by a given endpoint only to fail, for example, in local Bean processing instead. So, beware that this is a hint that helps pinpoint errors.

Now suppose the route above and a failure happens in the foo bean. Then the Exchange.TO_ENDPOINT and Exchange.FAILURE_ENDPOINT will still contain the value of http://someserver/somepath.

OnPrepareFailure

Available as of Camel 2.16

Before the exchange is sent to the dead letter queue, you can use onPrepare to allow a custom Processor to prepare the exchange, such as adding information why the Exchange failed.

For example, the following processor adds a header with the exception message:

Then configure the error handler to use the processor as follows:

Configuring this from XML DSL is as follows:

The onPrepare is also available using the default error handler.

Which Route Failed

Available as of Camel 2.10.4/2.11

When Camel error handler handles an error such as Dead Letter Channel or using Exception Clause with handled=true, then Camel will decorate the Exchange with the route id where the error occurred.

Example:

The Exchange.FAILURE_ROUTE_ID have the constant value CamelFailureRouteId. This allows for example you to fetch this information in your dead letter queue and use that for error reporting.

Control if Redelivery is Allowed During Stopping/Shutdown

Available as of Camel 2.11

Before Camel 2.10, Camel would perform redelivery while stopping a route, or shutting down Camel. This has improved a bit in Camel 2.10: Camel will no longer perform redelivery attempts when shutting down aggressively, e.g., during Graceful Shutdown and timeout hit.

From Camel 2.11: there is a new option allowRedeliveryWhileStopping which you can use to control if redelivery is allowed or not; notice that any in progress redelivery will still be executed. This option can only disallow any redelivery to be executed after the stopping of a route/shutdown of Camel has been triggered. If a redelivery is disallowed then a RejectedExcutionException is set on the Exchange and the processing of the Exchange stops. This means any consumer will see the Exchange as failed due the RejectedExcutionException. The default value is true for backward compatibility.

For example, the following snippet shows how to do this with Java DSL and XML DSL:{snippet:id=e1|lang=java|url=camel/trunk/camel-core/src/test/java/org/apache/camel/processor/RedeliveryErrorHandlerNoRedeliveryOnShutdownTest.java}And the sample sample with XML DSL{snippet:id=e1|lang=xml|url=camel/trunk/components/camel-spring/src/test/resources/org/apache/camel/spring/processor/SpringRedeliveryErrorHandlerNoRedeliveryOnShutdownTest.xml}

Samples

The following example shows how to configure the Dead Letter Channel configuration using the DSL{snippet:id=e3|lang=java|url=camel/trunk/camel-core/src/test/java/org/apache/camel/builder/ErrorHandlerTest.java}You can also configure the RedeliveryPolicy as this example shows{snippet:id=e4|lang=java|url=camel/trunk/camel-core/src/test/java/org/apache/camel/builder/ErrorHandlerTest.java}

How Can I Modify the Exchange Before Redelivery?

We support directly in Dead Letter Channel to set a Processor that is executed before each redelivery attempt. When Dead Letter Channel is doing redeliver its possible to configure a Processor that is executed just before every redelivery attempt. This can be used for the situations where you need to alter the message before its redelivered. Here we configure the Dead Letter Channel to use our processor MyRedeliveryProcessor to be executed before each redelivery.{snippet:id=e1|lang=java|url=camel/trunk/camel-core/src/test/java/org/apache/camel/processor/DeadLetterChannelOnRedeliveryTest.java}And this is the processor MyRedeliveryProcessor where we alter the message.{snippet:id=e2|lang=java|url=camel/trunk/camel-core/src/test/java/org/apache/camel/processor/DeadLetterChannelOnRedeliveryTest.java}

How Can I Log What Caused the Dead Letter Channel to be Invoked?

You often need to know what went wrong that caused the Dead Letter Channel to be used and it does not offer logging for this purpose. So the Dead Letter Channel's endpoint can be set to a endpoint of our own (such as direct:deadLetterChannel). We write a route to accept this Exchange and log the Exception, then forward on to where we want the failed Exchange moved to (which might be a DLQ queue for instance). See also http://stackoverflow.com/questions/13711462/logging-camel-exceptions-and-sending-to-the-dead-letter-channel

Using This Pattern

• Error Handler
• Exception Clause

Guaranteed Delivery

Camel supports the Guaranteed Delivery from the EIP patterns using among others the following components:

• File for using file systems as a persistent store of messages
• JMS when using persistent delivery (the default) for working with JMS Queues and Topics for high performance, clustering and load balancing
• JPA for using a database as a persistence layer, or use any of the many other database component such as SQL, JDBC, iBATIS/MyBatis, Hibernate
• HawtDB for a lightweight key-value persistent store

Example

The following example demonstrates illustrates the use of Guaranteed Delivery within the JMS component. By default, a message is not considered successfully delivered until the recipient has persisted the message locally guaranteeing its receipt in the event the destination becomes unavailable.

Using the Fluent Builders

from("direct:start")
	.to("jms:queue:foo");

Using the Spring XML Extensions

<route>
	<from uri="direct:start"/>
	<to uri="jms:queue:foo"/>
</route>

Message Bus

Camel supports the Message Bus from the EIP patterns. You could view Camel as a Message Bus itself as it allows producers and consumers to be decoupled.

Folks often assume that a Message Bus is a JMS though so you may wish to refer to the JMS component for traditional MOM support.
Also worthy of note is the XMPP component for supporting messaging over XMPP (Jabber)

Of course there are also ESB products such as Apache ServiceMix which serve as full fledged message busses.
You can interact with Apache ServiceMix from Camel in many ways, but in particular you can use the NMR or JBI component to access the ServiceMix message bus directly.

Example

The following demonstrates how the Camel message bus can be used to communicate with consumers and producers

Using the Fluent Builders

from("direct:start")
	.pollEnrich("file:inbox?fileName=data.txt")
	.to("jms:queue:foo");

Using the Spring XML Extensions

<route>
	<from uri="direct:start"/>
	<pollEnrich uri="file:inbox?fileName=data.txt"/>
	<to uri="jms:queue:foo"/>
</route>

Content Based Router

The Content Based Router from the EIP patterns allows you to route messages to the correct destination based on the contents of the message exchanges.

The following example shows how to route a request from an input seda:a endpoint to either seda:b, seda:c or seda:d depending on the evaluation of various Predicate expressions

Using the Fluent Builders

RouteBuilder builder = new RouteBuilder() {
    public void configure() {
        errorHandler(deadLetterChannel("mock:error"));
 
        from("direct:a")
            .choice()
                .when(header("foo").isEqualTo("bar"))
                    .to("direct:b")
                .when(header("foo").isEqualTo("cheese"))
                    .to("direct:c")
                .otherwise()
                    .to("direct:d");
    }
};

Using the Spring XML Extensions

<camelContext errorHandlerRef="errorHandler" xmlns="http://camel.apache.org/schema/spring">
    <route>
        <from uri="direct:a"/>
        <choice>
            <when>
                <xpath>$foo = 'bar'</xpath>
                <to uri="direct:b"/>
            </when>
            <when>
                <xpath>$foo = 'cheese'</xpath>
                <to uri="direct:c"/>
            </when>
            <otherwise>
                <to uri="direct:d"/>
            </otherwise>
        </choice>
    </route>
</camelContext>

Message Filter

The Message Filter from the EIP patterns allows you to filter messages

The following example shows how to create a Message Filter route consuming messages from an endpoint called queue:a, which if the Predicate is true will be dispatched to queue:b

Using the Fluent Builders{snippet:id=e2|lang=java|url=camel/trunk/camel-core/src/test/java/org/apache/camel/builder/RouteBuilderTest.java}You can, of course, use many different Predicate languages such as XPath, XQuery, SQL or various Scripting Languages. Here is an XPath example{snippet:id=example|lang=java|url=camel/trunk/camel-core/src/test/java/org/apache/camel/processor/XPathFilterTest.java}Here's another example of using a bean to define the filter behavior:

Using the Spring XML Extensions{snippet:id=example|lang=xml|url=camel/trunk/components/camel-spring/src/test/resources/org/apache/camel/spring/xml/buildSimpleRouteWithHeaderPredicate.xml}You can also use a method call expression (to call a method on a bean) in the Message Filter, as shown below:

For further examples of this pattern in use you could look at the junit test case

Using stop()

Stop is a bit different than a message filter as it will filter out all messages and end the route entirely (filter only applies to its child processor). Stop is convenient to use in a Content Based Router when you for example need to stop further processing in one of the predicates.

In the example below we do not want to route messages any further that has the word Bye in the message body. Notice how we prevent this in the when() predicate by using the .stop().{snippet:id=e1|lang=java|url=camel/trunk/camel-core/src/test/java/org/apache/camel/processor/RouteStopTest.java}

How To Determine If An  Exchange Was Filtered

Available as of Camel 2.5

The Message Filter EIP will add a property on the Exchange that states if it was filtered or not.

The property has the key Exchange.FILTER_MATCHED, which has the String value of CamelFilterMatched. Its value is a boolean indicating true or false. If the value is true then the Exchange was routed in the filter block. This property will be visible within the Message Filter block who's Predicate matches (value set to true), and to the steps immediately following the Message Filter with the value set based on the results of the last Message Filter Predicate evaluated.

Using This Pattern

Resequencer

The Resequencer from the EIP patterns allows you to reorganise messages based on some comparator. By default in Camel we use an Expression to create the comparator; so that you can compare by a message header or the body or a piece of a message etc.

Camel supports two resequencing algorithms:

• Batch resequencing collects messages into a batch, sorts the messages and sends them to their output.
• Stream resequencing re-orders (continuous) message streams based on the detection of gaps between messages.

By default the Resequencer does not support duplicate messages and will only keep the last message, in case a message arrives with the same message expression. However in the batch mode you can enable it to allow duplicates.

Batch Resequencing

The following example shows how to use the batch-processing resequencer so that messages are sorted in order of the body() expression. That is messages are collected into a batch (either by a maximum number of messages per batch or using a timeout) then they are sorted in order and then sent out to their output.

Using the Fluent Builders{snippet:id=example|lang=java|url=camel/trunk/camel-core/src/test/java/org/apache/camel/processor/ResequencerTest.java}This is equivalent to

The batch-processing resequencer can be further configured via the size() and timeout() methods.

This sets the batch size to 300 and the batch timeout to 4000 ms (by default, the batch size is 100 and the timeout is 1000 ms). Alternatively, you can provide a configuration object.

So the above example will reorder messages from endpoint direct:a in order of their bodies, to the endpoint mock:result.
Typically you'd use a header rather than the body to order things; or maybe a part of the body. So you could replace this expression with

for example to reorder messages using a custom sequence number in the header mySeqNo.

You can of course use many different Expression languages such as XPath, XQuery, SQL or various Scripting Languages.

Using the Spring XML Extensions

Allow Duplicates

Available as of Camel 2.4

In the batch mode, you can now allow duplicates. In Java DSL there is a allowDuplicates() method and in Spring XML there is an allowDuplicates=true attribute on the <batch-config/> you can use to enable it.

Reverse

Available as of Camel 2.4

In the batch mode, you can now reverse the expression ordering. By default the order is based on 0..9,A..Z, which would let messages with low numbers be ordered first, and thus also also outgoing first. In some cases you want to reverse order, which is now possible.

In Java DSL there is a reverse() method and in Spring XML there is an reverse=true attribute on the <batch-config/> you can use to enable it.

Resequence JMS messages based on JMSPriority

Available as of Camel 2.4

It's now much easier to use the Resequencer to resequence messages from JMS queues based on JMSPriority. For that to work you need to use the two new options allowDuplicates and reverse.{snippet:id=e1|lang=java|url=camel/trunk/components/camel-jms/src/test/java/org/apache/camel/component/jms/JmsBatchResequencerJMSPriorityTest.java}Notice this is only possible in the batch mode of the Resequencer.

Ignore invalid exchanges

Available as of Camel 2.9

The Resequencer EIP will from Camel 2.9 onwards throw a CamelExchangeException if the incoming Exchange is not valid for the resequencer - ie. the expression cannot be evaluated, such as a missing header. You can use the option ignoreInvalidExchanges to ignore these exceptions which means the Resequencer will then skip the invalid Exchange.{snippet:id=e1|lang=java|url=camel/trunk/camel-core/src/test/java/org/apache/camel/processor/ResequenceBatchIgnoreInvalidExchangesTest.java}This option is available for both batch and stream resequencer.

Reject Old Exchanges

Available as of Camel 2.11

This option can be used to prevent out of order messages from being sent regardless of the event that delivered messages downstream (capacity, timeout, etc). If enabled using rejectOld(), the Resequencer will throw a MessageRejectedException when an incoming Exchange is "older" (based on the Comparator) than the last delivered message. This provides an extra level of control with regards to delayed message ordering.

This option is available for the stream resequencer only.

Stream Resequencing

The next example shows how to use the stream-processing resequencer. Messages are re-ordered based on their sequence numbers given by a seqnum header using gap detection and timeouts on the level of individual messages.

Using the Fluent Builders{snippet:id=example|lang=java|url=camel/trunk/camel-core/src/test/java/org/apache/camel/processor/StreamResequencerTest.java}The stream-processing resequencer can be further configured via the capacity() and timeout() methods.

This sets the resequencer's capacity to 5000 and the timeout to 4000 ms (by default, the capacity is 1000 and the timeout is 1000 ms). Alternatively, you can provide a configuration object.

The stream-processing resequencer algorithm is based on the detection of gaps in a message stream rather than on a fixed batch size. Gap detection in combination with timeouts removes the constraint of having to know the number of messages of a sequence (i.e. the batch size) in advance. Messages must contain a unique sequence number for which a predecessor and a successor is known. For example a message with the sequence number 3 has a predecessor message with the sequence number 2 and a successor message with the sequence number 4. The message sequence 2,3,5 has a gap because the successor of 3 is missing. The resequencer therefore has to retain message 5 until message 4 arrives (or a timeout occurs).

If the maximum time difference between messages (with successor/predecessor relationship with respect to the sequence number) in a message stream is known, then the resequencer's timeout parameter should be set to this value. In this case it is guaranteed that all messages of a stream are delivered in correct order to the next processor. The lower the timeout value is compared to the out-of-sequence time difference the higher is the probability for out-of-sequence messages delivered by this resequencer. Large timeout values should be supported by sufficiently high capacity values. The capacity parameter is used to prevent the resequencer from running out of memory.

By default, the stream resequencer expects long sequence numbers but other sequence numbers types can be supported as well by providing a custom expression.{snippet:id=example|lang=java|url=camel/trunk/camel-core/src/test/java/org/apache/camel/processor/resequencer/MyFileNameExpression.java}{snippet:id=example|lang=java|url=camel/trunk/camel-core/src/test/java/org/apache/camel/processor/resequencer/ResequencerFileNameTest.java}or custom comparator via the comparator() method

or via a StreamResequencerConfig object.

Using the Spring XML Extensions

Further Examples

For further examples of this pattern in use you could look at the batch-processing resequencer junit test case and the stream-processing resequencer junit test case

Using This Pattern

Content Filter

Camel supports the Content Filter from the EIP patterns using one of the following mechanisms in the routing logic to transform content from the inbound message.

• Message Translator
• invoking a Java bean
• Processor object

A common way to filter messages is to use an Expression in the DSL like XQuery, SQL or one of the supported Scripting Languages.

Using the Fluent Builders

Here is a simple example using the DSL directly

In this example we add our own Processor

For further examples of this pattern in use you could look at one of the JUnit tests

• TransformTest
• TransformViaDSLTest

Using Spring XML

<route>
  <from uri="activemq:Input"/>
  <bean ref="myBeanName" method="doTransform"/>
  <to uri="activemq:Output"/>
</route>

You can also use XPath to filter out part of the message you are interested in:

<route>
  <from uri="activemq:Input"/>
  <setBody><xpath resultType="org.w3c.dom.Document">//foo:bar</xpath></setBody>
  <to uri="activemq:Output"/>
</route> 

Normalizer

Camel supports the Normalizer from the EIP patterns by using a Message Router in front of a number of Message Translator instances.

Example

This example shows a Message Normalizer that converts two types of XML messages into a common format. Messages in this common format are then filtered.

Using the Fluent Builders

In this case we're using a Java bean as the normalizer. The class looks like this

Using the Spring XML Extensions

The same example in the Spring DSL

See Also

• Message Router
• Content Based Router
• Message Translator

Messaging Mapper

Camel supports the Messaging Mapper from the EIP patterns by using either Message Translator pattern or the Type Converter module.

Example

The following example demonstrates the use of a Bean component to map between two messaging system

Using the Fluent Builders

from("activemq:foo")
	.beanRef("transformerBean", "transform")
	.to("jms:bar");

Using the Spring XML Extensions

<route>
	<from uri="activemq:foo"/>
	<bean ref="transformerBean" method="transform" />
	<to uri="jms:bar"/>
</route>

See also

• Message Translator
• Type Converter
• CXF for JAX-WS support for binding business logic to messaging & web services
• Pojo
• Bean

Event Driven Consumer

Camel supports the Event Driven Consumer from the EIP patterns. The default consumer model is event based (i.e. asynchronous) as this means that the Camel container can then manage pooling, threading and concurrency for you in a declarative manner.

The Event Driven Consumer is implemented by consumers implementing the Processor interface which is invoked by the Message Endpoint when a Message is available for processing.

Example

The following demonstrates a Processor defined in the Camel  Registry which is invoked when an event occurs from a JMS queue

Using the Fluent Builders

from("jms:queue:foo")
	.processRef("processor");

Using the Spring XML Extensions

<route>
	<from uri="jms:queue:foo"/>
	<to uri="processor"/>
</route>

For more details see

• Message
• Message Endpoint

Polling Consumer

Camel supports implementing the Polling Consumer from the EIP patterns using the PollingConsumer interface which can be created via the Endpoint.createPollingConsumer() method.

In Java:

The ConsumerTemplate (discussed below) is also available.

There are three main polling methods on PollingConsumer

EventDrivenPollingConsumer Options

The EventDrivePollingConsumer (the default implementation) supports the following options:

Notice that some Camel Components has their own implementation of PollingConsumer and therefore do not support the options above.

You can configure these options in endpoints URIs, such as shown below:

ConsumerTemplate

The ConsumerTemplate is a template much like Spring's JmsTemplate or JdbcTemplate supporting the Polling Consumer EIP. With the template you can consume Exchanges from an Endpoint. The template supports the three operations listed above. However, it also includes convenient methods for returning the body, etc consumeBody.

Example:

Or to extract and get the body you can do:

And you can provide the body type as a parameter and have it returned as the type:

You get hold of a ConsumerTemplate from the CamelContext with the createConsumerTemplate operation:

Using ConsumerTemplate with Spring DSL

With the Spring DSL we can declare the consumer in the CamelContext with the consumerTemplate tag, just like the ProducerTemplate. The example below illustrates this:{snippet:id=e1|lang=xml|url=camel/components/camel-spring/src/test/resources/org/apache/camel/spring/SpringConsumerTemplateTest-context.xml}Then we can get leverage Spring to inject the ConsumerTemplate in our java class. The code below is part of an unit test but it shows how the consumer and producer can work together.{snippet:id=e1|lang=java|url=camel/components/camel-spring/src/test/java/org/apache/camel/spring/SpringConsumerTemplateTest.java}

Timer Based Polling Consumer

In this sample we use a Timer to schedule a route to be started every 5th second and invoke our bean MyCoolBean where we implement the business logic for the Polling Consumer. Here we want to consume all messages from a JMS queue, process the message and send them to the next queue.

First we setup our route as:{snippet:id=e1|lang=java|url=camel/tags/camel-2.6.0/components/camel-jms/src/test/java/org/apache/camel/component/jms/JmsTimerBasedPollingConsumerTest.java}And then we have out logic in our bean:{snippet:id=e2|lang=java|url=camel/tags/camel-2.6.0/components/camel-jms/src/test/java/org/apache/camel/component/jms/JmsTimerBasedPollingConsumerTest.java}

Scheduled Poll Components

Quite a few inbound Camel endpoints use a scheduled poll pattern to receive messages and push them through the Camel processing routes. That is to say externally from the client the endpoint appears to use an Event Driven Consumer but internally a scheduled poll is used to monitor some kind of state or resource and then fire message exchanges.

Since this a such a common pattern, polling components can extend the ScheduledPollConsumer base class which makes it simpler to implement this pattern. There is also the Quartz Component which provides scheduled delivery of messages using the Quartz enterprise scheduler.

For more details see:

• PollingConsumer
• Scheduled Polling Components
  • ScheduledPollConsumer
  • Scheduler
  • Atom
  • Beanstalk
  • File
  • FTP
  • hbase
  • iBATIS
  • JPA
  • Mail
  • MyBatis
  • Quartz
  • SNMP
  • AWS-S3
  • AWS-SQS

ScheduledPollConsumer Options

The ScheduledPollConsumer supports the following options:

Using backoff to Let the Scheduler be Less Aggressive

Available as of Camel 2.12

The scheduled Polling Consumer is by default static by using the same poll frequency whether or not there is messages to pickup or not.

From Camel 2.12: you can configure the scheduled Polling Consumer to be more dynamic by using backoff. This allows the scheduler to skip N number of polls when it becomes idle, or there has been X number of errors in a row. See more details in the table above for the backoffXXX options.

For example to let a FTP consumer back-off if its becoming idle for a while you can do:

In this example, the FTP consumer will poll for new FTP files every 5th second. But if it has been idle for 5 attempts in a row, then it will back-off using a multiplier of 6, which means it will now poll every 5 x 6 = 30th second instead. When the consumer eventually pickup a file, then the back-off will reset, and the consumer will go back and poll every 5th second again.

Camel will log at DEBUG level using org.apache.camel.impl.ScheduledPollConsumer when back-off is kicking-in.

About Error Handling and Scheduled Polling Consumers

ScheduledPollConsumer is scheduled based and its run method is invoked periodically based on schedule settings. But errors can also occur when a poll is being executed. For instance if Camel should poll a file network, and this network resource is not available then a java.io.IOException could occur. As this error happens before any Exchange has been created and prepared for routing, then the regular Error handling in Camel does not apply. So what does the consumer do then? Well the exception is propagated back to the run method where its handled. Camel will by default log the exception at WARN level and then ignore it. At next schedule the error could have been resolved and thus being able to poll the endpoint successfully.

Using a Custom Scheduler

Available as of Camel 2.12:

The SPI interface org.apache.camel.spi.ScheduledPollConsumerScheduler allows to implement a custom scheduler to control when the Polling Consumer runs. The default implementation is based on the JDKs ScheduledExecutorService with a single thread in the thread pool. There is a CRON based implementation in the Quartz2, and Spring components.

For an example of developing and using a custom scheduler, see the unit test org.apache.camel.component.file.FileConsumerCustomSchedulerTest from the source code in camel-core.

Error Handling When Using PollingConsumerPollStrategy

org.apache.camel.PollingConsumerPollStrategy is a pluggable strategy that you can configure on the ScheduledPollConsumer. The default implementation org.apache.camel.impl.DefaultPollingConsumerPollStrategy will log the caused exception at WARN level and then ignore this issue.

The strategy interface provides the following three methods:

• begin
  
  • void begin(Consumer consumer, Endpoint endpoint)
• begin (Camel 2.3)
  • boolean begin(Consumer consumer, Endpoint endpoint)
• commit
  
  • void commit(Consumer consumer, Endpoint endpoint)
• commit (Camel 2.6)
  • void commit(Consumer consumer, Endpoint endpoint, int polledMessages)
• rollback
  
  • boolean rollback(Consumer consumer, Endpoint endpoint, int retryCounter, Exception e) throws Exception

In Camel 2.3: the begin method returns a boolean which indicates whether or not to skipping polling. So you can implement your custom logic and return false if you do not want to poll this time.

In Camel 2.6: the commit method has an additional parameter containing the number of message that was actually polled. For example if there was no messages polled, the value would be zero, and you can react accordingly.

The most interesting is the rollback as it allows you do handle the caused exception and decide what to do.

For instance if we want to provide a retry feature to a scheduled consumer we can implement the PollingConsumerPollStrategy method and put the retry logic in the rollback method. Lets just retry up till three times:

Notice that we are given the Consumer as a parameter. We could use this to restart the consumer as we can invoke stop and start:

Note: if you implement the begin operation make sure to avoid throwing exceptions as in such a case the poll operation is not invoked and Camel will invoke the rollback directly.

Configuring an Endpoint to Use PollingConsumerPollStrategy

To configure an Endpoint to use a custom PollingConsumerPollStrategy you use the option pollStrategy. For example in the file consumer below we want to use our custom strategy defined in the Registry with the bean id myPoll:

Using This Pattern

See Also

• POJO Consuming
• Batch Consumer

Competing Consumers

Camel supports the Competing Consumers from the EIP patterns using a few different components.

You can use the following components to implement competing consumers:-

• Seda for SEDA based concurrent processing using a thread pool
• JMS for distributed SEDA based concurrent processing with queues which support reliable load balancing, failover and clustering.

Enabling Competing Consumers with JMS

To enable Competing Consumers you just need to set the concurrentConsumers property on the JMS endpoint.

For example

from("jms:MyQueue?concurrentConsumers=5").bean(SomeBean.class);

or in Spring DSL

<route>
  <from uri="jms:MyQueue?concurrentConsumers=5"/>
  <to uri="bean:someBean"/>
</route>

Or just run multiple JVMs of any ActiveMQ or JMS route

Message Dispatcher

Camel supports the Message Dispatcher from the EIP patterns using various approaches.

You can use a component like JMS with selectors to implement a Selective Consumer as the Message Dispatcher implementation. Or you can use an Endpoint as the Message Dispatcher itself and then use a Content Based Router as the Message Dispatcher.

Example

The following example demonstrates Message Dispatcher pattern using the Competing Consumers functionality of the JMS component to offload messages to a Content Based Router and custom Processors registered in the Camel Registry running in separate threads from originating consumer.

Using the Fluent Builders

from("jms:queue:foo?concurrentConsumers=5")
	.threads(5)
	.choice()
		.when(header("type").isEqualTo("A")) 
			.processRef("messageDispatchProcessorA")
		.when(header("type").isEqualTo("B"))
			.processRef("messageDispatchProcessorB")
		.when(header("type").isEqualTo("C"))
			.processRef("messageDispatchProcessorC")		
		.otherwise()
			.to("jms:queue:invalidMessageType");

Using the Spring XML Extensions

<route>
	<from uri="jms:queue:foo?concurrentConsumers=5"/>
	<threads poolSize="5">
		<choice>
			<when>
				<simple>${in.header.type} == 'A'</simple>
				<to ref="messageDispatchProcessorA"/>
			</when>
			<when>
				<simple>${in.header.type} == 'B'</simple>
				<to ref="messageDispatchProcessorB"/>
			</when>
			<when>
				<simple>${in.header.type} == 'C'</simple>
				<to ref="messageDispatchProcessorC"/>
			</when>
			<otherwise>
				<to uri="jms:queue:invalidMessageType"/>
		</choice>
	</threads>
</route>

See Also

• JMS
• Selective Consumer
• Content Based Router
• Endpoint

Selective Consumer

The Selective Consumer from the EIP patterns can be implemented in two ways

The first solution is to provide a Message Selector to the underlying URIs when creating your consumer. For example when using JMS you can specify a selector parameter so that the message broker will only deliver messages matching your criteria.

The other approach is to use a Message Filter which is applied; then if the filter matches the message your consumer is invoked as shown in the following example

Using the Fluent Builders

Using the Spring XML Extensions

Durable Subscriber

Camel supports the Durable Subscriber from the EIP patterns using the JMS component which supports publish & subscribe using Topics with support for non-durable and durable subscribers.

Another alternative is to combine the Message Dispatcher or Content Based Router with File or JPA components for durable subscribers then something like Seda for non-durable.

Here is a simple example of creating durable subscribers to a JMS topic

Using the Fluent Builders

from("direct:start").to("activemq:topic:foo");

from("activemq:topic:foo?clientId=1&durableSubscriptionName=bar1").to("mock:result1");

from("activemq:topic:foo?clientId=2&durableSubscriptionName=bar2").to("mock:result2");

Using the Spring XML Extensions

<route>
    <from uri="direct:start"/>
    <to uri="activemq:topic:foo"/>
</route>

<route>
    <from uri="activemq:topic:foo?clientId=1&durableSubscriptionName=bar1"/>
    <to uri="mock:result1"/>
</route>

<route>
    <from uri="activemq:topic:foo?clientId=2&durableSubscriptionName=bar2"/>
    <to uri="mock:result2"/>
</route>

Here is another example of JMS durable subscribers, but this time using virtual topics (recommended by AMQ over durable subscriptions)

Using the Fluent Builders

from("direct:start").to("activemq:topic:VirtualTopic.foo");

from("activemq:queue:Consumer.1.VirtualTopic.foo").to("mock:result1");

from("activemq:queue:Consumer.2.VirtualTopic.foo").to("mock:result2");

Using the Spring XML Extensions

<route>
    <from uri="direct:start"/>
    <to uri="activemq:topic:VirtualTopic.foo"/>
</route>

<route>
    <from uri="activemq:queue:Consumer.1.VirtualTopic.foo"/>
    <to uri="mock:result1"/>
</route>

<route>
    <from uri="activemq:queue:Consumer.2.VirtualTopic.foo"/>
    <to uri="mock:result2"/>
</route>

See Also

• JMS
• File
• JPA
• Message Dispatcher
• Selective Consumer
• Content Based Router
• Endpoint

Idempotent Consumer

The Idempotent Consumer from the EIP patterns is used to filter out duplicate messages.

This pattern is implemented using the IdempotentConsumer class. This uses an Expression to calculate a unique message ID string for a given message exchange; this ID can then be looked up in the IdempotentRepository to see if it has been seen before; if it has the message is consumed; if its not then the message is processed and the ID is added to the repository.

The Idempotent Consumer essentially acts like a Message Filter to filter out duplicates.

Camel will add the message id eagerly to the repository to detect duplication also for Exchanges currently in progress.
On completion Camel will remove the message id from the repository if the Exchange failed, otherwise it stays there.

Camel provides the following Idempotent Consumer implementations:

• MemoryIdempotentRepository
• FileIdempotentRepository
• HazelcastIdempotentRepository (Available as of Camel 2.8)
• JdbcMessageIdRepository (Available as of Camel 2.7)
• JpaMessageIdRepository

• InfinispanIdempotentRepository (Available as of Camel 2.13.0)

• JCacheIdempotentRepository (Available as of Camel 2.17.0)

• SpringCacheIdempotentRepository (Available as of Camel 2.17.1)
  

• EhcacheIdempotentRepository (Available as of Camel 2.18.0)

• KafkaIdempotentRepository (Available as of Camel 2.19.0)

Options

The Idempotent Consumer has the following options:

Using the Fluent Builders

The following example will use the header myMessageId to filter out duplicates{snippet:id=idempotent|lang=java|url=camel/trunk/camel-core/src/test/java/org/apache/camel/builder/RouteBuilderTest.java}The above example will use an in-memory based MessageIdRepository which can easily run out of memory and doesn't work in a clustered environment. So you might prefer to use the JPA based implementation which uses a database to store the message IDs which have been processed{snippet:id=idempotent|lang=java|url=camel/trunk/components/camel-jpa/src/test/java/org/apache/camel/processor/jpa/JpaIdempotentConsumerTest.java}In the above example we are using the header messageId to filter out duplicates and using the collection myProcessorName to indicate the Message ID Repository to use. This name is important as you could process the same message by many different processors; so each may require its own logical Message ID Repository.

For further examples of this pattern in use you could look at the junit test case

Spring XML example

The following example will use the header myMessageId to filter out duplicates{snippet:id=e1|lang=xml|url=camel/trunk/components/camel-spring/src/test/resources/org/apache/camel/spring/processor/SpringIdempotentConsumerTest.xml}

How to handle duplicate messages in the route

Available as of Camel 2.8

You can now set the skipDuplicate option to false which instructs the idempotent consumer to route duplicate messages as well. However the duplicate message has been marked as duplicate by having a property on the Exchange set to true. We can leverage this fact by using a Content Based Router or Message Filter to detect this and handle duplicate messages.

For example in the following example we use the Message Filter to send the message to a duplicate endpoint, and then stop continue routing that message.{snippet:id=e1|lang=java|title=Filter duplicate messages|url=camel/trunk/camel-core/src/test/java/org/apache/camel/processor/IdempotentConsumerTest.java}The sample example in XML DSL would be:{snippet:id=e1|lang=xml|title=Filter duplicate messages|url=camel/trunk/components/camel-spring/src/test/resources/org/apache/camel/spring/processor/SpringIdempotentConsumerNoSkipDuplicateFilterTest.xml}

How to handle duplicate message in a clustered environment with a data grid

Available as of Camel 2.8

If you have running Camel in a clustered environment, a in memory idempotent repository doesn't work (see above). You can setup either a central database or use the idempotent consumer implementation based on the Hazelcast data grid. Hazelcast finds the nodes over multicast (which is default - configure Hazelcast for tcp-ip) and creates automatically a map based repository:

You have to define how long the repository should hold each message id (default is to delete it never). To avoid that you run out of memory you should create an eviction strategy based on the Hazelcast configuration. For additional information see camel-hazelcast.

See this little tutorial, how setup such an idempotent repository on two cluster nodes using Apache Karaf.

Available as of Camel 2.13.0

Another option for using Idempotent Consumer in a clustered environment is Infinispan. Infinispan is a data grid with replication and distribution clustering support. For additional information see camel-infinispan.

Using This Pattern

Transactional Client

Camel recommends supporting the Transactional Client from the EIP patterns using spring transactions.

Transaction Oriented Endpoints like JMS support using a transaction for both inbound and outbound message exchanges. Endpoints that support transactions will participate in the current transaction context that they are called from.

You should use the SpringRouteBuilder to setup the routes since you will need to setup the spring context with the TransactionTemplates that will define the transaction manager configuration and policies.

For inbound endpoint to be transacted, they normally need to be configured to use a Spring PlatformTransactionManager. In the case of the JMS component, this can be done by looking it up in the spring context.

You first define needed object in the spring configuration.

Then you look them up and use them to create the JmsComponent.

Transaction Policies

Outbound endpoints will automatically enlist in the current transaction context. But what if you do not want your outbound endpoint to enlist in the same transaction as your inbound endpoint? The solution is to add a Transaction Policy to the processing route. You first have to define transaction policies that you will be using. The policies use a spring TransactionTemplate under the covers for declaring the transaction demarcation to use. So you will need to add something like the following to your spring xml:

Then in your SpringRouteBuilder, you just need to create new SpringTransactionPolicy objects for each of the templates.

Once created, you can use the Policy objects in your processing routes:

OSGi Blueprint

If you are using OSGi Blueprint then you most likely have to explicit declare a policy and refer to the policy from the transacted in the route.

And then refer to "required" from the route:

Database Sample

In this sample we want to ensure that two endpoints is under transaction control. These two endpoints inserts data into a database.
The sample is in its full as a unit test.

First of all we setup the usual spring stuff in its configuration file. Here we have defined a DataSource to the HSQLDB and a most importantly the Spring DataSource TransactionManager that is doing the heavy lifting of ensuring our transactional policies. You are of course free to use any of the Spring based TransactionManager, eg. if you are in a full blown J2EE container you could use JTA or the WebLogic or WebSphere specific managers.

As we use the new convention over configuration we do not need to configure a transaction policy bean, so we do not have any PROPAGATION_REQUIRED beans. All the beans needed to be configured is standard Spring beans only, eg. there are no Camel specific configuration at all.{snippet:id=e1|lang=xml|url=camel/trunk/components/camel-spring/src/test/resources/org/apache/camel/spring/interceptor/springTransactionalClientDataSourceMinimalConfiguration.xml}Then we are ready to define our Camel routes. We have two routes: 1 for success conditions, and 1 for a forced rollback condition.
This is after all based on a unit test. Notice that we mark each route as transacted using the transacted tag.{snippet:id=e2|lang=xml|url=camel/trunk/components/camel-spring/src/test/resources/org/apache/camel/spring/interceptor/springTransactionalClientDataSourceMinimalConfiguration.xml}That is all that is needed to configure a Camel route as being transacted. Just remember to use the transacted DSL. The rest is standard Spring XML to setup the transaction manager.

JMS Sample

In this sample we want to listen for messages on a queue and process the messages with our business logic java code and send them along. Since its based on a unit test the destination is a mock endpoint.

First we configure the standard Spring XML to declare a JMS connection factory, a JMS transaction manager and our ActiveMQ component that we use in our routing.{snippet:id=e1|lang=xml|url=camel/trunk/components/camel-jms/src/test/resources/org/apache/camel/component/jms/tx/TransactionMinimalConfigurationTest.xml}And then we configure our routes. Notice that all we have to do is mark the route as transacted using the transacted tag.{snippet:id=e2|lang=xml|url=camel/trunk/components/camel-jms/src/test/resources/org/apache/camel/component/jms/tx/TransactionMinimalConfigurationTest.xml}

Integration Testing with Spring

When following the Spring Transactions documentation it is tempting to annotate your integration test with @Transactional then seed your database before firing up the route to be tested and sending a message in. This is incorrect as Spring will have an in-progress transaction, and Camel will wait on this before proceeding, leading to the route timing out.

Instead, remove the @Transactional annotation from the test method and seed the test data within a TransactionTemplate execution which will ensure the data is committed to the database before Camel attempts to pick up and use the transaction manager. A simple example can be found on GitHub.

Spring's transactional model ensures each transaction is bound to one thread. A Camel route may invoke additional threads which is where the blockage may occur. This is not a fault of Camel but as the programmer you must be aware of the consequences of beginning a transaction in a test thread and expecting a separate thread created by your Camel route to be participate, which it cannot. You can, in your test, mock the parts that cause separate threads to avoid this issue.

Using multiple routes with different propagation behaviors

Available as of Camel 2.2
Suppose you want to route a message through two routes and by which the 2nd route should run in its own transaction. How do you do that? You use propagation behaviors for that where you configure it as follows:

• The first route use PROPAGATION_REQUIRED
• The second route use PROPAGATION_REQUIRES_NEW

This is configured in the Spring XML file:{snippet:id=e1|lang=xml|url=camel/trunk/components/camel-spring/src/test/resources/org/apache/camel/spring/interceptor/MixedTransactionPropagationTest.xml}Then in the routes you use transacted DSL to indicate which of these two propagations it uses.{snippet:id=e1|lang=java|url=camel/trunk/components/camel-spring/src/test/java/org/apache/camel/spring/interceptor/MixedTransactionPropagationTest.java}Notice how we have configured the onException in the 2nd route to indicate in case of any exceptions we should handle it and just rollback this transaction. This is done using the markRollbackOnlyLast which tells Camel to only do it for the current transaction and not globally.

See Also

• Error handling in Camel
• TransactionErrorHandler
• Error Handler
• JMS

Using This Pattern

Messaging Gateway

Camel has several endpoint components that support the Messaging Gateway from the EIP patterns.

Components like Bean and CXF provide a a way to bind a Java interface to the message exchange.

However you may want to read the Using CamelProxy documentation as a true Messaging Gateway EIP solution.
Another approach is to use @Produce which you can read about in POJO Producing which also can be used as a Messaging Gateway EIP solution.

Example

The following example how the CXF and Bean components can be used to abstract the developer from the underlying messaging system API

Using the Fluent Builders

from("cxf:bean:soapMessageEndpoint")
	.to("bean:testBean?method=processSOAP");

Using the Spring XML Extensions

<route>
	<from uri="cxf:bean:soapMessageEndpoint"/>
	<to uri="bean:testBean?method=processSOAP"/>
</route>

See Also

• Bean
• CXF
• Using CamelProxy
• POJO Producing
• Spring Remoting

Service Activator

Camel has several endpoint components that support the Service Activator from the EIP patterns.

Components like Bean, CXF and Pojo provide a a way to bind the message exchange to a Java interface/service where the route defines the endpoints and wires it up to the bean.

In addition you can use the Bean Integration to wire messages to a bean using annotation.

Here is a simple example of using a Direct endpoint to create a messaging interface to a Pojo Bean service.

Using the Fluent Builders

from("direct:invokeMyService").to("bean:myService");

Using the Spring XML Extensions

<route>
    <from uri="direct:invokeMyService"/>
    <to uri="bean:myService"/>
</route>

See Also

• Bean
• Pojo
• CXF

Wire Tap

Wire Tap (from the EIP patterns) allows you to route messages to a separate location while they are being forwarded to the ultimate destination.

Options

WireTap Threadpool

The Wire Tap uses a thread pool to process the tapped messages. This thread pool will by default use the settings detailed at Threading Model. In particular, when the pool is exhausted (with all threads utilized), further wiretaps will be executed synchronously by the calling thread. To remedy this, you can configure an explicit thread pool on the Wire Tap having either a different rejection policy, a larger worker queue, or more worker threads.

WireTap Node

Camel's Wire Tap node supports two flavors when tapping an Exchange:

• With the traditional Wire Tap, Camel will copy the original Exchange and set its Exchange Pattern to InOnly, as we want the tapped Exchange to be sent in a fire and forget style. The tapped Exchange is then sent in a separate thread so it can run in parallel with the original. Beware that only the Exchange is copied - Wire Tap won't do a deep clone (unless you specify a custom processor via onPrepareRef which does that). So all copies could share objects from the original Exchange.
• Camel also provides an option of sending a new Exchange allowing you to populate it with new values.

Sending a Copy (traditional wiretap)

Using the Fluent Builders{snippet:id=e1|lang=java|url=camel/trunk/camel-core/src/test/java/org/apache/camel/processor/WireTapTest.java}Using the Spring XML Extensions{snippet:id=e1|lang=xml|url=camel/trunk/components/camel-spring/src/test/resources/org/apache/camel/spring/processor/SpringWireTapTest.xml}

Sending a New Exchange

Using the Fluent Builders
Camel supports either a processor or an Expression to populate the new Exchange. Using a processor gives you full power over how the Exchange is populated as you can set properties, headers, etc. An Expression can only be used to set the IN body.

From Camel 2.3: the Expression or Processor is pre-populated with a copy of the original Exchange, which allows you to access the original message when you prepare a new Exchange to be sent. You can use the copy option (enabled by default) to indicate whether you want this. If you set copy=false, then it works as in Camel 2.2 or older where the Exchange will be empty.

Below is the processor variation. This example is from Camel 2.3, where we disable copy by passing in false to create a new, empty Exchange.{snippet:id=e1|lang=java|url=camel/trunk/camel-core/src/test/java/org/apache/camel/processor/WireTapUsingFireAndForgetTest.java}Here is the Expression variation. In the following example we disable copying by setting copy=false which results in the creation of a new, empty Exchange.{snippet:id=e2|lang=java|url=camel/trunk/camel-core/src/test/java/org/apache/camel/processor/WireTapUsingFireAndForgetTest.java}Using the Spring XML Extensions
The processor variation, which uses a processorRef attribute to refer to a Spring bean by ID:{snippet:id=e2|lang=xml|url=camel/trunk/components/camel-spring/src/test/resources/org/apache/camel/spring/processor/SpringWireTapUsingFireAndForgetTest.xml}Here is the Expression variation, where the expression is defined in the body tag:{snippet:id=e1|lang=xml|url=camel/trunk/components/camel-spring/src/test/resources/org/apache/camel/spring/processor/SpringWireTapUsingFireAndForgetTest.xml}This variation accesses the body of the original message and creates a new Exchange based on the Expression. It will create a new Exchange and have the body contain "Bye ORIGINAL BODY MESSAGE HERE"{snippet:id=e1|lang=xml|url=camel/trunk/components/camel-spring/src/test/resources/org/apache/camel/spring/processor/SpringWireTapUsingFireAndForgetCopyTest.xml}

Further Example

For another example of this pattern, refer to the wire tap test case.

Using Dynamic URIs

Available as of Camel 2.16:

For example to wire tap to a dynamic URI, then it supports the same dynamic URIs as documented in Message Endpoint. For example to wire tap to a JMS queue where the header ID is part of the queue name:

Sending a New Exchange and Set Headers in DSL

Available as of Camel 2.8

If you send a new message using Wire Tap, then you could only set the message body using an Expression from the DSL. If you also need to set headers, you would have to use a Processor. From Camel 2.8: it's possible to set headers as well using the DSL.

The following example sends a new message which has

• Bye World as message body.
• A header with key id with the value 123.
• A header with key date which has current date as value.

Java DSL

{snippet:id=e1|lang=java|url=camel/trunk/camel-core/src/test/java/org/apache/camel/processor/WireTapNewExchangeTest.java}

XML DSL

The XML DSL is slightly different than Java DSL in how you configure the message body and headers using <body> and <setHeader>:{snippet:id=e1|lang=xml|url=camel/trunk/components/camel-spring/src/test/resources/org/apache/camel/spring/processor/SpringWireTapNewExchangeTest.xml}

Using onPrepare to Execute Custom Logic when Preparing Messages

Available as of Camel 2.8

See details at Multicast

Using This Pattern