Camel组件之ContentEnricher

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.

Enricher用来对已有的Message进行加工，如下图中所示的，Enricher从某个地方读取Resource，然后输出加工完毕的Message。如何将获取的Resource与原始的Message加工，则由用户指定。

 

 

对于Message的加工，可以通过一些方式实现，比如使用Velocity：

 

 

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

 

 

纯Java代码通过setBody()实现：

 

 

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

 

 

当然还可以使用Bean实现：

 

 

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

 

 

但是这里Content Enricher，来实现Message的加工。

 

 

Camel comes with two flavors of content enricher in the DSL

• enrich： 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.This operation merges data retrieved from another source using aproducer.
• pollEnrich：pollEnrich 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 operation merges data retrieved from another source usinga consumer.

对于这2种方式的理解：

 写道

The difference between pollEnrich and enrich
The difference between pollEnrich and enrich is that the former uses a consumer
and the latter a producer to retrieve data from the source. Knowing the difference is
important: the file component can be used with both, but using enrich will write the
message content as a file; using pollEnrich will read the file as the source, which
is most likely the scenario you’ll be facing when enriching with files. The HTTP component
only works with enrich; it allows you to invoke an external HTTP service and
use its reply as the source.

 

enrich以producer的方式获取数据，而pollEnrich则是以consumer的方式获取数据。何为producer，何为consumer？在这里（仅限这里），可以把在某个endpoint上提供服务叫做producer，获取服务叫做consumer。比如File组件，向文件夹中“提供”（写）新的文件为producer，而从文件夹里面“消费”（读取）文件为consumer。某些组件只能某种方式，比如HTTP组件，它只能作为consumer，调用一个HTTP的web 服务，然后获取reply作为Resource来加工。

看一个简单的enrich示例：

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

from("direct:resource")

 

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

 

上面提到“如何将获取的Resource与原始的Message加工，则由用户指定”，这里就需要使用AggregationStrategy了，方法 

 public Exchange aggregate(Exchange original, Exchange resource)

 

将original——原始的Message和resource——获取的Resource进行加工，返回新的Message。

 

下面看看pollEnricher的示例：

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

• receive
• receiveNoWait
• receive(timeout)

几个主要的参数：

uri		The endpoint uri for the external servie to enrich from. You must use either uri or ref. 获取Resource的Endpoint URI
ref		Refers to the endpoint for the external servie to enrich from. You must use either uri or ref. 获取Resource的Endpoint的reference（与URI而这必选其一）
strategyRef		Refers to an AggregationStrategy to be used to merge the reply from the external service, into a single outgoing message. By default Camel will use the reply from the external service as outgoing message. 加工的Strategy
timeout	0	Timeout in millis to use when polling from the external service. See below for important details about the timeout. 超时时间

By default Camel will use the receiveNoWait. 
If there is no data then the newExchange in the aggregation strategy is null.

没有数据的时候，其中的一个参数为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

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 theExchange. 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.

注意：当前的Exchange的数据没有用，也不要试图想从当前的Exchange（这里等同于Message）获取任何信息，比如Header。

简单的pollEnricher示例：

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