diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java index 78c76a52e3..fa13a3af6e 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2024 the original author or authors. + * Copyright 2002-2025 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. @@ -904,8 +904,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP this.logger.debug(() -> "Completing group with correlationKey [" + correlationKey + "]"); result = this.outputProcessor.processMessageGroup(group); - if (result instanceof Collection) { - verifyResultCollectionConsistsOfMessages((Collection) result); + if (isResultCollectionOfMessages(result)) { partialSequence = (Collection>) result; } @@ -944,12 +943,26 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP } + /** + * Probably the method is {@code protected} by mistake. + * @param elements the group processor result. + * @deprecated without replacement - out of use from now on. + */ + @Deprecated(since = "6.5", forRemoval = true) protected void verifyResultCollectionConsistsOfMessages(Collection elements) { Class commonElementType = CollectionUtils.findCommonElementType(elements); Assert.isAssignable(Message.class, commonElementType, () -> "The expected collection of Messages contains non-Message element: " + commonElementType); } + private static boolean isResultCollectionOfMessages(Object result) { + if (result instanceof Collection resultCollection) { + Class commonElementType = CollectionUtils.findCommonElementType(resultCollection); + return commonElementType != null && Message.class.isAssignableFrom(commonElementType); + } + return false; + } + protected Object obtainGroupTimeout(MessageGroup group) { if (this.groupTimeoutExpression != null) { Object timeout = this.groupTimeoutExpression.getValue(this.evaluationContext, group); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dsl/extensions/IntegrationFlowExtensionTests.java b/spring-integration-core/src/test/java/org/springframework/integration/dsl/extensions/IntegrationFlowExtensionTests.java index 35513864bf..a95c95271c 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/dsl/extensions/IntegrationFlowExtensionTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/dsl/extensions/IntegrationFlowExtensionTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2020-2022 the original author or authors. + * Copyright 2020-2025 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. @@ -18,8 +18,8 @@ package org.springframework.integration.dsl.extensions; import java.util.Arrays; import java.util.function.Consumer; -import java.util.stream.Collectors; +import org.assertj.core.api.InstanceOfAssertFactories; import org.junit.jupiter.api.Test; import org.springframework.beans.factory.annotation.Autowired; @@ -66,7 +66,8 @@ public class IntegrationFlowExtensionTests { assertThat(replyMessage) .isNotNull() .extracting(Message::getPayload) - .isEqualTo("ONE, TWO, THREE"); + .asInstanceOf(InstanceOfAssertFactories.LIST) + .containsOnly("ONE", "TWO", "THREE"); } @Configuration @@ -109,8 +110,7 @@ public class IntegrationFlowExtensionTests { group.getMessages() .stream() .map(Message::getPayload) - .map(String.class::cast) - .collect(Collectors.joining(", "))); + .toList()); } } diff --git a/src/reference/antora/modules/ROOT/pages/aggregator.adoc b/src/reference/antora/modules/ROOT/pages/aggregator.adoc index 8de7e67a71..06f726907d 100644 --- a/src/reference/antora/modules/ROOT/pages/aggregator.adoc +++ b/src/reference/antora/modules/ROOT/pages/aggregator.adoc @@ -144,6 +144,10 @@ Starting with version 6.0, the splitting behaviour, described above, works only Otherwise, with any other `MessageGroupProcessor` implementation that returns a `Collection`, only a single reply message is emitted with the whole collection of messages as its payload. Such logic is dictated by the canonical purpose of an aggregator - collect request messages by some key and produce a single grouped message. +Prior to version 6.5, if a `MessageGroupProcessor` (usually lambda from DSL) returns a collection of payloads, the `AbstractCorrelatingMessageHandler` has failed with the `IllegalArgumentException` stating that only collection of messages is possible. +From now on such a restriction is eliminated and returned collection of payloads is emitted as a single reply message from the aggregator with just headers from the last request message. +If headers aggregation is required alongside with a collection of payloads, an `AbstractAggregatingMessageGroupProcessor` implementations are recommended to be used instead of plain `MessageGroupProcessor` functional interface. + [[releasestrategy]] === `ReleaseStrategy` diff --git a/src/reference/antora/modules/ROOT/pages/whats-new.adoc b/src/reference/antora/modules/ROOT/pages/whats-new.adoc index a10d4d8b3c..2726f6db8f 100644 --- a/src/reference/antora/modules/ROOT/pages/whats-new.adoc +++ b/src/reference/antora/modules/ROOT/pages/whats-new.adoc @@ -21,3 +21,8 @@ The deprecated previously usage of `org.springframework.util.concurrent.Listenab The previously deprecated SpEL-based Control Bus components have been removed in favor of functionality around `ControlBusCommandRegistry`. The `` attribute is deprecated now without replacement since only `ControlBusCommandRegistry` functionality is available. The Java DSL `controlBusOnRegistry()` operator is deprecated in favor of restored `controlBus()` which is fully based now on the `ControlBusCommandRegistry`. +See xref:control-bus.adoc[Control Bus] for more information. + +The `AbstractCorrelatingMessageHandler` does not throw an `IllegalArgumentException` for the collection of payloads as a result of the `MessageGroupProcessor`. +Instead, such a collection is wrapped into a single reply message. +See xref:aggregator.adoc[Aggregator] for more information. \ No newline at end of file