apache-camel - 让 Camel 处理各种 URI 类型

Question

我想编写一个获取 URI(可以是 http、ftp、文件等)的 Camel Route，然后获取数据并将其存储在本地文件中。

这个 URI 字符串可以是，例如:

• "ftp://localhost/example.txt "
• “文件://tmp/example.txt”
• “jms:queue:dataInputQueue”
• ...

基于这个字符串，应该使用正确的 Camel 组件来访问数据。类似于 Java 中的 case/switch:

(1) 接收 URI(来自 uri="vm:incomingUri")

(2) 选择“正确的” Camel 组件

switch(URI)
case HTTP: use Camel HTTP component
case FTP: use Camel FTP component
case JMS: use Camel JMS component
...

(3) 使用“正确的”Camel 组件从该 URI 读取数据

(4) 本地存储文件(to uri="file://...)

例子:从“vm:incomingUri”我读到一个字符串“ftp://localhost/example.txt”。现在最终需要发生的事情应该等同于:

<route>
    <from uri="ftp://localhost/example.txt"/>
    <to uri="file://tmpDir/example.txt"/>
</route>

这在 Camel 中会是什么样子？

Answer 1

我相信一个困难是，对于您提到的组件(HTTP、FTP、文件、JMS)，您可能想要使用生产者或消费者:

• FTP、文件:绝对是读取文件的用户。
• HTTP(或HTTP4):绝对是生产者，向服务器发送请求(服务器的回复将以新消息体为准)
• JMS:取决于您是想从队列中读取(消费者)，还是向带有 ReplyTo header 的队列发送消息，然后等待答案(生产者)。

制作人:

如果您使用的是 Camel 2.16+，您可以使用新的 "dynamic to"句法。它与常规的“to”基本相同，除了可以使用 simple 动态评估端点 uri。表达式(或者，可选地，另一种类型的表达式)。或者，您可以使用 content-enricher pattern 的 enrich 风格，它还支持从 Camel 2.16 开始的动态 uris。

如果您使用的是旧版本的 Camel，或者如果您需要动态路由到多个端点(不仅仅是一个)，您可以使用 recipient list模式。

举个例子。我们将通过调用端点来转换消息体；该端点的 uri 将在名为 TargetUri 的 header 中找到，并将针对每条消息进行动态评估。

// An instance of this class is registered as 'testbean' in the registry. Instead of
// sending to this bean, I could send to a FTP or HTTP endpoint, or whatever.
public class TestBean {
    public String toUpperCase(final String str) {
        return str.toUpperCase();
    }
}

// This route sends a message to our example route for testing purpose. Of course, we
// could send any message as long as the 'TargetUri' header contains a valid endpoint uri
from("file:inbox?move=done&moveFailed=failed")
.setHeader("TargetUri").constant("bean:testbean?method=toUpperCase")
.setBody().constant("foo")
.to("direct:test");

// 1. The toD example :
from("direct:test")
.toD("${header.TargetUri}") 
.to("log:myRoute");

// 2. The recipient list example :
from("direct:test")
.recipientList(header("TargetUri"))
.to("log:myRoute");

// 3. The enrich example :
from("direct:test")
.enrich().simple("${header.TargetUri}") // add an AggregationStrategy if necessary
.to("log:myRoute");

消费者:

使用 Camel 2.16+，您可以使用 content-enricher pattern 的 pollEnrich 风格.

对于旧版本的 Camel，您可以在处理器中使用 ConsumerTemplate。

// 4. The pollEnrich example (assuming the TargetUri header contains, e.g., a file
// or ftp uri) :
from("direct:test")
.pollEnrich().simple("${header.TargetUri}") // add an AggregationStrategy if necessary
.to("log:myRoute");

// 5. The ConsumerTemplate example (same assumption as above)
from("direct:test")
.process(new Processor() {
    @Override
    public void process(Exchange exchange) throws Exception {
        String uri = exchange.getIn().getHeader("TargetUri", String.class);
        ConsumerTemplate consumer = exchange.getContext().createConsumerTemplate();
        final Object data = consumer.receiveBody(uri);
        exchange.getIn().setBody(data);
    }
})
.to("log:myRoute");

生产者还是消费者？

可悲的是，我想不出任何真正优雅的解决方案来处理这两个问题——我认为你将不得不根据 uri 和已知组件路由到两个分支……这是我可能会做的事情(使用 Camel 2.16+)，它不是很漂亮:

// This example only handles http and ftp endpoints properly
from("direct:test")
.choice()
    .when(header("TargetUri").startsWith("http"))
        .enrich().simple("${header.TargetUri}")
    .endChoice()
    .when(header("TargetUri").startsWith("ftp"))
        .pollEnrich().simple("${header.TargetUri}")
    .endChoice()
.end()
.to("log:myRoute");