GH-9706: Allow collection of payloads as an aggregator result
Fixes: https://github.com/spring-projects/spring-integration/issues/9706 Previously, if a `MessageGroupProcessor` 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. * Add `AbstractCorrelatingMessageHandler.isResultCollectionOfMessages()` to return `true` only if result is a collection of messages, treating them as a "partial sequence". * Deprecate `AbstractCorrelatingMessageHandler.verifyResultCollectionConsistsOfMessages()` since it is out of use now.
This commit is contained in:
@@ -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<Message<?>>) 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);
|
||||
|
||||
@@ -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());
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -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<Message>`, 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`
|
||||
|
||||
|
||||
@@ -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 `<control-bus use-registry="">` 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.
|
||||
Reference in New Issue
Block a user