Apache骆驼 - 如何同步"窃听"？或者只是发送一份交易所的副本？

Question

我有一个apache camel路线,正在交换机体上处理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??)
    ;

现在我需要put在hazelcast组件上使用操作,遗憾的是需要将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的行,我想将该交换的COPY发送给该storeInHazelcast路径.

首先,我试过.multicast(),但交换机构仍然搞砸了(-1).

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

然后我尝试了.wireTap(),它可以作为"火灾和忘记"(异步)模式,但实际上我需要它来阻止,并等待它完成.你能制作wireTap块吗？

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

所以我在这里寻找一些提示.据我所知,multicast()应该复制交换,但setBody()在我的storeInHazelcast路线中查看搞砸了原来的交换.

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

提前致谢.骆驼2.10

Answer 1

我想我已经偶然发现了答案,行2可以使用enrich()dsl这样:

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

哪里:

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

有趣的是,我找到了一个名为的聚合策略UseOriginalAggregationStrategy(),但我看不到如何指定Exchange original从DSL 命名的参数.

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

在getExchange()dsl中没有某种方法的情况下,我无法在这里看到如何使用这种聚合策略(但如果有人可以建议如何,请做).

Answer 2

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

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