Introduce a MessageGroup.condition (#3517)

* Introduce a `groupConditionSupplier` for MGS

* Add a `MessageGroup.condition` option
* Add a `MessageGroupStore.conditionSupplier` option
* Use it from the `SimpleMessageStore.addMessagesToGroup()` API
to populate `condition` (if any) into a `MessageGroup`
* Introduce a `GroupConditionProvider` contract to be implemented
on those `ReleaseStrategy` contracts which could be aware of group condition
* Populated a `GroupConditionProvider.getGroupConditionSupplier()`
into a `MessageGroupStore` from the `AbstractCorrelatingMessageHandler`
for end-user convenience
* Rework a `FileMarkerReleaseStrategy` to implement a `GroupConditionProvider`
to provide a function which produces a condition from `file_lineCount` header
of the `END` marker message
* Make the `FileMarkerReleaseStrategy` logic already based on the condition from a group
* Delegate `GroupConditionProvider` from the `FileAggregator`
* Add test for empty file aggregation

* * Implement `condition` in the `AbstractKeyValueMessageStore` and `MongoDbMessageStore`
* Test `condition` for `mongo-aggregator-config.xml` and `FileAggregatorTests` against GemFire

* * Implement `condition` in the `ConfigurableMongoDbMessageStore`

* * Implement `condition` in the `JdbcMessageStore`
* `FileAggregatorTests` against `JdbcMessageStore`
* Refactor `JdbcMessageStore` for better handling of message group metadata
* Remove unused `MARKED` column in the DDL in favor of newly introduced `CONDITION`

* * Add docs for message group condition

* * Move `conditionSupplier` option from MGS to AbstractCorrelatingMH
* Make it as a `BiFunction` to propagate existing condition alongside with the
message to consult
* Expose `groupConditionSupplier` in Java & XML DSLs

* * Fix language in docs

spring-integration-file/src/main/java/org/springframework/integration/file/aggregator/FileAggregator.java

@@ -16,10 +16,13 @@
 
 package org.springframework.integration.file.aggregator;
 
+import java.util.function.BiFunction;
+
 import org.springframework.beans.BeansException;
 import org.springframework.beans.factory.BeanFactory;
 import org.springframework.beans.factory.BeanFactoryAware;
 import org.springframework.integration.aggregator.CorrelationStrategy;
+import org.springframework.integration.aggregator.GroupConditionProvider;
 import org.springframework.integration.aggregator.HeaderAttributeCorrelationStrategy;
 import org.springframework.integration.aggregator.MessageGroupProcessor;
 import org.springframework.integration.aggregator.ReleaseStrategy;
@@ -35,7 +38,7 @@ import org.springframework.messaging.Message;
  * Delegates to {@link HeaderAttributeCorrelationStrategy} with {@link FileHeaders#FILENAME} attribute,
  * {@link FileMarkerReleaseStrategy} and {@link FileAggregatingMessageGroupProcessor}, respectively.
  * <p>
- * The default {@link FileSplitter} behavior with markers enabled is do not provide a sequence details
+ * The default {@link FileSplitter} behavior with markers enabled is about do not provide a sequence details
  * headers, therefore correlation in this aggregator implementation is done by the {@link FileHeaders#FILENAME}
  * header which is still populated by the {@link FileSplitter} for each line emitted, including
  * {@link FileSplitter.FileMarker} messages.
@@ -47,7 +50,8 @@ import org.springframework.messaging.Message;
  *
  * @since 5.5
  */
-public class FileAggregator implements CorrelationStrategy, ReleaseStrategy, MessageGroupProcessor, BeanFactoryAware {
+public class FileAggregator implements CorrelationStrategy, ReleaseStrategy, GroupConditionProvider,
+		MessageGroupProcessor, BeanFactoryAware {
 
 	private final CorrelationStrategy correlationStrategy = new HeaderAttributeCorrelationStrategy(FileHeaders.FILENAME);
 
@@ -70,6 +74,11 @@ public class FileAggregator implements CorrelationStrategy, ReleaseStrategy, Mes
 		return this.releaseStrategy.canRelease(group);
 	}
 
+	@Override
+	public BiFunction<Message<?>, String, String> getGroupConditionSupplier() {
+		return this.releaseStrategy.getGroupConditionSupplier();
+	}
+
 	@Override
 	public Object processMessageGroup(MessageGroup group) {
 		return this.groupProcessor.processMessageGroup(group);

spring-integration-file/src/main/java/org/springframework/integration/file/aggregator/FileMarkerReleaseStrategy.java

@@ -16,8 +16,10 @@
 
 package org.springframework.integration.file.aggregator;
 
-import java.util.Collection;
+import java.util.function.BiFunction;
+import java.util.function.Function;
 
+import org.springframework.integration.aggregator.GroupConditionProvider;
 import org.springframework.integration.aggregator.ReleaseStrategy;
 import org.springframework.integration.file.FileHeaders;
 import org.springframework.integration.file.splitter.FileSplitter;
@@ -29,35 +31,46 @@ import org.springframework.messaging.MessageHeaders;
  * A {@link ReleaseStrategy} which makes a decision based on the presence of
  * {@link org.springframework.integration.file.splitter.FileSplitter.FileMarker.Mark#END}
  * message in the group and its {@link org.springframework.integration.file.FileHeaders#LINE_COUNT} header.
+ * <p>
+ * The logic of this strategy is based on the {@link FileMarkerReleaseStrategy#GROUP_CONDITION}
+ * function populated to the {@link org.springframework.integration.aggregator.AbstractCorrelatingMessageHandler}.
  *
  * @author Artem Bilan
  *
  * @since 5.5
  */
-public class FileMarkerReleaseStrategy implements ReleaseStrategy {
+public class FileMarkerReleaseStrategy implements ReleaseStrategy, GroupConditionProvider {
+
+	/**
+	 * The {@link Function} for
+	 * {@link org.springframework.integration.aggregator.AbstractCorrelatingMessageHandler#setGroupConditionSupplier(BiFunction)}.
+	 */
+	public static final BiFunction<Message<?>, String, String> GROUP_CONDITION =
+			(message, existingCondition) -> {
+				MessageHeaders headers = message.getHeaders();
+				if (FileSplitter.FileMarker.Mark.END.name().equals(headers.get(FileHeaders.MARKER))) {
+					Long lineCount = headers.get(FileHeaders.LINE_COUNT, Long.class);
+					return lineCount != null ? "" + lineCount : existingCondition;
+				}
+				return existingCondition;
+			};
 
 	@Override
 	public boolean canRelease(MessageGroup group) {
 		int size = group.size();
 		if (size > 1) { // Need more than only a START marker
-			Collection<Message<?>> messages = group.getMessages();
-			for (Message<?> message : messages) {
-				if (checkForEndMarker(size, message.getHeaders())) {
-					return true;
-				}
+			String condition = group.getCondition();
+			if (condition != null) {
+				long lineCount = Long.parseLong(condition);
+				return lineCount == size - 2; // line count doesn't include START/END markers.
 			}
 		}
 		return false;
 	}
 
-	private boolean checkForEndMarker(int groupSize, MessageHeaders headers) {
-		if (FileSplitter.FileMarker.Mark.END.name().equals(headers.get(FileHeaders.MARKER))) {
-			Long lineCount = headers.get(FileHeaders.LINE_COUNT, Long.class);
-			return lineCount != null && lineCount == groupSize - 2;
-		}
-		else {
-			return false;
-		}
+	@Override
+	public BiFunction<Message<?>, String, String> getGroupConditionSupplier() {
+		return GROUP_CONDITION;
 	}
 
 }

spring-integration-file/src/test/java/org/springframework/integration/file/aggregator/FileAggregatorTests.java

@@ -24,6 +24,7 @@ import java.io.IOException;
 import java.nio.charset.StandardCharsets;
 import java.util.List;
 
+import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.io.TempDir;
@@ -35,11 +36,16 @@ import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.ImportResource;
 import org.springframework.core.task.TaskExecutor;
 import org.springframework.integration.IntegrationMessageHeaderAccessor;
+import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.config.EnableIntegration;
 import org.springframework.integration.dsl.IntegrationFlow;
 import org.springframework.integration.file.FileHeaders;
 import org.springframework.integration.file.dsl.Files;
 import org.springframework.integration.file.splitter.FileSplitter;
+import org.springframework.integration.jdbc.store.JdbcMessageStore;
+import org.springframework.jdbc.datasource.embedded.EmbeddedDatabase;
+import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseBuilder;
+import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseType;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.PollableChannel;
@@ -57,6 +63,8 @@ import org.springframework.util.FileCopyUtils;
 @DirtiesContext
 public class FileAggregatorTests {
 
+	static EmbeddedDatabase dataSource;
+
 	@TempDir
 	static File tmpDir;
 
@@ -66,6 +74,10 @@ public class FileAggregatorTests {
 	@Qualifier("fileSplitterAggregatorFlow.input")
 	MessageChannel fileSplitterAggregatorFlow;
 
+	@Autowired
+	@Qualifier("jdbcMessageStoreAggregatorFlow.input")
+	MessageChannel jdbcMessageStoreAggregatorFlow;
+
 	@Autowired
 	PollableChannel resultChannel;
 
@@ -84,6 +96,17 @@ public class FileAggregatorTests {
 						"second line\n" +
 						"last line";
 		FileCopyUtils.copy(content.getBytes(StandardCharsets.UTF_8), new FileOutputStream(file, false));
+
+		dataSource = new EmbeddedDatabaseBuilder()
+				.setType(EmbeddedDatabaseType.H2)
+				.addScript("classpath:/org/springframework/integration/jdbc/schema-drop-h2.sql")
+				.addScript("classpath:/org/springframework/integration/jdbc/schema-h2.sql")
+				.build();
+	}
+
+	@AfterAll
+	public static void destroy() {
+		dataSource.shutdown();
 	}
 
 	@Test
@@ -104,6 +127,25 @@ public class FileAggregatorTests {
 				.contains("SECOND LINE", "LAST LINE", "FIRST LINE");
 	}
 
+	@Test
+	void testEmptyFileAggregator() throws IOException {
+		File file = new File(tmpDir, "empty.txt");
+		file.createNewFile();
+		this.jdbcMessageStoreAggregatorFlow.send(new GenericMessage<>(file));
+
+		Message<?> receive = this.resultChannel.receive(10_000);
+		assertThat(receive).isNotNull();
+		assertThat(receive.getHeaders())
+				.containsEntry(FileHeaders.FILENAME, "empty.txt")
+				.containsEntry(FileHeaders.LINE_COUNT, 0L)
+				.doesNotContainKey(IntegrationMessageHeaderAccessor.CORRELATION_ID);
+
+		assertThat(receive.getPayload())
+				.isInstanceOf(List.class)
+				.asList()
+				.isEmpty();
+	}
+
 	@Test
 	void testFileAggregatorXmlConfig() {
 		this.input.send(new GenericMessage<>(file));
@@ -138,7 +180,22 @@ public class FileAggregatorTests {
 					.<String, String>transform(String::toUpperCase)
 					.channel("aggregatorChannel")
 					.aggregate(new FileAggregator())
-					.channel(c -> c.queue("resultChannel"));
+					.channel(resultChannel());
+		}
+
+		@Bean
+		public IntegrationFlow jdbcMessageStoreAggregatorFlow() {
+			return f -> f
+					.split(Files.splitter().markers())
+					.aggregate(aggregator ->
+							aggregator.processor(new FileAggregator())
+									.messageStore(new JdbcMessageStore(dataSource)))
+					.channel(resultChannel());
+		}
+
+		@Bean
+		PollableChannel resultChannel() {
+			return new QueueChannel();
 		}
 
 	}

spring-integration-file/src/test/java/org/springframework/integration/file/aggregator/FileAggregatorTests.xml

@@ -3,13 +3,23 @@
 	   xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
 	   xmlns:int-file="http://www.springframework.org/schema/integration/file"
 	   xmlns:int="http://www.springframework.org/schema/integration"
+	   xmlns:gfe="http://www.springframework.org/schema/geode"
 	   xsi:schemaLocation="http://www.springframework.org/schema/beans https://www.springframework.org/schema/beans/spring-beans.xsd
 	   http://www.springframework.org/schema/integration https://www.springframework.org/schema/integration/spring-integration.xsd
-	   http://www.springframework.org/schema/integration/file https://www.springframework.org/schema/integration/file/spring-integration-file.xsd">
+	   http://www.springframework.org/schema/integration/file https://www.springframework.org/schema/integration/file/spring-integration-file.xsd
+	   http://www.springframework.org/schema/geode https://www.springframework.org/schema/geode/spring-geode.xsd">
+
+	<gfe:cache />
+
+	<gfe:local-region id="region1"/>
+
+	<bean id="gemfireMessageStore" class="org.springframework.integration.gemfire.store.GemfireMessageStore">
+		<constructor-arg ref="region1"/>
+	</bean>
 
 	<int:chain input-channel="input" output-channel="output">
 		<int-file:splitter markers="true"/>
-		<int:aggregator>
+		<int:aggregator message-store="gemfireMessageStore">
 			<bean class="org.springframework.integration.file.aggregator.FileAggregator"/>
 		</int:aggregator>
 	</int:chain>