camel-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From davscl...@apache.org
Subject [1/3] git commit: CAMEL-6744: Aggregator - Using groupExchanges should store them on body by default
Date Fri, 13 Sep 2013 12:53:15 GMT
Updated Branches:
  refs/heads/camel-2.12.x 78de11e39 -> ed6e772b0
  refs/heads/master 745a85abb -> 71b216526


CAMEL-6744: Aggregator - Using groupExchanges should store them on body by default


Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/f9760f7b
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/f9760f7b
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/f9760f7b

Branch: refs/heads/master
Commit: f9760f7bf518cecfc74c0aa17a3e66ecdaf366a5
Parents: 745a85a
Author: Claus Ibsen <davsclaus@apache.org>
Authored: Fri Sep 13 13:21:31 2013 +0200
Committer: Claus Ibsen <davsclaus@apache.org>
Committed: Fri Sep 13 13:21:31 2013 +0200

----------------------------------------------------------------------
 .../apache/camel/model/AggregateDefinition.java |  3 +-
 .../AbstractListAggregationStrategy.java        | 15 +++-
 .../GroupedExchangeAggregationStrategy.java     | 29 +++++--
 ...gregateGroupedExchangeBackwardsCompTest.java | 79 ++++++++++++++++++++
 .../AggregateGroupedExchangeBatchSizeTest.java  |  4 +-
 ...eGroupedExchangeMultipleCorrelationTest.java |  4 +-
 ...gregateGroupedExchangeSizePredicateTest.java |  4 +-
 .../AggregateGroupedExchangeSizeTest.java       |  4 +-
 .../AggregateGroupedExchangeTest.java           |  2 +-
 9 files changed, 126 insertions(+), 18 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/camel/blob/f9760f7b/camel-core/src/main/java/org/apache/camel/model/AggregateDefinition.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/model/AggregateDefinition.java b/camel-core/src/main/java/org/apache/camel/model/AggregateDefinition.java
index 0af2206..061ba21 100644
--- a/camel-core/src/main/java/org/apache/camel/model/AggregateDefinition.java
+++ b/camel-core/src/main/java/org/apache/camel/model/AggregateDefinition.java
@@ -775,8 +775,7 @@ public class AggregateDefinition extends ProcessorDefinition<AggregateDefinition
 
     /**
      * Enables grouped exchanges, so the aggregator will group all aggregated exchanges into
a single
-     * combined Exchange holding all the aggregated exchanges in a {@link java.util.List}
as a exchange
-     * property with the key {@link org.apache.camel.Exchange#GROUPED_EXCHANGE}.
+     * combined Exchange holding all the aggregated exchanges in a {@link java.util.List}.
      *
      * @return the builder
      */

http://git-wip-us.apache.org/repos/asf/camel/blob/f9760f7b/camel-core/src/main/java/org/apache/camel/processor/aggregate/AbstractListAggregationStrategy.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/processor/aggregate/AbstractListAggregationStrategy.java
b/camel-core/src/main/java/org/apache/camel/processor/aggregate/AbstractListAggregationStrategy.java
index a19bdbc..ae382dc 100644
--- a/camel-core/src/main/java/org/apache/camel/processor/aggregate/AbstractListAggregationStrategy.java
+++ b/camel-core/src/main/java/org/apache/camel/processor/aggregate/AbstractListAggregationStrategy.java
@@ -20,7 +20,6 @@ import java.util.ArrayList;
 import java.util.List;
 
 import org.apache.camel.Exchange;
-import org.apache.camel.impl.DefaultExchange;
 
 /**
  * Aggregate all exchanges into a {@link List} of values defined by the {@link #getValue(Exchange)}
call.
@@ -101,10 +100,22 @@ public abstract class AbstractListAggregationStrategy<V> implements
CompletionAw
     private List<V> getList(Exchange exchange) {
         List<V> list = exchange.getProperty(Exchange.GROUPED_EXCHANGE, List.class);
         if (list == null) {
-            list = new ArrayList<V>();
+            list = new GroupedExchangeList();
             exchange.setProperty(Exchange.GROUPED_EXCHANGE, list);
         }
         return list;
     }
 
+    /**
+     * A list to contains grouped {@link Exchange}s.
+     */
+    private static final class GroupedExchangeList extends ArrayList {
+
+        @Override
+        public String toString() {
+            // lets override toString so we don't write data for all the Exchanges by default
+            return "List<Exchange>(" + size() + " elements)";
+        }
+    }
+
 }

