0

我目前有以下骆驼路线：

<camelContext id="my-camel-context" xmlns="http://camel.apache.org/schema/spring">
    <propertyPlaceholder id="envProps" location="classpath:myapp.properties" />
    <route id="my-camel-route"> 
        <from uri="{{start.uri}}"/>

        <setHeader headerName="id">
            <constant>1</constant>
        </setHeader>

        <to uri="bean:preProcessor?method=process" />

        <aggregate strategyRef="myAggregationStrategy" completionSize="1">
            <correlationExpression> 
                <simple>${header.id} == 1</simple> 
            </correlationExpression>
            <to uri="bean:postProcessor?method=process" /> 
        </aggregate> 

        <to uri="bean:mailer?method=process" /> 
    </route> 
</camelContext>

<bean id="myAggregationStrategy" class="com.me.myapp.MyAggregationStrategy" />
<bean id="postProcessor" class="com.me.myapp.PostProcessor" />
<bean id="mailer" class="com.me.myapp.Mailer" />

现在，我并没有真正聚合任何有意义的东西（completionSize=1），我只是在测试AggregationStrategy。这是我的策略：

public class MyAggregationStrategy implements AggregationStrategy {
    @Override
    public Exchange aggregate(Exchange aggregatingExchange, Exchange incomingExchange) {
        AppPayload payload = null;

        if(aggregatingExchange == null)
            payload = new AppPayload(); // This should prevent it from being NULL below in PostProcessor...
        else
            payload = (AppPayload)incomingExchange.getIn().getBody();

        payload.setCargo((Order)incomingExchange.getIn().getBody());

        if(aggregatingExchange == null) {
            incomingExchange.getIn().setBody(payload);
            return incomingExchange;
        }
        else
            return aggregatingExchange;
    }
}

还有我的postProcessor豆子：

public class PostProcessor implement Processor {
    @Override
    public void process(Exchange exchange) {
        try {
            System.out.println("In PostProcessor...");
            AppPayload payload = (AppPayload)exchange.getIn().getBody();
            System.out.println("\t...payload acquired...");

            if(payload == null)
                System.out.println("Payload is NULL.");
        } catch(Throwable throwable) {
            System.out.println(ExceptionUtils.getFullStackTrace(throwable));
        }
    }
}

当我运行此代码时，我看到来自我的preProcessorbean 的日志消息表明它正在正确执行。而且我还看到这MyAggregationStrategy是正确地“聚合”消息，然后让它postProcessor在第一条消息到达后传递（再次，因为completionSize=1）。但是，我得到以下输出postProcessor：

In PostProcessor...
    ...payload acquired...
Payload is NULL.

谁能看到为什么payload会是NULL？它不应该已经在里面初始化了MyAggregationStrategy吗？！？我很高兴发布更多代码，但我相信这源于我AggregationStrategy错误地使用了 API。

4

2 回答 2

1

我相信您对aggregatingExchangeand感到困惑incomingExchange。你可以试试这个：

public class MyAggregationStrategy implements AggregationStrategy {
    @Override
    public Exchange aggregate(Exchange aggregatingExchange, Exchange incomingExchange) {
        AppPayload payload = null;

        if(aggregatingExchange == null) {
        payload = new AppPayload(); // This should prevent it from being NULL below in PostProcessor...
        } else {
            payload = (AppPayload)aggregatingExchange.getIn().getBody();
        }

        payload.setCargo((Order)incomingExchange.getIn().getBody());

        if(aggregatingExchange == null) {
            incomingExchange.getIn().setBody(payload);
            return incomingExchange;
        } else {
            return aggregatingExchange;
        }
    }
}
于 2014-01-28T16:31:41.300 回答
0

添加到@hveiga 已经提到的内容。我有一个类似的问题，我通过在我的消息中添加标题来解决。但是，在您的情况下，我看到您没有使用拆分器，并且您已经定义了标头。所以从克劳斯·易卜生那里得到的一条信息是第一次交换是空的，我们需要检查空对象。

有关更多说明，请参阅此内容 - Apache Camel - 拆分和聚合 - 旧 Exchange 始终为空

在此处跟踪完整的解释 - http://camel.465427.n5.nabble.com/Split-and-Aggregate-Old-Exchange-is-null-everytime-in-AggregationStrategy-td5746365.html

于 2014-01-29T18:43:01.080 回答