Page edited by Claus Ibsen

Content Enricher

Camel supports the Content Enricher from the EIP patterns using a Message Translator, an artibrary Processor in the routing logic or using the enrich DSL element to enrich the message.

Content enrichment using a Message Translator or a Processor

Using the Fluent Builders

You can 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)

from("activemq:My.Queue").
  to("velocity:com/acme/MyResponse.vm").
  to("activemq:Another.Queue");

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.

from("activemq:My.Queue").
  to("velocity:com/acme/MyResponse.vm");

Here is a simple example using the DSL directly to transform the message body

from("direct:start").setBody(body().append(" World!")).to("mock:result");

In this example we add our own Processor using explicit Java code

from("direct:start").process(new Processor() {
    public void process(Exchange exchange) {
        Message in = exchange.getIn();
        in.setBody(in.getBody(String.class) + " World!");
    }
}).to("mock:result");

Finally we can use Bean Integration to use any Java method on any bean to act as the transformer

from("activemq:My.Queue").
  beanRef("myBeanName", "myMethodName").
  to("activemq:Another.Queue");

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



Using Spring XML

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




Content enrichment using the enrich DSL element

Camel comes with two flavors of content enricher in the DSL

	enrich
	pollEnrich



enrich is using a Producer to obtain the additional data. It is usually used for Request Reply messaging, for instance to invoke an external web service.

poolEnrich on the other hand is using a Polling Consumer to obtain the additional data. It is usually used for Event Message messaging, for instance to read a file or download a FTP file.

This feature is available since Camel 2.0

Using the Fluent Builders


AggregationStrategy aggregationStrategy = ...

from("direct:start")
.enrich("direct:resource", aggregationStrategy)
.to("direct:result");

from("direct:resource")
...



The content enricher (enrich) retrieves additional data from a resource endpoint in order to enrich an incoming message (contained in the orginal exchange). An aggregation strategy is used to combine the original exchange and the resource exchange. The first parameter of the AggregationStrategy.aggregate(Exchange, Exchange) method corresponds to the the original exchange, the second parameter the resource exchange. The results from the resource endpoint are stored in the resource exchange's out-message. Here's an example template for implementing an aggregation strategy.


public class ExampleAggregationStrategy implements AggregationStrategy {

    public Exchange aggregate(Exchange original, Exchange resource) {
        Object originalBody = original.getIn().getBody();
        Object resourceResponse = resource.getOut().getBody();
        Object mergeResult = ... // combine original body and resource response
        if (original.getPattern().isOutCapable()) {
            original.getOut().setBody(mergeResult);
        } else {
            original.getIn().setBody(mergeResult);
        }
        return original;
    }
    
}



Using this template the original exchange can be of any pattern. The resource exchange created by the enricher is always an in-out exchange.

Using Spring XML

The same example in the Spring DSL


<camelContext id="camel" xmlns="http://camel.apache.org/schema/spring">
  <route>
    <from uri="direct:start"/>
    <enrich uri="direct:resource" strategyRef="aggregationStrategy"/>
    <to uri="direct:result"/>
  </route>
  <route>
    <from uri="direct:resource"/>
    ...
  </route>
</camelContext>

<bean id="aggregationStrategy" class="..." />



Aggregation strategy is optional
The aggregation strategy is optional. If you do not provide it Camel will by default just use the body obtained from the resource.

from("direct:start")
  .enrich("direct:resource")
  .to("direct:result");


In the route above the message send to the direct:result endpoint will contain the output from the direct:resource as we do not use any custom aggregation.

And in Spring DSL you just omit the strategyRef attribute:

  <route>
    <from uri="direct:start"/>
    <enrich uri="direct:resource"/>
    <to uri="direct:result"/>
  </route>



Content enrich using pollEnrich
The pollEnrich works just as the enrich however as it uses a Polling Consumer we have 3 methods when polling

	receive
	receiveNoWait
	receive(timeout)



By default Camel will use the receiveNoWait. 

If there is no data then the newExchange in the aggregation strategy is null. 

You can pass in a timeout value that determines which method to use

	timeout is -1 or negative then receive is selected
	timeout is 0 then receiveNoWait is selected
	otherwise receive(timeout) is selected



The timeout values is in millis.

Data from current Exchange not used
pollEnrich does not access any data from the current Exchange which means when polling it cannot use any of the existing headers you may have set on the Exchange. For example you cannot set a filename in the Exchange.FILE_NAME header and use pollEnrich to consume only that file. For that you must set the filename in the endpoint URI.

The sample below reads a file based on a JMS message that contains a header with the filename.

from("activemq:queue:order")
   .setHeader(Exchange.FILE_NAME, header("orderId"))
   .pollEnrich("file://order/data/additional")
   .to("bean:processOrder");



And if we want to wait at most 20 seconds for the file to be ready we can use a timeout:


from("activemq:queue:order")
   .setHeader(Exchange.FILE_NAME, header("orderId"))
   .pollEnrich("file://order/data/additional", 20000)
   .to("bean:processOrder");



And yes pollEnrich also supports the aggregation strategy so we can pass it in as an argument too:

   .pollEnrich("file://order/data/additional", 20000, aggregationStrategy)



Using This Pattern

If you would like to use this EIP Pattern then please read the Getting Started, you may also find the Architecture useful particularly the description of Endpoint and URIs. Then you could try out some of the Examples first before trying this pattern out.

Change Notification Preferences

View Online | View Change | Add Comment

[CONF] Apache Camel > Content Enricher

Content Enricher

Page edited by Claus Ibsen

Content Enricher

Content enrichment using a Message Translator or a Processor

Content enrichment using the `enrich` DSL element

Aggregation strategy is optional

Content enrich using `pollEnrich`

Using This Pattern

Reply via email to