http://git-wip-us.apache.org/repos/asf/camel/blob/f9760f7b/camel-core/src/main/java/org/apache/camel/processor/aggregate/GroupedExchangeAggregationStrategy.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/processor/aggregate/GroupedExchangeAggregationStrategy.java
b/camel-core/src/main/java/org/apache/camel/processor/aggregate/GroupedExchangeAggregationStrategy.java
index 8a748b5..587746c 100644
--- a/camel-core/src/main/java/org/apache/camel/processor/aggregate/GroupedExchangeAggregationStrategy.java
+++ b/camel-core/src/main/java/org/apache/camel/processor/aggregate/GroupedExchangeAggregationStrategy.java
@@ -16,21 +16,40 @@
  */
 package org.apache.camel.processor.aggregate;
 
+import java.util.List;
+
 import org.apache.camel.Exchange;
 
 /**
  * Aggregate all exchanges into a single combined Exchange holding all the aggregated exchanges
- * in a {@link java.util.List} as a exchange property with the key
- * {@link org.apache.camel.Exchange#GROUPED_EXCHANGE}.
+ * in a {@link java.util.List<Exchange>} as the message body.
  *
  * @version 
  */
 public class GroupedExchangeAggregationStrategy extends AbstractListAggregationStrategy<Exchange>
{
 
     @Override
-    public boolean isStoreAsBodyOnCompletion() {
-        // keep the list as a property to be compatible with old behavior
-        return false;
+    @SuppressWarnings("unchecked")
+    public void onCompletion(Exchange exchange) {
+        if (isStoreAsBodyOnCompletion()) {
+            // lets be backwards compatible
+            // TODO: Remove this method in Camel 3.0
+            List list = (List) exchange.getProperty(Exchange.GROUPED_EXCHANGE);
+            if (list != null) {
+                exchange.getIn().setBody(list);
+            }
+        }
+    }
+
+    @Override
+    public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
+        Exchange answer = super.aggregate(oldExchange, newExchange);
+        if (oldExchange == null) {
+            // for the first time we must do a copy as the answer, so the outgoing
+            // exchange is not one of the grouped exchanges, as that causes a endless circular
reference
+            answer = answer.copy();
+        }
+        return answer;
     }
 
