You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by da...@apache.org on 2013/01/18 17:12:00 UTC
svn commit: r1435219 - in /camel/trunk/camel-core/src:
main/java/org/apache/camel/processor/aggregate/
test/java/org/apache/camel/processor/aggregator/
Author: davsclaus
Date: Fri Jan 18 16:11:59 2013
New Revision: 1435219
URL: http://svn.apache.org/viewvc?rev=1435219&view=rev
Log:
CAMEL-5957: Fixed aggregator should fallback and do timeout completion check if size was done but not completed.
Added:
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateExpressionSizeOverrideFixedTimeoutTest.java
- copied, changed from r1435197, camel/trunk/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateExpressionSizeOverrideFixedTest.java
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/aggregate/AggregateProcessor.java
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateExpressionSizeOverrideFixedTest.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=1435219&r1=1435218&r2=1435219&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 Fri Jan 18 16:11:59 2013
@@ -307,19 +307,19 @@ public class AggregateProcessor extends
}
}
+ boolean sizeChecked = false;
if (getCompletionSizeExpression() != null) {
Integer value = getCompletionSizeExpression().evaluate(exchange, Integer.class);
if (value != null && value > 0) {
+ // mark as already checked size as expression takes precedence over static configured
+ sizeChecked = true;
int size = exchange.getProperty(Exchange.AGGREGATED_SIZE, 1, Integer.class);
if (size >= value) {
return "size";
- } else {
- // not completed yet
- return null;
}
}
}
- if (getCompletionSize() > 0) {
+ if (!sizeChecked && getCompletionSize() > 0) {
int size = exchange.getProperty(Exchange.AGGREGATED_SIZE, 1, Integer.class);
if (size >= getCompletionSize()) {
return "size";
Modified: camel/trunk/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateExpressionSizeOverrideFixedTest.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateExpressionSizeOverrideFixedTest.java?rev=1435219&r1=1435218&r2=1435219&view=diff
==============================================================================
--- camel/trunk/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateExpressionSizeOverrideFixedTest.java (original)
+++ camel/trunk/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateExpressionSizeOverrideFixedTest.java Fri Jan 18 16:11:59 2013
@@ -20,6 +20,7 @@ import java.util.HashMap;
import java.util.Map;
import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Exchange;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.processor.BodyInAggregatingStrategy;
@@ -30,6 +31,7 @@ public class AggregateExpressionSizeOver
public void testAggregateExpressionSize() throws Exception {
getMockEndpoint("mock:aggregated").expectedBodiesReceived("A+B+C");
+ getMockEndpoint("mock:aggregated").expectedPropertyReceived(Exchange.AGGREGATED_COMPLETED_BY, "size");
Map<String, Object> headers = new HashMap<String, Object>();
headers.put("id", 123);
Copied: camel/trunk/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateExpressionSizeOverrideFixedTimeoutTest.java (from r1435197, camel/trunk/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateExpressionSizeOverrideFixedTest.java)
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateExpressionSizeOverrideFixedTimeoutTest.java?p2=camel/trunk/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateExpressionSizeOverrideFixedTimeoutTest.java&p1=camel/trunk/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateExpressionSizeOverrideFixedTest.java&r1=1435197&r2=1435219&rev=1435219&view=diff
==============================================================================
--- camel/trunk/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateExpressionSizeOverrideFixedTest.java (original)
+++ camel/trunk/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateExpressionSizeOverrideFixedTimeoutTest.java Fri Jan 18 16:11:59 2013
@@ -20,25 +20,29 @@ import java.util.HashMap;
import java.util.Map;
import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Exchange;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.processor.BodyInAggregatingStrategy;
/**
* @version
*/
-public class AggregateExpressionSizeOverrideFixedTest extends ContextTestSupport {
+public class AggregateExpressionSizeOverrideFixedTimeoutTest extends ContextTestSupport {
public void testAggregateExpressionSize() throws Exception {
getMockEndpoint("mock:aggregated").expectedBodiesReceived("A+B+C");
+ getMockEndpoint("mock:aggregated").expectedPropertyReceived(Exchange.AGGREGATED_COMPLETED_BY, "timeout");
Map<String, Object> headers = new HashMap<String, Object>();
headers.put("id", 123);
- headers.put("mySize", 3);
+ headers.put("mySize", 4);
template.sendBodyAndHeaders("direct:start", "A", headers);
template.sendBodyAndHeaders("direct:start", "B", headers);
template.sendBodyAndHeaders("direct:start", "C", headers);
+ // do not send the 4th as we want to test for timeout then
+
assertMockEndpointsSatisfied();
}
@@ -50,6 +54,7 @@ public class AggregateExpressionSizeOver
from("direct:start")
.aggregate(header("id"), new BodyInAggregatingStrategy())
.completionSize(2).completionSize(header("mySize"))
+ .completionTimeout(1000)
.to("mock:aggregated");
}
};