如何在AggregationStrategy中将消息正文转换为pojo

Question

我有一条路由（Camel 2.17.3），它使用丰富的 DSL 调用Rest服务并将结果聚合到消息正文中。不过，我遇到了序列化问题。这是我正在尝试的。我的路线如下所示：

rest("myService").produces("application/json")
            .get("test")
            .param().name("text").required(true).type(RestParamType.query).description("The text to be processed.").endParam()
                .to("direct:StepA")
    ;

    from("direct:StepA")
            .removeHeader("CamelHttpQuery")
            .removeHeader("CamelHttpRawQuery")
            .enrich("netty4-http://myOtherService:8080/endpoint?input=${header.text}", new MyAggregator())
            .to("direct:StepB")
    ;

    from("direct:StepB")
            // not really implemented yet
            .transform().simple("Query: ${header.text}\nBody: ${body}")
    ;

正如您所看到的，我想使用enrich（）DSL调用一些现有服务，并将这些结果聚合起来，形成一个新的消息体，以便进一步处理。我需要将rest调用的结果从Json转换为MyResponse。我想用这个：

.unmarshal().json(JsonLibrary.Jackson, MyResponse.class)

但我需要它已经在我的AggregationStrategy中解组，如下所示:

public class MyAggregator implements AggregationStrategy {
    @Override
    public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
        final MyResponse intermediateResponse = oldExchange.getIn().getBody(MyResponse.class);

        final MyAggregate response = new MyAggregate(oldExchange.getIn().getHeader("text", String.class), intermediateResponse);

        newExchange.getIn().setBody(response);
        return newExchange;
}

然而，当我说getBody(..).所以我学习了TypeConverters，并试着用它来让系统自动转换成我的类型:

public class MyConverter extends TypeConverterSupport {

    private static final ObjectMapper MAPPER =
        new ObjectMapper().enable(SerializationFeature.INDENT_OUTPUT);

    @Override
    public <T> T convertTo(Class<T> type, Exchange exchange, Object value) throws TypeConversionException {
        try {
            return (T)MAPPER.readValue(value.toString(), MyResponse.class);
        } catch (IOException e) {
           throw new TypeConversionException(value, type, e);
        }
}

然后在我的主要:

    public static void main(String... args) throws Exception {
        Main main = new Main();
        main.setApplicationContextUri("classpath:app-ctx.xml");

        for (final CamelContext ctx : main.getCamelContexts()) {
           ctx.getTypeConverterRegistry().addTypeConverter(MyResponse.class, String.class, new MyConverter());
        }

        main.run(args);
    }

但这似乎不会改变结果。在我的聚合器中，MyResponse仍然为空。我不知道如何设置这种类型转换，希望有人能给我指出正确的方向。

Answer 1

好的，当我意识到我可以丰富一个执行unmarshall 调用的子路由，然后聚合它时，我最终解决了这个问题......

from("direct:StepA")
        .removeHeader("CamelHttpQuery")
        .removeHeader("CamelHttpRawQuery")
        .enrich("direct:StepAEnricher", new MyAggregator())
        .to("direct:StepB")
;

from("direct:StepAEnricher")
    .toD("netty4-http://myOtherService:8080/endpoint?input=${header.text}")
    .unmarshal().json(JsonLibrary.Jackson, MyResponse.class)
;

这非常有效！我很好奇一个经验丰富的骆驼开发人员会如何看待这一点。使用这种“额外”的路由来处理这样的响应是否代价高昂？我仍然很好奇如何克服自定义类型转换器方法。