阿帕奇骆驼——如何同步“窃听”？还是只发一份交换？

Question

我有一个apache骆驼路由，它正在交换主体上处理POJO。

请看从1到3标记的行序列。

    from("direct:foo")
        .to("direct:doSomething")         // 1 (POJO on the exchange body)
        .to("direct:storeInHazelcast")    // 2 (destroys my pojo! it gets -1)
        .to("direct:doSomethingElse")     // 3 (Where is my POJO??)
    ;

现在，我需要对< code>hazelcast组件使用< code>put操作，不幸的是，该组件需要将body设置为值-1。

    from("direct:storeInHazelcast")
            .setBody(constant(-1))
            .setHeader(HazelcastConstants.OPERATION, constant(HazelcastConstants.PUT_OPERATION))
            .setHeader(HazelcastConstants.OBJECT_ID, constant(LAST_FLIGHT_UPDATE_SEQ))
            .to("hazelcast:map:MyNumber")
    ;

对于标记为 2 的行，我想将交换的副本发送到商店InHazelcast 路由。

首先，我尝试了. Multicast（），但交换体仍然被搞砸了（到-1）。

        // shouldnt this copy the exchange?
        .multicast().to("direct:storeInHazelcast").end()

然后我尝试。wireTap（），它是一种“fire-and-forget”（异步）模式，但我实际上需要它来阻止，并等待它完成。你会制作wireTap块吗？

        // this works but I need it to be sync processing (not async)
        .wireTap("direct:storeInHazelcast").end()

所以我在这里寻找一些提示。据我所知，< code>multicast()应该已经复制了交换，但是我的< code>storeInHazelcast路线中的< code>setBody()似乎破坏了原始交换。

或者，也许有其他方法可以做到这一点。

先谢了。骆驼2.10

Answer 1

将其保存在标题中并恢复。

from("direct:foo")
    .to("direct:doSomething")         // 1 (POJO on the exchange body)
    .setHeader("old_body", body())    // save body
    .to("direct:storeInHazelcast")    // 2 (destroys my pojo! it gets -1)
    .setBody(header("old_body"))      // RESTORE the body
    .removeHeader("old_body")         // cleanup header
    .to("direct:doSomethingElse")     // 3 (Where is my POJO??)
;

对于破坏性组件，这是一种相当常见的范式。

Answer 2

无需编写自己的聚合策略即可使用

.enrich("direct:storeInHazelcast", AggregationStrategies.useOriginal())

Answer 3

我想我已经偶然发现了答案，第 2 行可以使用 dsl 中的 enrich（），如下所示：

    .enrich("direct:storeInHazelcast", new KeepOriginalAggregationStrategy())

其中:

public class KeepOriginalAggregationStrategy implements AggregationStrategy {
    @Override
    public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
        return oldExchange;
    }
}

有趣的是，我发现了一个名为< code > UseOriginalAggregationStrategy()的聚合策略，但是我看不到如何从DSL中指定名为< code>Exchange original的参数。

    .enrich("direct:storeInHazelcast",
        new UseOriginalAggregationStrategy(???, false))

在dsl中缺少某种< code>getExchange()方法的情况下，我看不出如何在这里使用这种聚合策略(但是如果有人能提供建议，请提供)。