INT-4116: Introduce FileAggregator (#3511)

* INT-4116: Introduce FileAggregator

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

* Implement a `FileSplitter.FileMarker`-based aggregation strategies
and utilize them in a general `FileAggregator` component
* Make `HeaderAttributeCorrelationStrategy.attributeName` as `final`; add `Assert.notEmpty()`
* Fix `AggregatorFactoryBean` and `AggregatorSpec` to parse the provided processor
for possible `CorrelationStrategy` and/or `ReleaseStrategy`
* Introduce short-cut methods into Java & Kotlin DSL for an `aggregate()` configuration
* Introduce a `FileHeaders.LINE_COUNT` for header to be populated in the `FileSplitter`.
We need this info in the `FileAggregator` to avoid possible overhead with JSON deserialization
of the `FileSplitter.FileMarker` messages
* Test and document the feature
* Improve `FileSplitter` doc for code block switch (tabs)

* * Rework FileAggregator do not use Java Streams

spring-integration-core/src/main/java/org/springframework/integration/aggregator/HeaderAttributeCorrelationStrategy.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2021 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.
@@ -17,23 +17,25 @@
 package org.springframework.integration.aggregator;
 
 import org.springframework.messaging.Message;
+import org.springframework.util.Assert;
 
 /**
- * Default implementation of {@link CorrelationStrategy}. Uses a header
- * attribute to determine the correlation key value.
+ * Default implementation of {@link CorrelationStrategy}.
+ * Uses a provided header attribute to determine the correlation key value.
  *
  * @author Marius Bogoevici
+ * @author Artem Bilan
  */
 public class HeaderAttributeCorrelationStrategy implements CorrelationStrategy {
 
-	private String attributeName;
+	private final String attributeName;
 
 
 	public HeaderAttributeCorrelationStrategy(String attributeName) {
+		Assert.hasText(attributeName, "the 'attributeName' must not be empty");
 		this.attributeName = attributeName;
 	}
 
-
 	public Object getCorrelationKey(Message<?> message) {
 		return message.getHeaders().get(this.attributeName);
 	}

spring-integration-core/src/main/java/org/springframework/integration/config/AggregatorFactoryBean.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2020 the original author or authors.
+ * Copyright 2015-2021 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.
@@ -22,6 +22,7 @@ import java.util.Map;
 import java.util.function.Function;
 
 import org.aopalliance.aop.Advice;
+import org.jetbrains.annotations.Nullable;
 
 import org.springframework.expression.Expression;
 import org.springframework.integration.aggregator.AbstractAggregatingMessageGroupProcessor;
@@ -211,14 +212,15 @@ public class AggregatorFactoryBean extends AbstractSimpleMessageHandlerFactoryBe
 		}
 
 		AggregatingMessageHandler aggregator = new AggregatingMessageHandler(outputProcessor);
+
 		JavaUtils.INSTANCE
 				.acceptIfNotNull(this.expireGroupsUponCompletion, aggregator::setExpireGroupsUponCompletion)
 				.acceptIfNotNull(this.sendTimeout, aggregator::setSendTimeout)
 				.acceptIfNotNull(this.outputChannelName, aggregator::setOutputChannelName)
 				.acceptIfNotNull(this.lockRegistry, aggregator::setLockRegistry)
 				.acceptIfNotNull(this.messageStore, aggregator::setMessageStore)
-				.acceptIfNotNull(this.correlationStrategy, aggregator::setCorrelationStrategy)
-				.acceptIfNotNull(this.releaseStrategy, aggregator::setReleaseStrategy)
+				.acceptIfNotNull(obtainCorrelationStrategy(), aggregator::setCorrelationStrategy)
+				.acceptIfNotNull(obtainReleaseStrategy(), aggregator::setReleaseStrategy)
 				.acceptIfNotNull(this.groupTimeoutExpression, aggregator::setGroupTimeoutExpression)
 				.acceptIfNotNull(this.forceReleaseAdviceChain, aggregator::setForceReleaseAdviceChain)
 				.acceptIfNotNull(this.taskScheduler, aggregator::setTaskScheduler)
@@ -236,6 +238,28 @@ public class AggregatorFactoryBean extends AbstractSimpleMessageHandlerFactoryBe
 		return aggregator;
 	}
 
