Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-9521: Make middle-flow endpoints started later

Browse Source 
Fixes: https://github.com/spring-projects/spring-integration/issues/9521
This commit is contained in:
Artem Bilan
2024-09-30 11:39:37 -04:00
parent 90f06b1f76
commit cdcffd7312
4 changed files with 34 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java
View File

@@ -416,17 +416,9 @@ public class ConsumerEndpointFactoryBean
if (this.autoStartup != null) {
this.endpoint.setAutoStartup(this.autoStartup);
}
int phaseToSet = this.phase;
if (!this.isPhaseSet) {
if (this.endpoint instanceof PollingConsumer) {
phaseToSet = Integer.MAX_VALUE / 2;
}
else {
phaseToSet = Integer.MIN_VALUE;
}
if (this.isPhaseSet) {
this.endpoint.setPhase(this.phase);
}
this.endpoint.setPhase(phaseToSet);
}

7
spring-integration-core/src/main/java/org/springframework/integration/endpoint/EventDrivenConsumer.java
View File

@@ -45,7 +45,12 @@ public class EventDrivenConsumer extends AbstractEndpoint implements Integration
Assert.notNull(handler, "handler must not be null");
this.inputChannel = inputChannel;
this.handler = handler;
setPhase(Integer.MIN_VALUE);
if (this.handler instanceof MessageProducer) {
setPhase(Integer.MIN_VALUE + 1000);
}
else {
setPhase(Integer.MIN_VALUE);
}
}
@Override

14
spring-integration-core/src/test/java/org/springframework/integration/config/xml/ServiceActivatorParserTests.java
View File

@@ -16,8 +16,7 @@
package org.springframework.integration.config.xml;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
@@ -29,8 +28,8 @@ import org.springframework.integration.handler.ServiceActivatingHandler;
import org.springframework.integration.handler.advice.AbstractRequestHandlerAdvice;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.fail;
@@ -39,11 +38,10 @@ import static org.assertj.core.api.Assertions.fail;
* @author Mark Fisher
* @author Gary Russell
* @author Artem Bilan
*
* @since 2.0
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
@SpringJUnitConfig
@DirtiesContext
public class ServiceActivatorParserTests {
@Autowired
@@ -211,7 +209,7 @@ public class ServiceActivatorParserTests {
@Test
public void testConsumerEndpointFactoryBeanDefaultPhase() {
assertThat(this.testAliasEndpoint.getPhase()).isEqualTo(Integer.MIN_VALUE);
assertThat(this.testAliasEndpoint.getPhase()).isEqualTo(Integer.MIN_VALUE + 1000);
}
private Object sendAndReceive(MessageChannel channel, Object payload) {

21
spring-integration-core/src/test/java/org/springframework/integration/dsl/correlation/CorrelationHandlerTests.java
View File

@@ -38,6 +38,7 @@ import org.springframework.core.task.TaskExecutor;
import org.springframework.integration.IntegrationMessageHeaderAccessor;
import org.springframework.integration.aggregator.FluxAggregatorMessageHandler;
import org.springframework.integration.aggregator.HeaderAttributeCorrelationStrategy;
import org.springframework.integration.aggregator.MessageCountReleaseStrategy;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.config.EnableIntegration;
import org.springframework.integration.dsl.IntegrationFlow;
@@ -47,6 +48,8 @@ import org.springframework.integration.dsl.Transformers;
import org.springframework.integration.dsl.context.IntegrationFlowContext;
import org.springframework.integration.handler.MessageTriggerAction;
import org.springframework.integration.json.ObjectToJsonTransformer;
import org.springframework.integration.store.MessageGroupStore;
import org.springframework.integration.store.SimpleMessageStore;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
@@ -60,7 +63,6 @@ import static org.assertj.core.api.Assertions.assertThat;
/**
* @author Artem Bilan
* @author Gary Russell
*
* @since 5.0
*/
@SpringJUnitConfig
@@ -320,6 +322,23 @@ public class CorrelationHandlerTests {
.get();
}
@Bean
IntegrationFlow purgeOrphanedGroupsDoesNotFailStartupFlow() {
MessageGroupStore messageStore = new SimpleMessageStore();
// Add two messages that 'complete' the aggregation immediately on startup
messageStore.addMessagesToGroup("test", new GenericMessage<>("1"));
messageStore.addMessagesToGroup("test", new GenericMessage<>("2"));
return f -> f.aggregate((a) -> a
.messageStore(messageStore)
.id("purgeOrphanedGroups")
.expireTimeout(1) // Expire immediately
// The group above is actually 'complete' on startup
.releaseStrategy(new MessageCountReleaseStrategy(2)))
.channel("purgeOrphanedGroupsChannel")
.handle(message -> { }, e -> e.id("endOfFlowEndpoint"));
}
}
record TestSplitterPojo(List<String> first, List<String> second) {