diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/ExpressionEvaluatingReleaseStrategy.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/ExpressionEvaluatingReleaseStrategy.java
index d28bce3f5b..60fc6b8680 100644
--- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/ExpressionEvaluatingReleaseStrategy.java
+++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/ExpressionEvaluatingReleaseStrategy.java
@@ -1,5 +1,5 @@
/*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-2017 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -16,26 +16,37 @@
package org.springframework.integration.aggregator;
+import org.springframework.expression.Expression;
import org.springframework.integration.store.MessageGroup;
+import org.springframework.integration.util.AbstractExpressionEvaluator;
+import org.springframework.util.Assert;
/**
* A {@link ReleaseStrategy} that evaluates an expression.
*
* @author Dave Syer
+ * @author Artem Bilan
*/
-public class ExpressionEvaluatingReleaseStrategy extends ExpressionEvaluatingMessageListProcessor implements
- ReleaseStrategy {
+public class ExpressionEvaluatingReleaseStrategy extends AbstractExpressionEvaluator implements ReleaseStrategy {
+
+ private final Expression expression;
public ExpressionEvaluatingReleaseStrategy(String expression) {
- super(expression, Boolean.class);
+ Assert.hasText(expression, "'expression' must not be empty");
+ this.expression = EXPRESSION_PARSER.parseExpression(expression);
+ }
+
+ public ExpressionEvaluatingReleaseStrategy(Expression expression) {
+ Assert.notNull(expression, "'expression' must not be null");
+ this.expression = expression;
}
/**
- * Evaluate the expression provided on the messages (a collection) in the group and return the result (must
- * be boolean).
+ * Evaluate the expression provided on the {@link MessageGroup}
+ * and return the result (must be boolean).
*/
public boolean canRelease(MessageGroup messages) {
- return (Boolean) process(messages.getMessages());
+ return evaluateExpression(this.expression, messages, Boolean.class);
}
}
diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/ExpressionEvaluatingReleaseStrategyTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/ExpressionEvaluatingReleaseStrategyTests.java
index bd82b5db11..ac4e1e6593 100644
--- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/ExpressionEvaluatingReleaseStrategyTests.java
+++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/ExpressionEvaluatingReleaseStrategyTests.java
@@ -1,5 +1,5 @@
/*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-2017 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -31,6 +31,7 @@ import org.springframework.integration.store.SimpleMessageGroup;
* @author Alex Peters
* @author Dave Syer
* @author Gary Russell
+ * @author Artem Bilan
*
*/
public class ExpressionEvaluatingReleaseStrategyTests {
@@ -56,14 +57,14 @@ public class ExpressionEvaluatingReleaseStrategyTests {
@Test
public void testCompletedWithFilterSpelEvaluated() throws Exception {
- strategy = new ExpressionEvaluatingReleaseStrategy("!?[payload==5].empty");
+ strategy = new ExpressionEvaluatingReleaseStrategy("!messages.?[payload==5].empty");
strategy.setBeanFactory(mock(BeanFactory.class));
assertThat(strategy.canRelease(messages), is(true));
}
@Test
public void testCompletedWithFilterSpelReturnsNotCompleted() throws Exception {
- strategy = new ExpressionEvaluatingReleaseStrategy("!?[payload==6].empty");
+ strategy = new ExpressionEvaluatingReleaseStrategy("!messages.?[payload==6].empty");
strategy.setBeanFactory(mock(BeanFactory.class));
assertThat(strategy.canRelease(messages), is(false));
}
diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/AggregatorIntegrationTests-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/AggregatorIntegrationTests-context.xml
index 0492fd04c1..48b4040ada 100644
--- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/AggregatorIntegrationTests-context.xml
+++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/AggregatorIntegrationTests-context.xml
@@ -43,13 +43,13 @@
+ release-strategy-expression="messages[0].headers.sequenceNumber == messages[0].headers.sequenceSize"/>
+ release-strategy-expression="messages[0].headers.sequenceNumber == messages[0].headers.sequenceSize"/>
diff --git a/spring-integration-core/src/test/java/org/springframework/integration/scattergather/config/ScatterGatherTests-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/scattergather/config/ScatterGatherTests-context.xml
index 8c6918c912..e50f4df0c1 100644
--- a/spring-integration-core/src/test/java/org/springframework/integration/scattergather/config/ScatterGatherTests-context.xml
+++ b/spring-integration-core/src/test/java/org/springframework/integration/scattergather/config/ScatterGatherTests-context.xml
@@ -12,7 +12,7 @@
-
+
@@ -34,7 +34,7 @@
-
+
@@ -55,7 +55,7 @@
-
+
diff --git a/src/reference/asciidoc/aggregator.adoc b/src/reference/asciidoc/aggregator.adoc
index c73ed7c7b6..3776f8f830 100644
--- a/src/reference/asciidoc/aggregator.adoc
+++ b/src/reference/asciidoc/aggregator.adoc
@@ -95,7 +95,7 @@ It creates a single Message whose payload is a List of the payloads received for
This works well for simple Scatter Gather implementations with either a Splitter, Publish Subscribe Channel, or Recipient List Router upstream.
NOTE: When using a Publish Subscribe Channel or Recipient List Router in this type of scenario, be sure to enable the flag to `apply-sequence`.
-That will add the necessary headers (CORRELATION_ID, SEQUENCE_NUMBER and SEQUENCE_SIZE).
+That will add the necessary headers (`CORRELATION_ID`, `SEQUENCE_NUMBER` and `SEQUENCE_SIZE`).
That behavior is enabled by default for Splitters in Spring Integration, but it is not enabled for the Publish Subscribe Channel or Recipient List Router because those components may be used in a variety of contexts in which these headers are not necessary.
When implementing a specific aggregator strategy for an application, a developer can extend `AbstractAggregatingMessageGroupProcessor` and implement the `aggregatePayloads` method.
@@ -443,7 +443,7 @@ _Optional, with restrictions (requires `release-strategy` to be present)._
-<17> A SpEL expression representing the release strategy; the root object for the expression is a `Collection` of `Message` s.
+<17> A SpEL expression representing the release strategy; the root object for the expression is a `MessageGroup`.
Example: `"size() == 5"`.
Only one of `release-strategy` or `release-strategy-expression` is allowed.
@@ -601,7 +601,8 @@ For example, this aggregator would group numbers by some criterion (in our case
NOTE: Wherever it makes sense, the release strategy method, correlation strategy method and the aggregator method can be combined in a single bean (all of them or any two).
-_Aggregators and Spring Expression Language (SpEL)_
+[[aggregator-spel]]
+====== Aggregators and Spring Expression Language (SpEL)
Since Spring Integration 2.0, the various strategies (correlation, release, and aggregation) may be handled with http://docs.spring.io/spring/docs/current/spring-framework-reference/html/expressions.html[SpEL] which is recommended if the logic behind such _release strategy_ is relatively simple.
Let's say you have a legacy component that was designed to receive an array of objects.
@@ -644,16 +645,18 @@ correlation-strategy-expression="payload.person.id"
In the above example it is assumed that the payload has an attribute `person` with an `id` which is going to be used to correlate messages.
Likewise, for the `ReleaseStrategy` you can implement your release logic as a SpEL expression and configure it via the `release-strategy-expression` attribute.
-The only difference is that since ReleaseStrategy is passed the List of Messages, the root object in the SpEL evaluation context is the List itself.
-That List can be referenced as `#this` within the expression.
+The root object for evaluation context is the `MessageGroup` itself.
+The List of messages can be referenced using the `message` property of the group within the expression.
+
+NOTE: In releases prior to _version 5.0_, the root object was the collection of `Message`.
For example:
[source,xml]
----
-release-strategy-expression="#this.size() gt 5"
+release-strategy-expression="!messages.?[payload==5].empty"
----
-In this example the root object of the SpEL Evaluation Context is the `MessageGroup` itself, and you are simply stating that as soon as there are more than 5 messages in this group, it should be released.
+In this example the root object of the SpEL Evaluation Context is the `MessageGroup` itself, and you are simply stating that as soon as there are a message with payload as `5` in this group, it should be released.
[[agg-and-group-to]]
====== Aggregator and Group Timeout
@@ -666,7 +669,7 @@ For this purpose the `groupTimeout` option allows scheduling the `MessageGroup`
+ release-strategy-expression="messages[0].headers.sequenceNumber == messages[0].headers.sequenceSize"/>
----
With this example, the normal _release_ will be possible if the aggregator receives the last message in sequence as defined by the `release-strategy-expression`.
@@ -764,6 +767,7 @@ All state is carried by the `MessageGroup` and its management is delegated to th
[source,java]
----
public interface MessageGroupStore {
+
int getMessageCountForAllMessageGroups();
int getMarkedMessageCountForAllMessageGroups();
diff --git a/src/reference/asciidoc/resequencer.adoc b/src/reference/asciidoc/resequencer.adoc
index 53d9492c67..2eeab2b9e3 100644
--- a/src/reference/asciidoc/resequencer.adoc
+++ b/src/reference/asciidoc/resequencer.adoc
@@ -128,13 +128,13 @@ _Optional, with restrictions (requires `release-strategy` to be present)._
-<14> A SpEL expression representing the release strategy; the root object for the expression is a `Collection` of `Message` s.
+<14> A SpEL expression representing the release strategy; the root object for the expression is a `MessageGroup`.
Example: `"size() == 5"`.
Only one of `release-strategy` or `release-strategy-expression` is allowed.
-<15> Only applies if a `MessageGroupStoreReaper` is configured for the `` `MessageStore`.
+<15> Only applies if a `MessageGroupStoreReaper` is configured for the `` `MessageStore`.
By default, when a `MessageGroupStoreReaper` is configured to expire partial groups, empty groups are also removed.
Empty groups exist after a group is released normally.
This is to enable the detection and discarding of late-arriving messages.
diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc
index 3998fa912b..3c72f76318 100644
--- a/src/reference/asciidoc/whats-new.adoc
+++ b/src/reference/asciidoc/whats-new.adoc
@@ -52,6 +52,9 @@ A simple `PassThroughTransactionSynchronizationFactory` is provided to always st
That message is used as a `failedMessage` property of the `MessagingException` which wraps a raw exception thrown during transaction completion.
See <> for more information.
+The aggregator expression-based `ReleaseStrategy` now evaluates the expression against the `MesageGroup` instead of just the collection of `Message`.
+See <> for more information.
+
==== JMS Changes
Previously, Spring Integration JMS XML configuration used a default bean name `connectionFactory` for the JMS Connection Factory, allowing the property to be omitted from component definitions.