+	@Nullable
+	private CorrelationStrategy obtainCorrelationStrategy() {
+		if (this.correlationStrategy == null && this.processorBean != null) {
+			CorrelationStrategyFactoryBean correlationStrategyFactoryBean = new CorrelationStrategyFactoryBean();
+			correlationStrategyFactoryBean.setTarget(this.processorBean);
+			correlationStrategyFactoryBean.afterPropertiesSet();
+			return correlationStrategyFactoryBean.getObject();
+		}
+		return this.correlationStrategy;
+	}
+
+	@Nullable
+	private ReleaseStrategy obtainReleaseStrategy() {
+		if (this.releaseStrategy == null && this.processorBean != null) {
+			ReleaseStrategyFactoryBean releaseStrategyFactoryBean = new ReleaseStrategyFactoryBean();
+			releaseStrategyFactoryBean.setTarget(this.processorBean);
+			releaseStrategyFactoryBean.afterPropertiesSet();
+			return releaseStrategyFactoryBean.getObject();
+		}
+		return this.releaseStrategy;
+	}
+
 	@Override
 	protected Class<? extends MessageHandler> getPreCreationHandlerType() {
 		return AggregatingMessageHandler.class;

spring-integration-core/src/main/java/org/springframework/integration/dsl/AggregatorSpec.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2020 the original author or authors.
+ * Copyright 2016-2021 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.
@@ -69,7 +69,10 @@ public class AggregatorSpec extends CorrelationHandlerSpec<AggregatorSpec, Aggre
 		return super.processor(target)
 				.outputProcessor(methodName != null
 						? new MethodInvokingMessageGroupProcessor(target, methodName)
-						: new MethodInvokingMessageGroupProcessor(target));
+						:
+						(target instanceof MessageGroupProcessor
+								? (MessageGroupProcessor) target
+								: new MethodInvokingMessageGroupProcessor(target)));
 	}
 
 	/**

spring-integration-core/src/main/java/org/springframework/integration/dsl/BaseIntegrationFlowDefinition.java

@@ -1743,6 +1743,17 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
 		return aggregate(null);
 	}
 
+	/**
+	 * A short-cut for the {@code aggregate((aggregator) -> aggregator.processor(aggregatorProcessor))}
+	 * @param aggregatorProcessor the POJO representing aggregation strategies.
+	 * @return the current {@link BaseIntegrationFlowDefinition}.
+	 * @since 5.5
+	 * @see AggregatorSpec
+	 */
+	public B aggregate(Object aggregatorProcessor) {
+		return aggregate((aggregator) -> aggregator.processor(aggregatorProcessor));
+	}
+
 	/**
 	 * Populate the {@link AggregatingMessageHandler} with provided options from {@link AggregatorSpec}.
 	 * In addition accept options for the integration endpoint using {@link GenericEndpointSpec}.

spring-integration-core/src/main/kotlin/org/springframework/integration/dsl/KotlinIntegrationFlowDefinition.kt

@@ -1,5 +1,5 @@
 /*
- * Copyright 2020 the original author or authors.
+ * Copyright 2020-2021 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.
@@ -718,6 +718,14 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ
 		this.delegate.resequence(resequencer)
 	}
 
+	/**
+	 * A short-cut for the `aggregate { processor(aggregatorProcessor) }`
+	 * @since 5.5
+	 */
+	fun aggregate(aggregator: Any) {
+		this.delegate.aggregate(aggregator)
+	}
+
 	/**
 	 * Populate the [AggregatingMessageHandler] with provided options from [AggregatorSpec].
 	 * In addition accept options for the integration endpoint using [GenericEndpointSpec].