     @Override

http://git-wip-us.apache.org/repos/asf/camel/blob/f9760f7b/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateGroupedExchangeBackwardsCompTest.java
----------------------------------------------------------------------
diff --git a/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateGroupedExchangeBackwardsCompTest.java
b/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateGroupedExchangeBackwardsCompTest.java
new file mode 100644
index 0000000..c27d55f
--- /dev/null
+++ b/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateGroupedExchangeBackwardsCompTest.java
@@ -0,0 +1,79 @@
+/**
+ * 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.List;
+
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+
+/**
+ * Unit test for aggregate grouped exchanges.
+ */
+public class AggregateGroupedExchangeBackwardsCompTest extends ContextTestSupport {
+
+    @SuppressWarnings("unchecked")
+    public void testGrouped() throws Exception {
+        // START SNIPPET: e2
+        MockEndpoint result = getMockEndpoint("mock:result");
+
+        // we expect 1 messages since we group all we get in using the same correlation key
+        result.expectedMessageCount(1);
+
+        // then we sent all the message at once
+        template.sendBody("direct:start", "100");
+        template.sendBody("direct:start", "150");
+        template.sendBody("direct:start", "130");
+        template.sendBody("direct:start", "200");
+        template.sendBody("direct:start", "190");
+
+        assertMockEndpointsSatisfied();
+
+        Exchange out = result.getExchanges().get(0);
+        List<Exchange> grouped = out.getProperty(Exchange.GROUPED_EXCHANGE, List.class);
+
+        assertEquals(5, grouped.size());
+
+        assertEquals("100", grouped.get(0).getIn().getBody(String.class));
+        assertEquals("150", grouped.get(1).getIn().getBody(String.class));
+        assertEquals("130", grouped.get(2).getIn().getBody(String.class));
+        assertEquals("200", grouped.get(3).getIn().getBody(String.class));
+        assertEquals("190", grouped.get(4).getIn().getBody(String.class));
+        // END SNIPPET: e2
+    }
+
+    @Override
+    protected RouteBuilder createRouteBuilder() throws Exception {
+        return new RouteBuilder() {
+            public void configure() throws Exception {
+                // START SNIPPET: e1
+                // our route is aggregating from the direct queue and sending the response
to the mock
+                from("direct:start")
+                    // aggregate all using same expression
+                    .aggregate(constant(true))
+                    // wait for 0.5 seconds to aggregate
+                    .completionTimeout(500L)
+                    // group the exchanges so we get one single exchange containing all the
others
+                    .groupExchanges()
+                    .to("mock:result");
+                // END SNIPPET: e1
+            }
+        };
+    }
+}

http://git-wip-us.apache.org/repos/asf/camel/blob/f9760f7b/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateGroupedExchangeBatchSizeTest.java
----------------------------------------------------------------------
diff --git a/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateGroupedExchangeBatchSizeTest.java
b/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateGroupedExchangeBatchSizeTest.java
index 31e0912..f958547 100644
--- a/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateGroupedExchangeBatchSizeTest.java
+++ b/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateGroupedExchangeBatchSizeTest.java
@@ -44,7 +44,7 @@ public class AggregateGroupedExchangeBatchSizeTest extends ContextTestSupport
{
         assertMockEndpointsSatisfied();
 
         Exchange out = result.getExchanges().get(0);
-        List<Exchange> grouped = out.getProperty(Exchange.GROUPED_EXCHANGE, List.class);
+        List<Exchange> grouped = out.getIn().getBody(List.class);
 
         assertTrue("Should be either 2 or 4, was " + grouped.size(), grouped.size() == 2
|| grouped.size() == 4);
 
@@ -57,7 +57,7 @@ public class AggregateGroupedExchangeBatchSizeTest extends ContextTestSupport
{
         if (result.getReceivedCounter() == 2) {
 
             out = result.getExchanges().get(1);
-            grouped = out.getProperty(Exchange.GROUPED_EXCHANGE, List.class);
+            grouped = out.getIn().getBody(List.class);
 
             assertEquals(2, grouped.size());
 

http://git-wip-us.apache.org/repos/asf/camel/blob/f9760f7b/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateGroupedExchangeMultipleCorrelationTest.java
----------------------------------------------------------------------
diff --git a/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateGroupedExchangeMultipleCorrelationTest.java
b/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateGroupedExchangeMultipleCorrelationTest.java
index 2f0b329..b1a10b7 100644
--- a/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateGroupedExchangeMultipleCorrelationTest.java
+++ b/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateGroupedExchangeMultipleCorrelationTest.java
@@ -50,7 +50,7 @@ public class AggregateGroupedExchangeMultipleCorrelationTest extends ContextTest
         assertMockEndpointsSatisfied();
 
         Exchange out = result.getExchanges().get(0);
-        List<Exchange> grouped = out.getProperty(Exchange.GROUPED_EXCHANGE, List.class);
+        List<Exchange> grouped = out.getIn().getBody(List.class);
 
         assertEquals(3, grouped.size());
 
@@ -60,7 +60,7 @@ public class AggregateGroupedExchangeMultipleCorrelationTest extends ContextTest
         assertEquals("180", grouped.get(2).getIn().getBody(String.class));
 
         out = result.getExchanges().get(1);
-        grouped = out.getProperty(Exchange.GROUPED_EXCHANGE, List.class);
+        grouped = out.getIn().getBody(List.class);
 
         assertEquals(3, grouped.size());
 

http://git-wip-us.apache.org/repos/asf/camel/blob/f9760f7b/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateGroupedExchangeSizePredicateTest.java
----------------------------------------------------------------------
diff --git a/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateGroupedExchangeSizePredicateTest.java
b/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateGroupedExchangeSizePredicateTest.java
index 701cdb9..9e386fb 100644
--- a/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateGroupedExchangeSizePredicateTest.java
+++ b/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateGroupedExchangeSizePredicateTest.java
@@ -44,14 +44,14 @@ public class AggregateGroupedExchangeSizePredicateTest extends ContextTestSuppor
         assertMockEndpointsSatisfied();
 
         Exchange out = result.getExchanges().get(0);
-        List<Exchange> grouped = out.getProperty(Exchange.GROUPED_EXCHANGE, List.class);
+        List<Exchange> grouped = out.getIn().getBody(List.class);
         assertEquals(3, grouped.size());
         assertEquals("100", grouped.get(0).getIn().getBody(String.class));
         assertEquals("150", grouped.get(1).getIn().getBody(String.class));
         assertEquals("130", grouped.get(2).getIn().getBody(String.class));
 
         out = result.getExchanges().get(1);
-        grouped = out.getProperty(Exchange.GROUPED_EXCHANGE, List.class);
+        grouped = out.getIn().getBody(List.class);
         assertEquals(2, grouped.size());
 
         assertEquals("200", grouped.get(0).getIn().getBody(String.class));

http://git-wip-us.apache.org/repos/asf/camel/blob/f9760f7b/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateGroupedExchangeSizeTest.java
----------------------------------------------------------------------
diff --git a/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateGroupedExchangeSizeTest.java
b/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateGroupedExchangeSizeTest.java
index 346dcbc..ca8911a 100644
--- a/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateGroupedExchangeSizeTest.java
+++ b/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateGroupedExchangeSizeTest.java
@@ -45,14 +45,14 @@ public class AggregateGroupedExchangeSizeTest extends ContextTestSupport
{
         assertMockEndpointsSatisfied();
 
         Exchange out = result.getExchanges().get(0);
-        List<Exchange> grouped = out.getProperty(Exchange.GROUPED_EXCHANGE, List.class);
+        List<Exchange> grouped = out.getIn().getBody(List.class);
         assertEquals(3, grouped.size());
         assertEquals("100", grouped.get(0).getIn().getBody(String.class));
         assertEquals("150", grouped.get(1).getIn().getBody(String.class));
         assertEquals("130", grouped.get(2).getIn().getBody(String.class));
 
         out = result.getExchanges().get(1);
-        grouped = out.getProperty(Exchange.GROUPED_EXCHANGE, List.class);
+        grouped = out.getIn().getBody(List.class);
         assertEquals(3, grouped.size());
 
         assertEquals("200", grouped.get(0).getIn().getBody(String.class));

http://git-wip-us.apache.org/repos/asf/camel/blob/f9760f7b/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateGroupedExchangeTest.java
----------------------------------------------------------------------
diff --git a/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateGroupedExchangeTest.java
b/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateGroupedExchangeTest.java
index 18be5d3..4628462 100644
--- a/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateGroupedExchangeTest.java
+++ b/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateGroupedExchangeTest.java
@@ -46,7 +46,7 @@ public class AggregateGroupedExchangeTest extends ContextTestSupport {
         assertMockEndpointsSatisfied();
 
         Exchange out = result.getExchanges().get(0);
-        List<Exchange> grouped = out.getProperty(Exchange.GROUPED_EXCHANGE, List.class);
+        List<Exchange> grouped = out.getIn().getBody(List.class);
 
         assertEquals(5, grouped.size());
 


Mime
View raw message