diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java index a46824b2de..d8d484f666 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2023 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. @@ -34,10 +34,14 @@ import io.micrometer.observation.Observation; import io.micrometer.observation.ObservationRegistry; import org.springframework.beans.factory.BeanFactory; +import org.springframework.context.ApplicationContext; +import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.context.Lifecycle; import org.springframework.core.OrderComparator; import org.springframework.core.log.LogAccessor; import org.springframework.integration.IntegrationPattern; import org.springframework.integration.IntegrationPatternType; +import org.springframework.integration.MessageDispatchingException; import org.springframework.integration.context.IntegrationContextUtils; import org.springframework.integration.context.IntegrationObjectSupport; import org.springframework.integration.history.MessageHistory; @@ -110,6 +114,10 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport private volatile String fullChannelName; + private volatile boolean applicationRunning; + + private volatile Lifecycle applicationRunningController; + @Override public String getComponentType() { return "channel"; @@ -319,6 +327,7 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport public boolean send(Message messageArg, long timeout) { Assert.notNull(messageArg, "message must not be null"); Assert.notNull(messageArg.getPayload(), "message payload must not be null"); + assertApplicationRunning(messageArg); Message message = messageArg; if (this.shouldTrack) { message = MessageHistory.write(message, this, getMessageBuilderFactory()); @@ -335,6 +344,39 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport } } + private void assertApplicationRunning(Message message) { + if (!this.applicationRunning) { + ApplicationContext applicationContext = getApplicationContext(); + this.applicationRunning = + applicationContext == null || + !applicationContext.containsBean( + IntegrationContextUtils.APPLICATION_RUNNING_CONTROLLER_BEAN_NAME); + + if (!this.applicationRunning) { + if (((ConfigurableApplicationContext) applicationContext).isActive()) { + this.applicationRunningController = + applicationContext.getBean(IntegrationContextUtils.APPLICATION_RUNNING_CONTROLLER_BEAN_NAME, + Lifecycle.class); + this.applicationRunning = this.applicationRunningController.isRunning(); + } + } + } + + if (this.applicationRunning && this.applicationRunningController != null) { + this.applicationRunning = this.applicationRunningController.isRunning(); + } + + if (!this.applicationRunning) { + throw new MessageDispatchingException(message, + """ + The application context is not ready to dispatch messages. \ + It has to be refreshed or started first. \ + Also, messages must not be emitted from initialization phase, \ + like 'afterPropertiesSet()', '@PostConstruct' or bean definition methods. \ + Consider to use 'SmartLifecycle.start()' instead."""); + } + } + private boolean sendWithObservation(Message message, long timeout) { MutableMessage messageToSend = MutableMessage.of(message); Observation observation = IntegrationObservation.PRODUCER.observation( @@ -343,15 +385,15 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport () -> new MessageSenderContext(messageToSend, getComponentName()), this.observationRegistry); Boolean observe = observation.observe(() -> { - Message messageToSendInternal = messageToSend; - if (message instanceof ErrorMessage errorMessage) { - messageToSendInternal = - new ErrorMessage(errorMessage.getPayload(), - messageToSend.getHeaders(), - errorMessage.getOriginalMessage()); - } - return sendInternal(messageToSendInternal, timeout); - }); + Message messageToSendInternal = messageToSend; + if (message instanceof ErrorMessage errorMessage) { + messageToSendInternal = + new ErrorMessage(errorMessage.getPayload(), + messageToSend.getHeaders(), + errorMessage.getOriginalMessage()); + } + return sendInternal(messageToSendInternal, timeout); + }); return Boolean.TRUE.equals(observe); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/ApplicationRunningController.java b/spring-integration-core/src/main/java/org/springframework/integration/config/ApplicationRunningController.java new file mode 100644 index 0000000000..d72ac5d9a4 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/ApplicationRunningController.java @@ -0,0 +1,62 @@ +/* + * Copyright 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. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.config; + +import org.springframework.context.SmartLifecycle; + +/** + * An infrastructure bean to hold the status of the application context when + * it is ready for interaction: refreshed or started. + *

+ * Well-known {@link org.springframework.context.ConfigurableApplicationContext#isRunning()} + * (or {@link org.springframework.context.event.ContextRefreshedEvent}) + * is good for target applications, when all the beans are already started, + * but most of Spring Integration channel adapters initiate their logic + * from the {@link SmartLifecycle#start()} implementation, so it would be false report + * that application is not running during start. + *

+ * This implementation uses {@value Integer#MIN_VALUE} for its phase to be started as early as possible. + * + * @author Artem Bilan + * + * @since 6.5 + */ +class ApplicationRunningController implements SmartLifecycle { + + private volatile boolean running; + + @Override + public void start() { + this.running = true; + } + + @Override + public void stop() { + this.running = false; + } + + @Override + public boolean isRunning() { + return this.running; + } + + @Override + public int getPhase() { + return Integer.MIN_VALUE; + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/DefaultConfiguringBeanFactoryPostProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/config/DefaultConfiguringBeanFactoryPostProcessor.java index 136034b038..663e51509f 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/DefaultConfiguringBeanFactoryPostProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/DefaultConfiguringBeanFactoryPostProcessor.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. @@ -129,6 +129,7 @@ public class DefaultConfiguringBeanFactoryPostProcessor implements BeanDefinitio registerListMessageHandlerMethodFactory(); registerIntegrationConfigurationReport(); registerControlBusCommandRegistry(); + registerApplicationRunningController(); } @Override @@ -453,6 +454,17 @@ public class DefaultConfiguringBeanFactoryPostProcessor implements BeanDefinitio } } + private void registerApplicationRunningController() { + if (!this.beanFactory.containsBean(IntegrationContextUtils.APPLICATION_RUNNING_CONTROLLER_BEAN_NAME)) { + BeanDefinitionBuilder builder = + BeanDefinitionBuilder.genericBeanDefinition(ApplicationRunningController.class) + .setRole(BeanDefinition.ROLE_INFRASTRUCTURE); + + this.registry.registerBeanDefinition(IntegrationContextUtils.APPLICATION_RUNNING_CONTROLLER_BEAN_NAME, + builder.getBeanDefinition()); + } + } + private static BeanDefinitionBuilder createMessageHandlerMethodFactoryBeanDefinition(boolean listCapable) { return BeanDefinitionBuilder.genericBeanDefinition(IntegrationMessageHandlerMethodFactory.class) .addConstructorArgValue(listCapable) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationContextUtils.java b/spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationContextUtils.java index f7d2f8a857..dd605557cf 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationContextUtils.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationContextUtils.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. @@ -102,6 +102,8 @@ public abstract class IntegrationContextUtils { public static final String CONTROL_BUS_COMMAND_REGISTRY_BEAN_NAME = "controlBusCommandRegistry"; + public static final String APPLICATION_RUNNING_CONTROLLER_BEAN_NAME = "applicationRunningController"; + /** * The default timeout for blocking operations like send and receive messages. * @since 6.1 diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java index 42c8bcf6cc..4d7be00354 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java @@ -334,7 +334,7 @@ public abstract class MessageProducerSupport extends AbstractEndpoint return ErrorMessageUtils.getAttributeAccessor(message, null); } - private MessageChannel getRequiredOutputChannel() { + protected MessageChannel getRequiredOutputChannel() { MessageChannel messageChannel = getOutputChannel(); Assert.state(messageChannel != null, "The 'outputChannel' or `outputChannelName` must be configured"); return messageChannel; diff --git a/spring-integration-core/src/test/java/org/springframework/integration/bus/ApplicationContextMessageBusTests.java b/spring-integration-core/src/test/java/org/springframework/integration/bus/ApplicationContextMessageBusTests.java index 8e8de36635..a5a5ccb9bc 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/bus/ApplicationContextMessageBusTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/bus/ApplicationContextMessageBusTests.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. @@ -24,8 +24,10 @@ import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import org.springframework.beans.BeanUtils; import org.springframework.beans.factory.BeanFactory; import org.springframework.context.support.ClassPathXmlApplicationContext; +import org.springframework.integration.MessageDispatchingException; import org.springframework.integration.channel.PublishSubscribeChannel; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.context.IntegrationContextUtils; @@ -43,8 +45,10 @@ import org.springframework.messaging.PollableChannel; import org.springframework.messaging.support.ErrorMessage; import org.springframework.messaging.support.GenericMessage; import org.springframework.scheduling.support.PeriodicTrigger; +import org.springframework.util.ClassUtils; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatExceptionOfType; import static org.mockito.Mockito.mock; /** @@ -67,14 +71,21 @@ public class ApplicationContextMessageBusTests { } @Test - public void endpointRegistrationWithInputChannelReference() { + public void endpointRegistrationWithInputChannelReference() throws ClassNotFoundException { + this.context.registerBean(IntegrationContextUtils.APPLICATION_RUNNING_CONTROLLER_BEAN_NAME, + BeanUtils.instantiateClass( + ClassUtils.forName( + IntegrationContextUtils.BASE_PACKAGE + ".config.ApplicationRunningController", null))); QueueChannel sourceChannel = new QueueChannel(); + sourceChannel.setApplicationContext(this.context); QueueChannel targetChannel = new QueueChannel(); this.context.registerChannel("sourceChannel", sourceChannel); this.context.registerChannel("targetChannel", targetChannel); Message message = MessageBuilder.withPayload("test") .setReplyChannelName("targetChannel").build(); - sourceChannel.send(message); + assertThatExceptionOfType(MessageDispatchingException.class) + .isThrownBy(() -> sourceChannel.send(message)) + .withMessageStartingWith("The application context is not ready to dispatch messages."); AbstractReplyProducingMessageHandler handler = new AbstractReplyProducingMessageHandler() { @Override @@ -88,8 +99,15 @@ public class ApplicationContextMessageBusTests { endpoint.setBeanFactory(mock(BeanFactory.class)); this.context.registerEndpoint("testEndpoint", endpoint); this.context.refresh(); + sourceChannel.send(message); Message result = targetChannel.receive(10000); assertThat(result.getPayload()).isEqualTo("test"); + + this.context.stop(); + + assertThatExceptionOfType(MessageDispatchingException.class) + .isThrownBy(() -> sourceChannel.send(message)) + .withMessageStartingWith("The application context is not ready to dispatch messages."); } @Test diff --git a/spring-integration-event/src/main/java/org/springframework/integration/event/inbound/ApplicationEventListeningMessageProducer.java b/spring-integration-event/src/main/java/org/springframework/integration/event/inbound/ApplicationEventListeningMessageProducer.java index eb9f6efd0e..8c30b34250 100644 --- a/spring-integration-event/src/main/java/org/springframework/integration/event/inbound/ApplicationEventListeningMessageProducer.java +++ b/spring-integration-event/src/main/java/org/springframework/integration/event/inbound/ApplicationEventListeningMessageProducer.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. @@ -26,8 +26,8 @@ import org.springframework.context.event.ContextClosedEvent; import org.springframework.context.event.ContextStoppedEvent; import org.springframework.context.event.GenericApplicationListener; import org.springframework.context.support.AbstractApplicationContext; -import org.springframework.core.Ordered; import org.springframework.core.ResolvableType; +import org.springframework.integration.channel.AbstractMessageChannel; import org.springframework.integration.endpoint.ExpressionMessageProducerSupport; import org.springframework.messaging.Message; import org.springframework.util.Assert; @@ -49,10 +49,10 @@ import org.springframework.util.Assert; public class ApplicationEventListeningMessageProducer extends ExpressionMessageProducerSupport implements GenericApplicationListener { - private volatile Set eventTypes; - private ApplicationEventMulticaster applicationEventMulticaster; + private volatile Set eventTypes; + private volatile long stoppedAt; /** @@ -106,12 +106,17 @@ public class ApplicationEventListeningMessageProducer extends ExpressionMessageP @Override public void onApplicationEvent(ApplicationEvent event) { - if (isActive() || ((event instanceof ContextStoppedEvent || event instanceof ContextClosedEvent) - && stoppedRecently())) { + boolean contextFinished = event instanceof ContextStoppedEvent || event instanceof ContextClosedEvent; + if (isActive() || (contextFinished && stoppedRecently())) { - Object source = event.getSource(); - if (source instanceof Message) { - sendMessage((Message) source); + if (contextFinished && getRequiredOutputChannel() instanceof AbstractMessageChannel) { + logger.warn("Messages for 'ContextStoppedEvent' or 'ContextClosedEvent' cannot be dispatched " + + "via 'AbstractMessageChannel' beans: the application context is in the finished state." + + "Consider to use custom 'MessageChannel' implementation without dispatching logic."); + } + + if (event.getSource() instanceof Message message) { + sendMessage(message); } else { Message message; @@ -128,8 +133,8 @@ public class ApplicationEventListeningMessageProducer extends ExpressionMessageP } private Object extractObjectToSend(Object root) { - if (root instanceof PayloadApplicationEvent) { - return ((PayloadApplicationEvent) root).getPayload(); + if (root instanceof PayloadApplicationEvent payloadApplicationEvent) { + return payloadApplicationEvent.getPayload(); } return evaluatePayloadExpression(root); } @@ -167,14 +172,9 @@ public class ApplicationEventListeningMessageProducer extends ExpressionMessageP return false; } - @Override - public boolean supportsSourceType(Class sourceType) { - return true; - } - @Override public int getOrder() { - return Ordered.LOWEST_PRECEDENCE; + return HIGHEST_PRECEDENCE; } @Override diff --git a/spring-integration-event/src/test/java/org/springframework/integration/event/inbound/ApplicationEventListeningMessageProducerTests.java b/spring-integration-event/src/test/java/org/springframework/integration/event/inbound/ApplicationEventListeningMessageProducerTests.java index 93d09a75cb..c3a320b152 100644 --- a/spring-integration-event/src/test/java/org/springframework/integration/event/inbound/ApplicationEventListeningMessageProducerTests.java +++ b/spring-integration-event/src/test/java/org/springframework/integration/event/inbound/ApplicationEventListeningMessageProducerTests.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. @@ -18,6 +18,9 @@ package org.springframework.integration.event.inbound; import java.util.Map; import java.util.Set; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import org.junit.jupiter.api.Test; @@ -341,13 +344,7 @@ public class ApplicationEventListeningMessageProducerTests { } - private static class TestApplicationListener implements ApplicationListener { - - private final AtomicInteger counter; - - private TestApplicationListener(AtomicInteger counter) { - this.counter = counter; - } + private record TestApplicationListener(AtomicInteger counter) implements ApplicationListener { @Override public void onApplicationEvent(ApplicationEvent event) { @@ -356,4 +353,35 @@ public class ApplicationEventListeningMessageProducerTests { } + static final class ContextEventsChannel implements PollableChannel { + + private final BlockingQueue> internalQueue = new LinkedBlockingQueue<>(); + + @Override + public Message receive() { + return receive(-1); + } + + @Override + public Message receive(long timeout) { + try { + return this.internalQueue.poll(timeout, TimeUnit.MILLISECONDS); + } + catch (InterruptedException e) { + throw new RuntimeException(e); + } + } + + @Override + public boolean send(Message message, long timeout) { + try { + return this.internalQueue.offer(message, timeout, TimeUnit.MILLISECONDS); + } + catch (InterruptedException e) { + throw new RuntimeException(e); + } + } + + } + } diff --git a/spring-integration-event/src/test/java/org/springframework/integration/event/inbound/applicationEventInboundChannelAdapterTests.xml b/spring-integration-event/src/test/java/org/springframework/integration/event/inbound/applicationEventInboundChannelAdapterTests.xml index 7da1ae9b1b..125496a300 100644 --- a/spring-integration-event/src/test/java/org/springframework/integration/event/inbound/applicationEventInboundChannelAdapterTests.xml +++ b/spring-integration-event/src/test/java/org/springframework/integration/event/inbound/applicationEventInboundChannelAdapterTests.xml @@ -1,13 +1,10 @@ + xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" + xsi:schemaLocation="http://www.springframework.org/schema/beans https://www.springframework.org/schema/beans/spring-beans.xsd"> - - - + diff --git a/src/reference/antora/modules/ROOT/pages/whats-new.adoc b/src/reference/antora/modules/ROOT/pages/whats-new.adoc index 2e22e85917..73e5e6838f 100644 --- a/src/reference/antora/modules/ROOT/pages/whats-new.adoc +++ b/src/reference/antora/modules/ROOT/pages/whats-new.adoc @@ -27,6 +27,10 @@ The `AbstractCorrelatingMessageHandler` does not throw an `IllegalArgumentExcept Instead, such a collection is wrapped into a single reply message. See xref:aggregator.adoc[Aggregator] for more information. +The `AbstractMessageChannel` beans now throw a special `MessageDispatchingException` when an attempt to send a message to not running application is done. +In general, it is a design error to try to produce a message from `afterPropertiesSet()`, `@PostConstruct` or bean definition methods. +The `SmartLifecycle.start()` is preferred way for this kind of logic, or better to do that via inbound channel adapters. + [[x6.5-correlation-changes]] == The `discardIndividuallyOnExpiry` Option For Correlation Handlers