You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by dk...@apache.org on 2011/09/19 22:05:12 UTC
svn commit: r1172784 - in /camel/branches/camel-2.8.x: ./
camel-core/src/main/java/org/apache/camel/
camel-core/src/main/java/org/apache/camel/processor/aggregate/
camel-core/src/test/java/org/apache/camel/processor/aggregator/
components/camel-hawtdb/...
Author: dkulp
Date: Mon Sep 19 20:05:12 2011
New Revision: 1172784
URL: http://svn.apache.org/viewvc?rev=1172784&view=rev
Log:
Merged revisions 1153737 via svnmerge from
https://svn.apache.org/repos/asf/camel/trunk
........
r1153737 | boday | 2011-08-03 23:30:08 -0400 (Wed, 03 Aug 2011) | 1 line
CAMEL-4118 added support for manually completing all message groups with a signal message
........
Added:
camel/branches/camel-2.8.x/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateForceCompletionHeaderTest.java
- copied unchanged from r1153737, camel/trunk/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateForceCompletionHeaderTest.java
camel/branches/camel-2.8.x/components/camel-hawtdb/src/test/java/org/apache/camel/component/hawtdb/HawtDBAggregateForceCompletionHeaderTest.java
- copied unchanged from r1153737, camel/trunk/components/camel-hawtdb/src/test/java/org/apache/camel/component/hawtdb/HawtDBAggregateForceCompletionHeaderTest.java
camel/branches/camel-2.8.x/components/camel-sql/src/test/java/org/apache/camel/processor/aggregate/jdbc/JdbcAggregateForceCompletionHeaderTest.java
- copied unchanged from r1153737, camel/trunk/components/camel-sql/src/test/java/org/apache/camel/processor/aggregate/jdbc/JdbcAggregateForceCompletionHeaderTest.java
Modified:
camel/branches/camel-2.8.x/ (props changed)
camel/branches/camel-2.8.x/camel-core/src/main/java/org/apache/camel/Exchange.java
camel/branches/camel-2.8.x/camel-core/src/main/java/org/apache/camel/processor/aggregate/AggregateProcessor.java
camel/branches/camel-2.8.x/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateProcessorTest.java
Propchange: camel/branches/camel-2.8.x/
------------------------------------------------------------------------------
Binary property 'svnmerge-integrated' - no diff available.
Modified: camel/branches/camel-2.8.x/camel-core/src/main/java/org/apache/camel/Exchange.java
URL: http://svn.apache.org/viewvc/camel/branches/camel-2.8.x/camel-core/src/main/java/org/apache/camel/Exchange.java?rev=1172784&r1=1172783&r2=1172784&view=diff
==============================================================================
--- camel/branches/camel-2.8.x/camel-core/src/main/java/org/apache/camel/Exchange.java (original)
+++ camel/branches/camel-2.8.x/camel-core/src/main/java/org/apache/camel/Exchange.java Mon Sep 19 20:05:12 2011
@@ -78,6 +78,7 @@ public interface Exchange {
String AGGREGATED_COMPLETED_BY = "CamelAggregatedCompletedBy";
String AGGREGATED_CORRELATION_KEY = "CamelAggregatedCorrelationKey";
String AGGREGATION_STRATEGY = "CamelAggregationStrategy";
+ String AGGREGATION_COMPLETE_ALL_GROUPS = "CamelAggregationCompleteAllGroups";
String ASYNC_WAIT = "CamelAsyncWait";
String BATCH_INDEX = "CamelBatchIndex";
Modified: camel/branches/camel-2.8.x/camel-core/src/main/java/org/apache/camel/processor/aggregate/AggregateProcessor.java
URL: http://svn.apache.org/viewvc/camel/branches/camel-2.8.x/camel-core/src/main/java/org/apache/camel/processor/aggregate/AggregateProcessor.java?rev=1172784&r1=1172783&r2=1172784&view=diff
==============================================================================
--- camel/branches/camel-2.8.x/camel-core/src/main/java/org/apache/camel/processor/aggregate/AggregateProcessor.java (original)
+++ camel/branches/camel-2.8.x/camel-core/src/main/java/org/apache/camel/processor/aggregate/AggregateProcessor.java Mon Sep 19 20:05:12 2011
@@ -157,6 +157,14 @@ public class AggregateProcessor extends
}
public void process(Exchange exchange) throws Exception {
+
+ //check for the special header to force completion of all groups (and ignore the exchange otherwise)
+ boolean completeAllGroups = exchange.getIn().getHeader(Exchange.AGGREGATION_COMPLETE_ALL_GROUPS, false, boolean.class);
+ if (completeAllGroups) {
+ forceCompletionOfAllGroups();
+ return;
+ }
+
// compute correlation expression
String key = correlationExpression.evaluate(exchange, String.class);
if (ObjectHelper.isEmpty(key)) {
@@ -879,4 +887,37 @@ public class AggregateProcessor extends
super.doShutdown();
}
+ public void forceCompletionOfAllGroups() {
+
+ // only run if CamelContext has been fully started
+ if (!camelContext.getStatus().isStarted()) {
+ LOG.warn("cannot start force completion because CamelContext({}) has not been started yet", camelContext.getName());
+ return;
+ }
+
+ LOG.trace("Starting force completion of all groups task");
+
+ // trigger completion for all in the repository
+ Set<String> keys = aggregationRepository.getKeys();
+
+ if (keys != null && !keys.isEmpty()) {
+ // must acquire the shared aggregation lock to be able to trigger force completion
+ lock.lock();
+ try {
+ for (String key : keys) {
+ Exchange exchange = aggregationRepository.get(camelContext, key);
+ if (exchange != null) {
+ LOG.trace("force completion triggered for correlation key: {}", key);
+ // indicate it was completed by a force completion request
+ exchange.setProperty(Exchange.AGGREGATED_COMPLETED_BY, "forceCompletion");
+ onCompletion(key, exchange, false);
+ }
+ }
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ LOG.trace("Completed force completion of all groups task");
+ }
}
Modified: camel/branches/camel-2.8.x/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateProcessorTest.java
URL: http://svn.apache.org/viewvc/camel/branches/camel-2.8.x/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateProcessorTest.java?rev=1172784&r1=1172783&r2=1172784&view=diff
==============================================================================
--- camel/branches/camel-2.8.x/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateProcessorTest.java (original)
+++ camel/branches/camel-2.8.x/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateProcessorTest.java Mon Sep 19 20:05:12 2011
@@ -559,4 +559,50 @@ public class AggregateProcessorTest exte
ap.stop();
}
+ public void testAggregateForceCompletion() throws Exception {
+ // camel context must be started
+ context.start();
+
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceived("B+END", "A+END");
+ mock.expectedPropertyReceived(Exchange.AGGREGATED_COMPLETED_BY, "forceCompletion");
+
+ Processor done = new SendProcessor(context.getEndpoint("mock:result"));
+ Expression corr = header("id");
+ AggregationStrategy as = new BodyInAggregatingStrategy();
+
+ AggregateProcessor ap = new AggregateProcessor(context, done, corr, as, executorService);
+ ap.setCompletionSize(10);
+ ap.start();
+
+ Exchange e1 = new DefaultExchange(context);
+ e1.getIn().setBody("A");
+ e1.getIn().setHeader("id", 123);
+
+ Exchange e2 = new DefaultExchange(context);
+ e2.getIn().setBody("B");
+ e2.getIn().setHeader("id", 456);
+
+ Exchange e3 = new DefaultExchange(context);
+ e3.getIn().setBody("END");
+ e3.getIn().setHeader("id", 123);
+
+ Exchange e4 = new DefaultExchange(context);
+ e4.getIn().setBody("END");
+ e4.getIn().setHeader("id", 456);
+
+ ap.process(e1);
+ ap.process(e2);
+ ap.process(e3);
+ ap.process(e4);
+
+ assertEquals("should not have completed yet", 0, mock.getExchanges().size());
+
+ ap.forceCompletionOfAllGroups();
+
+ assertMockEndpointsSatisfied();
+
+ ap.stop();
+ }
+
}