diff --git a/spring-integration-test-support/src/main/java/org/springframework/integration/test/util/OnlyOnceTrigger.java b/spring-integration-test-support/src/main/java/org/springframework/integration/test/util/OnlyOnceTrigger.java index 753588cd63..a9a6e3f980 100644 --- a/spring-integration-test-support/src/main/java/org/springframework/integration/test/util/OnlyOnceTrigger.java +++ b/spring-integration-test-support/src/main/java/org/springframework/integration/test/util/OnlyOnceTrigger.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2022 the original author or authors. + * Copyright 2002-2024 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. @@ -25,6 +25,12 @@ import org.springframework.scheduling.Trigger; import org.springframework.scheduling.TriggerContext; /** + * The {@link Trigger} implementation which returns the current time + * to schedule the task immediately and does that only once. + *

+ * The {@link #reset()} method can be called to reuse this trigger from scratch. + *

+ * The {@link #await()} method can be used for synchronization barriers. * * @author Gunnar Hillert * @author Gary Russell diff --git a/spring-integration-test/src/main/java/org/springframework/integration/test/context/MockIntegrationContext.java b/spring-integration-test/src/main/java/org/springframework/integration/test/context/MockIntegrationContext.java index 6273340426..5d3a63183d 100644 --- a/spring-integration-test/src/main/java/org/springframework/integration/test/context/MockIntegrationContext.java +++ b/spring-integration-test/src/main/java/org/springframework/integration/test/context/MockIntegrationContext.java @@ -1,5 +1,5 @@ /* - * Copyright 2017-2023 the original author or authors. + * Copyright 2017-2024 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. @@ -19,13 +19,9 @@ package org.springframework.integration.test.context; import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; -import java.util.HashMap; import java.util.List; -import java.util.Map; import reactor.core.publisher.Mono; -import reactor.util.function.Tuple2; -import reactor.util.function.Tuples; import org.springframework.beans.BeansException; import org.springframework.beans.DirectFieldAccessor; @@ -39,6 +35,7 @@ import org.springframework.context.SmartLifecycle; import org.springframework.integration.core.MessageProducer; import org.springframework.integration.core.MessageSource; import org.springframework.integration.endpoint.AbstractEndpoint; +import org.springframework.integration.endpoint.AbstractPollingEndpoint; import org.springframework.integration.endpoint.IntegrationConsumer; import org.springframework.integration.endpoint.ReactiveStreamsConsumer; import org.springframework.integration.endpoint.SourcePollingChannelAdapter; @@ -47,7 +44,10 @@ import org.springframework.integration.test.util.TestUtils; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; import org.springframework.messaging.ReactiveMessageHandler; +import org.springframework.scheduling.Trigger; import org.springframework.util.Assert; +import org.springframework.util.LinkedMultiValueMap; +import org.springframework.util.MultiValueMap; import org.springframework.util.ObjectUtils; /** @@ -75,7 +75,7 @@ public class MockIntegrationContext implements BeanPostProcessor, SmartInitializ */ public static final String MOCK_INTEGRATION_CONTEXT_BEAN_NAME = "mockIntegrationContext"; - private final Map beans = new HashMap<>(); + private final MultiValueMap beans = new LinkedMultiValueMap<>(); private final List autoStartupCandidates = new ArrayList<>(); @@ -131,8 +131,12 @@ public class MockIntegrationContext implements BeanPostProcessor, SmartInitializ this.beans.entrySet() .stream() - .filter(e -> names == null || names.contains(e.getKey())) - .forEach(e -> resetBean(this.beanFactory.getBean(e.getKey()), e.getValue())); + .filter((bean) -> names == null || names.contains(bean.getKey())) + .forEach((bean) -> { + Object endpoint = this.beanFactory.getBean(bean.getKey()); + bean.getValue() + .forEach((value) -> resetBean(endpoint, value)); + }); if (!ObjectUtils.isEmpty(beanNames)) { for (String name : beanNames) { @@ -144,28 +148,24 @@ public class MockIntegrationContext implements BeanPostProcessor, SmartInitializ } } - private void resetBean(Object endpoint, Object handler) { + private void resetBean(Object endpoint, Object component) { DirectFieldAccessor directFieldAccessor = new DirectFieldAccessor(endpoint); SmartLifecycle lifecycle = null; - if (endpoint instanceof SmartLifecycle && ((SmartLifecycle) endpoint).isRunning()) { - lifecycle = (SmartLifecycle) endpoint; + if (endpoint instanceof SmartLifecycle lifecycleEndpoint && lifecycleEndpoint.isRunning()) { + lifecycle = lifecycleEndpoint; lifecycle.stop(); } - if (endpoint instanceof SourcePollingChannelAdapter) { - directFieldAccessor.setPropertyValue("source", handler); + if (endpoint instanceof SourcePollingChannelAdapter && component instanceof MessageSource) { + directFieldAccessor.setPropertyValue("source", component); } - else if (endpoint instanceof ReactiveStreamsConsumer) { - if (handler instanceof Tuple2 value) { - directFieldAccessor.setPropertyValue(HANDLER, value.getT1()); - directFieldAccessor.setPropertyValue(REACTIVE_MESSAGE_HANDLER, value.getT2()); - } - else { - directFieldAccessor.setPropertyValue(HANDLER, handler); - directFieldAccessor.setPropertyValue(REACTIVE_MESSAGE_HANDLER, null); - } + else if (endpoint instanceof IntegrationConsumer && component instanceof MessageHandler) { + directFieldAccessor.setPropertyValue(HANDLER, component); } - else if (endpoint instanceof IntegrationConsumer) { - directFieldAccessor.setPropertyValue(HANDLER, handler); + else if (endpoint instanceof ReactiveStreamsConsumer && component instanceof ReactiveMessageHandler) { + directFieldAccessor.setPropertyValue(REACTIVE_MESSAGE_HANDLER, component); + } + else if (component instanceof Trigger) { + directFieldAccessor.setPropertyValue("trigger", component); } if (lifecycle != null && lifecycle.isAutoStartup()) { lifecycle.start(); @@ -196,7 +196,8 @@ public class MockIntegrationContext implements BeanPostProcessor, SmartInitializ */ public void substituteMessageSourceFor(String pollingAdapterId, MessageSource mockMessageSource, boolean autoStartup) { - substituteMessageSourceFor(pollingAdapterId, mockMessageSource, SourcePollingChannelAdapter.class, "source", + + substituteComponentFor(pollingAdapterId, mockMessageSource, SourcePollingChannelAdapter.class, "source", autoStartup); } @@ -208,29 +209,24 @@ public class MockIntegrationContext implements BeanPostProcessor, SmartInitializ MessageHandler mockMessageHandler, boolean autoStartup) { Object endpoint = this.beanFactory.getBean(consumerEndpointId, IntegrationConsumer.class); - if (autoStartup && endpoint instanceof Lifecycle) { - ((Lifecycle) endpoint).stop(); + if (autoStartup && endpoint instanceof Lifecycle lifecycle) { + lifecycle.stop(); } DirectFieldAccessor directFieldAccessor = new DirectFieldAccessor(endpoint); Object targetMessageHandler = directFieldAccessor.getPropertyValue(HANDLER); Assert.notNull(targetMessageHandler, () -> "'handler' must not be null in the: " + endpoint); + this.beans.add(consumerEndpointId, targetMessageHandler); if (endpoint instanceof ReactiveStreamsConsumer) { Object targetReactiveMessageHandler = directFieldAccessor.getPropertyValue(REACTIVE_MESSAGE_HANDLER); if (targetReactiveMessageHandler != null) { - this.beans.put(consumerEndpointId, Tuples.of(targetMessageHandler, targetReactiveMessageHandler)); + this.beans.add(consumerEndpointId, targetReactiveMessageHandler); } - else { - this.beans.put(consumerEndpointId, targetMessageHandler); - } - } - else { - this.beans.put(consumerEndpointId, targetMessageHandler); } - if (mockMessageHandler instanceof MessageProducer) { - if (targetMessageHandler instanceof MessageProducer) { - MessageChannel outputChannel = ((MessageProducer) targetMessageHandler).getOutputChannel(); - ((MessageProducer) mockMessageHandler).setOutputChannel(outputChannel); + if (mockMessageHandler instanceof MessageProducer mockMessageProducer) { + if (targetMessageHandler instanceof MessageProducer messageProducer) { + MessageChannel outputChannel = messageProducer.getOutputChannel(); + mockMessageProducer.setOutputChannel(outputChannel); } else { if (mockMessageHandler instanceof MockMessageHandler) { @@ -254,23 +250,45 @@ public class MockIntegrationContext implements BeanPostProcessor, SmartInitializ directFieldAccessor.setPropertyValue(REACTIVE_MESSAGE_HANDLER, reactiveMessageHandler); } - if (autoStartup && endpoint instanceof Lifecycle) { - ((Lifecycle) endpoint).start(); + if (autoStartup && endpoint instanceof Lifecycle lifecycle) { + lifecycle.start(); } } - private void substituteMessageSourceFor(String endpointId, Object messagingComponent, Class endpointClass, + /** + * Replace the real {@link Trigger} in the {@link AbstractPollingEndpoint} bean with provided instance. + * @param pollingEndpointId the {@link AbstractPollingEndpoint} bean id to replace + * @param trigger the {@link Trigger} to set into {@link AbstractPollingEndpoint} + * @since 6.3 + */ + public void substituteTriggerFor(String pollingEndpointId, Trigger trigger) { + substituteTriggerFor(pollingEndpointId, trigger, true); + } + + /** + * Replace the real {@link Trigger} in the {@link AbstractPollingEndpoint} bean with provided instance. + * The endpoint is not started when {@code autoStartup == false}. + * @param pollingEndpointId the {@link AbstractPollingEndpoint} bean id to replace + * @param trigger the {@link Trigger} to set into {@link AbstractPollingEndpoint} + * @param autoStartup start or not the endpoint after replacing its {@link MessageSource} + * @since 6.3 + */ + public void substituteTriggerFor(String pollingEndpointId, Trigger trigger, boolean autoStartup) { + substituteComponentFor(pollingEndpointId, trigger, AbstractPollingEndpoint.class, "trigger", autoStartup); + } + + private void substituteComponentFor(String endpointId, Object messagingComponent, Class endpointClass, String property, boolean autoStartup) { Object endpoint = this.beanFactory.getBean(endpointId, endpointClass); - if (autoStartup && endpoint instanceof Lifecycle) { - ((Lifecycle) endpoint).stop(); + if (autoStartup && endpoint instanceof Lifecycle lifecycle) { + lifecycle.stop(); } DirectFieldAccessor directFieldAccessor = new DirectFieldAccessor(endpoint); - this.beans.put(endpointId, directFieldAccessor.getPropertyValue(property)); + this.beans.add(endpointId, directFieldAccessor.getPropertyValue(property)); directFieldAccessor.setPropertyValue(property, messagingComponent); - if (autoStartup && endpoint instanceof Lifecycle) { - ((Lifecycle) endpoint).start(); + if (autoStartup && endpoint instanceof Lifecycle lifecycle) { + lifecycle.start(); } } diff --git a/spring-integration-test/src/test/java/org/springframework/integration/test/mock/MockMessageSourceTests.java b/spring-integration-test/src/test/java/org/springframework/integration/test/mock/MockMessageSourceTests.java index 6c3357547f..689acbfc7b 100644 --- a/spring-integration-test/src/test/java/org/springframework/integration/test/mock/MockMessageSourceTests.java +++ b/spring-integration-test/src/test/java/org/springframework/integration/test/mock/MockMessageSourceTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2017-2022 the original author or authors. + * Copyright 2017-2024 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. @@ -39,6 +39,7 @@ import org.springframework.integration.dsl.context.IntegrationFlowContext.Integr import org.springframework.integration.scheduling.PollerMetadata; import org.springframework.integration.test.context.MockIntegrationContext; import org.springframework.integration.test.context.SpringIntegrationTest; +import org.springframework.integration.test.util.OnlyOnceTrigger; import org.springframework.messaging.Message; import org.springframework.messaging.support.GenericMessage; import org.springframework.test.annotation.DirtiesContext; @@ -94,7 +95,16 @@ public class MockMessageSourceTests { assertThat(receive.getPayload()).isEqualTo("BAZ"); } - this.applicationContext.getBean("mySourceEndpoint", Lifecycle.class).stop(); + this.mockIntegrationContext.resetBeans("mySourceEndpoint"); + + this.mockIntegrationContext.substituteTriggerFor("mySourceEndpoint", new OnlyOnceTrigger()); + + receive = this.results.receive(10_000); + assertThat(receive).isNotNull() + .extracting(Message::getPayload) + .isEqualTo("MYDATA"); + + assertThat(this.results.receive(10)).isNull(); } @Test diff --git a/src/reference/antora/modules/ROOT/pages/testing.adoc b/src/reference/antora/modules/ROOT/pages/testing.adoc index b1d19df9ca..4e6aae22da 100644 --- a/src/reference/antora/modules/ROOT/pages/testing.adoc +++ b/src/reference/antora/modules/ROOT/pages/testing.adoc @@ -235,6 +235,12 @@ public void tearDown() { } ---- +Starting with version 6.3, the `MockIntegrationContext.substituteTriggerFor()` API has been introduced. +This can be used to replace the real `Trigger` in the `AbstractPollingEndpoint`. +For example the production configuration may rely on daily (or even weekly) cron schedule. +Any custom `Trigger` can be injected into the target endpoint to mitigate the time span. +For example, the mentioned above <> suggests a behavior to schedule polling task immediately and do that only once. + See the https://docs.spring.io/spring-integration/api/org/springframework/integration/test/context/MockIntegrationContext.html[Javadoc] for more information. [[testing-mocks]] diff --git a/src/reference/antora/modules/ROOT/pages/whats-new.adoc b/src/reference/antora/modules/ROOT/pages/whats-new.adoc index aaf9169868..11ff43c9d2 100644 --- a/src/reference/antora/modules/ROOT/pages/whats-new.adoc +++ b/src/reference/antora/modules/ROOT/pages/whats-new.adoc @@ -29,4 +29,10 @@ See xref:security.adoc[Security in Spring Integration] for more information. === MQTT Support Changes The fine-grained configuration based on `MqttSubscription` API is exposed on the `Mqttv5PahoMessageDrivenChannelAdapter`. -See xref:mqtt.adoc[MQTT Support] for more information. \ No newline at end of file +See xref:mqtt.adoc[MQTT Support] for more information. + +[[x6.3-testing]] +=== Testing Support Changes + +The `MockIntegrationContext.substituteTriggerFor()` API has been introduced. +See xref:testing.adoc[Testing Support] for more information. \ No newline at end of file