第 10 章 消息转换

内容丰富的 模式描述了消息目的地在原始消息中比存在更多的数据的情况。在这种情况下，您将使用消息转换器、路由逻辑中的任意处理器、路由逻辑中的任意处理器或内容丰富的方法从外部资源拉取额外数据。

图 10.1. 内容增强模式

Apache Camel 支持多种丰富内容的方法：

Camel 提供 流畅的构建器，用于通过提供智能完成且安全重构的 type-safe IDE 友好的方式创建路由和调解规则。当您测试分布式系统时，必须非常常见的要求才能根特定外部系统，以便您可以在特定系统可用或编写特定系统前测试系统的其他部分。其中一种实现方式是通过生成 具有最多 静态正文的动态消息来生成对请求的响应。使用模板的另一种方法是使用一个目的地的消息，使用诸如 Velocity 或 XQuery 之类的内容转换，然后将其发送到另一目的地。以下示例显示 这个信息 ：

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

假设您要使用 InOut （请求）消息传递来处理 ActiveMQ 上 My.Queue 队列上的请求。您希望模板生成的响应进入 JMSReplyTo 目标。以下示例演示了如何进行此操作：

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

以下简单示例演示了如何使用 DSL 来转换消息正文：

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

以下示例使用显式 Java 代码来添加处理器：

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");

下一个示例使用 bean 集成来启用使用任意 bean 作为转换器：

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

以下示例显示了 Spring XML 实现：

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

AggregationStrategy aggregationStrategy = ...

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

from("direct:resource")
...

内容丰富（增强 ）从 资源端点 检索额外的数据，以便增强传入消息（包含在组织 交换中）。聚合策略结合了原始交换 和资源交换。AggregationStrategy.aggregate (Exchange, Exchange) 方法的第一个参数对应于原始交换，第二个参数对应于资源交换。资源端点的结果存储在资源交换的 Out 消息中。以下是实施您自己的聚合策略类的示例模板：

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

}

使用此模板时，原始交换可以有任何交换模式。增强程序创建的资源交换始终是一个 InOut 交换。

前面的示例也可以在 Spring XML 中实施：

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

聚合策略是可选的。如果没有提供，Apache Camel 将使用默认情况下从资源获得的正文。例如：

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

在前面的路由中，发送到 direct:result 端点的消息包含 direct:resource 的输出，因为本示例不使用任何自定义聚合。

在 XML DSL 中，只需省略 strategyRef 属性，如下所示：

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

enrich DSL 命令支持以下选项：

enrich （） 方法从资源端点检索额外的数据，以丰富传入消息，该消息包含在原始交换中。您可以使用聚合策略来组合原始交换和资源交换。AggregationStrategy.aggregate (Exchange, Exchange) 方法的第一个参数对应于原始交换。第二个参数对应于资源交换。资源端点的结果存储在资源交换的 Out 消息中。例如：

AggregationStrategy aggregationStrategy = ...

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

   from("direct:resource")
...

以下代码是实施聚合策略的模板。在使用此模板的实施中，原始交换可以是任何消息交换模式。增强器创建的资源交换始终是 InOut 消息交换模式。

public class ExampleAggregationStrategy implements AggregationStrategy {

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

}

以下示例显示了使用 Spring XML DSL 来实施聚合策略：

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

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

从 Camel 2.16 开始，enrich （） 和 pollEnrich （） 方法支持根据当前交换的信息计算使用动态 URI。例如，若要从 HTTP 端点提供 orderId 键的标头作为 HTTP URL 的内容路径的一部分，您可以执行如下操作：

from("direct:start")
  .enrich().simple("http:myserver/${header.orderId}/order")
  .to("direct:result");

以下是 XML DSL 中的相同示例：

<camelContext id="camel" xmlns="http://camel.apache.org/schema/spring">
<route>
   <from uri="direct:start"/>
   <enrich>
      <simple>http:myserver/${header.orderId}/order</simple>
   </enrich>
   <to uri="direct:result"/>
</route>

pollEnrich 命令将资源端点视为 使用者。它将交换发送到资源端点，而不是 轮询 端点。默认情况下，如果资源端点没有交换，则轮询会立即返回。例如，以下路由会读取从传入 JMS 消息的标头中提取名称的文件：

from("activemq:queue:order")
   .pollEnrich("file://order/data/additional?fileName=orderId")
   .to("bean:processOrder");

您可以限制等待文件就绪的时间。以下示例显示了最多等待 20 秒：

from("activemq:queue:order")
   .pollEnrich("file://order/data/additional?fileName=orderId", 20000) // timeout is in milliseconds
   .to("bean:processOrder");

您还可以为 pollEnrich （） 指定聚合策略，例如：

   .pollEnrich("file://order/data/additional?fileName=orderId", 20000, aggregationStrategy)

pollEnrich （） 方法支持配置了 consumer.bridgeErrorHandler=true 的使用者。这可让轮询中的任何异常传播到路由错误处理程序，例如，可以重试轮询。

如果在接收交换前轮询超时，则传递到聚合策略的 aggregate （） 方法的资源交换可能为空。

pollEnrich （） 方法通过调用以下轮询方法之一来轮询消费者端点：

pollEnrich （） 命令的 timeout 参数（以毫秒为单位指定）决定要调用的方法，如下所示：

如果没有数据，则聚合策略中的 newExchange 为 null。

以下示例显示了通过从 inbox/data.txt 文件中加载内容来丰富消息：

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

以下是 XML DSL 中的相同示例：

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

如果指定的文件不存在，则消息为空。您可以指定一个超时等待（可能永久），直到文件存在或最多等待特定时长。在以下示例中，命令会等待不超过 5 秒：

<route>
   <from uri="direct:start"/>
   <pollEnrich timeout="5000">
      <constant>file:inbox?fileName=data.txt"</constant>
   </pollEnrich>
   <to uri="direct:result"/>
</route>

从 Camel 2.16 开始，enrich （） 和 pollEnrich （） 方法支持根据当前交换的信息计算使用动态 URI。例如，要从使用标头来代表 SEDA 队列名称的端点中轮询增强，您可以这样做：

from("direct:start")
  .pollEnrich().simple("seda:${header.name}")
  .to("direct:result");

以下是 XML DSL 中的相同示例：

<route>
   <from uri="direct:start"/>
   <pollEnrich>
      <simple>seda${header.name}</simple>
   </pollEnrich>
   <to uri="direct:result"/>
</route>

pollEnrich DSL 命令支持以下选项：