svn commit: r1435219 - in /camel/trunk/camel-core/src: main/java/org/apache/camel/processor/aggregate/ test/java/org/apache/camel/processor/aggregator/

[davsclaus]

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");
             }
         };