INT-4192: Improve ExprEvaluatingReleaseStrategy

JIRA: https://jira.spring.io/browse/INT-4192

To avoid implicit overhead for loading messages from persistent message store, use entire `MessageGroup` as root object for expression evaluation context

Also see https://github.com/spring-projects/spring-integration/wiki/Spring-Integration-4.3-to-5.0-Migration-Guide

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

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

spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/AggregatorIntegrationTests-context.xml

@@ -43,13 +43,13 @@
 	<aggregator input-channel="groupTimeoutExpressionAggregatorInput" output-channel="output" discard-channel="discard"
 				send-partial-result-on-expiry="true"
 				group-timeout-expression="size() ge 2 ? 100 : -1"
-			    release-strategy-expression="[0].headers.sequenceNumber == [0].headers.sequenceSize"/>
+			    release-strategy-expression="messages[0].headers.sequenceNumber == messages[0].headers.sequenceSize"/>
 
 	<aggregator input-channel="zeroGroupTimeoutExpressionAggregatorInput" output-channel="output" discard-channel="discard"
 				send-partial-result-on-expiry="true"
 				send-timeout="10"
 				group-timeout-expression="@bool.getAndSet(true) ? 0 : 10"
-			    release-strategy-expression="[0].headers.sequenceNumber == [0].headers.sequenceSize"/>
+			    release-strategy-expression="messages[0].headers.sequenceNumber == messages[0].headers.sequenceSize"/>
 
 	<beans:bean id="bool" class="java.util.concurrent.atomic.AtomicBoolean" />
 

spring-integration-core/src/test/java/org/springframework/integration/scattergather/config/ScatterGatherTests-context.xml

@@ -12,7 +12,7 @@
 
 	<!--Auction scenario-->
 	<scatter-gather input-channel="inputAuction" output-channel="output" scatter-channel="auctionChannel">
-		<gatherer release-strategy-expression="^[payload gt 5] != null or size() == 3"/>
+		<gatherer release-strategy-expression="messages.^[payload gt 5] != null or size() == 3"/>
 	</scatter-gather>
 
 	<task:executor id="threadPoolTaskExecutor" pool-size="10"/>
@@ -34,7 +34,7 @@
 			<recipient channel="distribution2Channel"/>
 			<recipient channel="distribution3Channel"/>
 		</scatterer>
-		<gatherer release-strategy-expression="^[payload gt 5] != null or size() == 3"/>
+		<gatherer release-strategy-expression="messages.^[payload gt 5] != null or size() == 3"/>
 	</scatter-gather>
 
 	<channel id="gatherChannel">
@@ -55,7 +55,7 @@
 	<gateway id="gateway" default-request-channel="gatewayAuction"/>
 
 	<scatter-gather input-channel="gatewayAuction" scatter-channel="auctionChannel">
-		<gatherer release-strategy-expression="^[payload gt 5] != null or size() == 3"/>
+		<gatherer release-strategy-expression="messages.^[payload gt 5] != null or size() == 3"/>
 	</scatter-gather>
 
 	<chain input-channel="scatterGatherWithinChain" output-channel="output">

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`
 <aggregator input-channel="input" output-channel="output"
         send-partial-result-on-expiry="true"
         group-timeout-expression="size() ge 2 ? 10000 : -1"
-        release-strategy-expression="[0].headers.sequenceNumber == [0].headers.sequenceSize"/>
+        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();

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 `<resequcencer>` `MessageStore`.
+<15> Only applies if a `MessageGroupStoreReaper` is configured for the `<resequencer>` `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.

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 <<transaction-synchronization>> for more information.
 
+The aggregator expression-based `ReleaseStrategy` now evaluates the expression against the `MesageGroup` instead of just the collection of `Message<?>`.
+See <<aggregator-spel>> 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.