Author: ningjiang Date: Wed Jun 9 10:55:47 2010 New Revision: 952952 URL: http://svn.apache.org/viewvc?rev=952952&view=rev Log: CAMEL-2798 Fixed the NPE of aggregator
Added: camel/trunk/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateExpressionTest.java (with props) Modified: camel/trunk/camel-core/src/main/java/org/apache/camel/processor/aggregate/AggregateProcessor.java Modified: camel/trunk/camel-core/src/main/java/org/apache/camel/processor/aggregate/AggregateProcessor.java URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/processor/aggregate/AggregateProcessor.java?rev=952952&r1=952951&r2=952952&view=diff ============================================================================== --- camel/trunk/camel-core/src/main/java/org/apache/camel/processor/aggregate/AggregateProcessor.java (original) +++ camel/trunk/camel-core/src/main/java/org/apache/camel/processor/aggregate/AggregateProcessor.java Wed Jun 9 10:55:47 2010 @@ -244,8 +244,12 @@ public class AggregateProcessor extends } else { // if batch consumer completion is enabled then we need to complete the group if ("consumer".equals(complete)) { - for (String batchKey : batchConsumerCorrelationKeys) { + for (String batchKey : batchConsumerCorrelationKeys) { Exchange batchAnswer = aggregationRepository.get(camelContext, batchKey); + // There is no aggregated exchange + if (batchAnswer == null) { + batchAnswer = answer; + } batchAnswer.setProperty(Exchange.AGGREGATED_COMPLETED_BY, complete); onCompletion(batchKey, batchAnswer, false); } @@ -318,7 +322,7 @@ public class AggregateProcessor extends timeoutMap.put(key, exchange.getExchangeId(), getCompletionTimeout()); } - if (isCompletionFromBatchConsumer()) { + if (isCompletionFromBatchConsumer()) { batchConsumerCorrelationKeys.add(key); batchConsumerCounter.incrementAndGet(); int size = exchange.getProperty(Exchange.BATCH_SIZE, 0, Integer.class); Added: camel/trunk/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateExpressionTest.java URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateExpressionTest.java?rev=952952&view=auto ============================================================================== --- camel/trunk/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateExpressionTest.java (added) +++ camel/trunk/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateExpressionTest.java Wed Jun 9 10:55:47 2010 @@ -0,0 +1,58 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.processor.aggregator; + +import java.util.HashMap; +import java.util.Map; + +import org.apache.camel.ContextTestSupport; +import org.apache.camel.Exchange; +import org.apache.camel.Processor; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.component.mock.MockEndpoint; +import org.apache.camel.processor.BodyInAggregatingStrategy; + +public class AggregateExpressionTest extends ContextTestSupport { + + public void testAggregateExpressionSize() throws Exception { + MockEndpoint result = getMockEndpoint("mock:result"); + result.expectedBodiesReceived("A+A", "B+B", "Z"); + + template.sendBody("direct:start", "A"); + template.sendBody("direct:start", "B"); + template.sendBody("direct:start", "A"); + template.sendBody("direct:start", "B"); + // send the last one with the batch size property + template.sendBodyAndProperty("direct:start", "Z", Exchange.BATCH_SIZE, 5); + + assertMockEndpointsSatisfied(); + } + + @Override + protected RouteBuilder createRouteBuilder() throws Exception { + return new RouteBuilder() { + @Override + public void configure() throws Exception { + // START SNIPPET: e1 + from("direct:start") + .aggregate(body(), new BodyInAggregatingStrategy()).completionFromBatchConsumer() + .to("mock:result"); + // END SNIPPET: e1 + } + }; + } +} Propchange: camel/trunk/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateExpressionTest.java ------------------------------------------------------------------------------ svn:eol-style = native Propchange: camel/trunk/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateExpressionTest.java ------------------------------------------------------------------------------ svn:keywords = Rev Date