第 10 章 消息转换

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

图 10.1. 内容增强模式

Apache Camel 支持多种增强内容的方法：

Camel 提供了 流畅的构建器，以使用提供智能完成且安全重构安全的类型安全 IDE 创建路由和调解规则。当您测试分布式系统时，需要存根某些外部系统，以便测试系统的其他部分，直到特定的系统可用或写出。其中一种实现方式是使用某种 模板 系统通过生成具有 mostly-static 正文的动态消息来生成对请求的响应。使用模板的另一种方法是使用来自一个目的地的消息，使用 Velocity 或 XQuery 等内容进行转换，然后将其发送到另一个目的地。以下示例显示了 InOnly (one way)信息：

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

假设您想要使用 InOut (request-reply)消息传递来处理 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 中，仅省略 policyRef 属性，如下所示：

<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 （） 方法的资源交换可能为 null。

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>

如果指定的文件不存在，则消息为空。您可以指定等待的超时时间（可能为ever），直到文件存在或等待到特定时间长度。在以下示例中，命令会等待不超过 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 命令支持以下选项：