diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/ChannelPublishingJmsMessageListener.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/ChannelPublishingJmsMessageListener.java index b6bba27257..1d1daa9834 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/ChannelPublishingJmsMessageListener.java +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/ChannelPublishingJmsMessageListener.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2023 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. @@ -17,6 +17,7 @@ package org.springframework.integration.jms; import java.util.Map; +import java.util.concurrent.atomic.AtomicInteger; import io.micrometer.observation.ObservationRegistry; import jakarta.jms.DeliveryMode; @@ -30,13 +31,18 @@ import org.springframework.beans.BeansException; import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.BeanFactoryAware; import org.springframework.beans.factory.InitializingBean; +import org.springframework.core.AttributeAccessor; import org.springframework.core.log.LogAccessor; import org.springframework.expression.Expression; import org.springframework.expression.spel.support.StandardEvaluationContext; +import org.springframework.integration.IntegrationMessageHeaderAccessor; +import org.springframework.integration.StaticMessageHeaderAccessor; import org.springframework.integration.core.MessagingTemplate; import org.springframework.integration.expression.ExpressionUtils; import org.springframework.integration.gateway.MessagingGatewaySupport; +import org.springframework.integration.jms.support.JmsMessageHeaderErrorMessageStrategy; import org.springframework.integration.support.DefaultMessageBuilderFactory; +import org.springframework.integration.support.ErrorMessageUtils; import org.springframework.integration.support.MessageBuilderFactory; import org.springframework.integration.support.management.TrackableComponent; import org.springframework.integration.support.management.metrics.MetricsCaptor; @@ -54,6 +60,10 @@ import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessagingException; import org.springframework.messaging.support.ErrorMessage; +import org.springframework.retry.RecoveryCallback; +import org.springframework.retry.RetryOperations; +import org.springframework.retry.support.RetrySynchronizationManager; +import org.springframework.retry.support.RetryTemplate; import org.springframework.util.Assert; /** @@ -347,6 +357,30 @@ public class ChannelPublishingJmsMessageListener this.gatewayDelegate.setReceiverObservationConvention(observationConvention); } + /** + * Set a {@link RetryTemplate} to use for retrying a message delivery within the + * adapter. Unlike adding retry at the container level, this can be used with an + * {@code ErrorMessageSendingRecoverer} {@link RecoveryCallback} to publish to the + * error channel after retries are exhausted. You generally should not configure an + * error channel when using retry here, use a {@link RecoveryCallback} instead. + * @param retryTemplate the template. + * @since 6.3 + * @see #setRecoveryCallback(RecoveryCallback) + */ + public void setRetryTemplate(RetryTemplate retryTemplate) { + this.gatewayDelegate.retryTemplate = retryTemplate; + } + + /** + * Set a {@link RecoveryCallback} when using retry within the adapter. + * @param recoveryCallback the callback. + * @since 6.3 + * @see #setRetryTemplate(RetryTemplate) + */ + public void setRecoveryCallback(RecoveryCallback> recoveryCallback) { + this.gatewayDelegate.recoveryCallback = recoveryCallback; + } + @Override public void setBeanFactory(BeanFactory beanFactory) throws BeansException { this.beanFactory = beanFactory; @@ -367,6 +401,9 @@ public class ChannelPublishingJmsMessageListener } Map headers = this.headerMapper.toHeaders(jmsMessage); + if (this.gatewayDelegate.retryTemplate != null) { + headers.put(IntegrationMessageHeaderAccessor.DELIVERY_ATTEMPT, new AtomicInteger()); + } requestMessage = (result instanceof Message) ? this.messageBuilderFactory.fromMessage((Message) result).copyHeaders(headers).build() : @@ -385,10 +422,10 @@ public class ChannelPublishingJmsMessageListener } if (!this.expectReply) { - this.gatewayDelegate.send(requestMessage); + this.gatewayDelegate.send(jmsMessage, requestMessage); } else { - Message replyMessage = this.gatewayDelegate.sendAndReceiveMessage(requestMessage); + Message replyMessage = this.gatewayDelegate.sendAndReceiveMessage(jmsMessage, requestMessage); if (replyMessage != null) { Destination destination = getReplyDestination(jmsMessage, session); this.logger.debug(() -> "Reply destination: " + destination); @@ -424,6 +461,12 @@ public class ChannelPublishingJmsMessageListener this.gatewayDelegate.setBeanFactory(this.beanFactory); } this.gatewayDelegate.afterPropertiesSet(); + if (this.gatewayDelegate.retryTemplate != null) { + Assert.state(this.gatewayDelegate.getErrorChannel() == null, + "Cannot have an 'errorChannel' property when a 'RetryTemplate' is " + + "provided; use an 'ErrorMessageSendingRecoverer' in the 'recoveryCallback' property to " + + "send an error message when retries are exhausted"); + } this.messageBuilderFactory = IntegrationUtils.getMessageBuilderFactory(this.beanFactory); this.evaluationContext = ExpressionUtils.createStandardEvaluationContext(this.beanFactory); } @@ -551,21 +594,65 @@ public class ChannelPublishingJmsMessageListener private class GatewayDelegate extends MessagingGatewaySupport { + private static final ThreadLocal ATTRIBUTES_HOLDER = new ThreadLocal<>(); + + @Nullable + private RetryOperations retryTemplate; + + @Nullable + private RecoveryCallback> recoveryCallback; + GatewayDelegate() { + setErrorMessageStrategy(new JmsMessageHeaderErrorMessageStrategy()); } - @Override - protected void send(Object request) { // NOSONAR - not useless, increases visibility - super.send(request); + private void send(jakarta.jms.Message jmsMessage, Message requestMessage) { + try { + if (this.retryTemplate == null) { + setAttributesIfNecessary(jmsMessage, requestMessage); + send(requestMessage); + } + else { + this.retryTemplate.execute( + context -> { + StaticMessageHeaderAccessor.getDeliveryAttempt(requestMessage).incrementAndGet(); + setAttributesIfNecessary(jmsMessage, requestMessage); + send(requestMessage); + return null; + }, this.recoveryCallback); + } + } + finally { + if (this.retryTemplate == null) { + ATTRIBUTES_HOLDER.remove(); + } + } } - @Override - protected Message sendAndReceiveMessage(Object request) { // NOSONAR - not useless, increases visibility - return super.sendAndReceiveMessage(request); + private Message sendAndReceiveMessage(jakarta.jms.Message jmsMessage, Message requestMessage) { + try { + if (this.retryTemplate == null) { + setAttributesIfNecessary(jmsMessage, requestMessage); + return sendAndReceiveMessage(requestMessage); + } + else { + return this.retryTemplate.execute( + context -> { + StaticMessageHeaderAccessor.getDeliveryAttempt(requestMessage).incrementAndGet(); + setAttributesIfNecessary(jmsMessage, requestMessage); + return sendAndReceiveMessage(requestMessage); + }, this.recoveryCallback); + } + } + finally { + if (this.retryTemplate == null) { + ATTRIBUTES_HOLDER.remove(); + } + } } protected ErrorMessage buildErrorMessage(Throwable throwable) { - return super.buildErrorMessage(null, throwable); + return buildErrorMessage(null, throwable); } protected MessagingTemplate getMessagingTemplate() { @@ -582,6 +669,29 @@ public class ChannelPublishingJmsMessageListener } } + @Override + protected AttributeAccessor getErrorMessageAttributes(@Nullable Message message) { + AttributeAccessor attributes = ATTRIBUTES_HOLDER.get(); + return (attributes != null) ? attributes : super.getErrorMessageAttributes(message); + } + + private void setAttributesIfNecessary(Object jmsMessage, Message message) { + boolean needHolder = getErrorChannel() != null && this.retryTemplate == null; + boolean needAttributes = needHolder || this.retryTemplate != null; + if (needHolder) { + ATTRIBUTES_HOLDER.set(ErrorMessageUtils.getAttributeAccessor(null, null)); + } + if (needAttributes) { + AttributeAccessor attributes = + this.retryTemplate != null + ? RetrySynchronizationManager.getContext() + : ATTRIBUTES_HOLDER.get(); + if (attributes != null) { + attributes.setAttribute(ErrorMessageUtils.INPUT_MESSAGE_CONTEXT_KEY, message); + attributes.setAttribute(JmsMessageHeaderErrorMessageStrategy.JMS_RAW_MESSAGE, jmsMessage); + } + } + } } } diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/dsl/JmsInboundGatewaySpec.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/dsl/JmsInboundGatewaySpec.java index 22e1d4548e..ca1cab8849 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/dsl/JmsInboundGatewaySpec.java +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/dsl/JmsInboundGatewaySpec.java @@ -32,6 +32,8 @@ import org.springframework.integration.util.CheckedFunction; import org.springframework.jms.listener.AbstractMessageListenerContainer; import org.springframework.jms.support.converter.MessageConverter; import org.springframework.jms.support.destination.DestinationResolver; +import org.springframework.retry.RecoveryCallback; +import org.springframework.retry.support.RetryTemplate; import org.springframework.util.Assert; /** @@ -221,9 +223,35 @@ public class JmsInboundGatewaySpec> } /** - * Set to false to prevent listener container shutdown when the endpoint is stopped. + * Set a {@link RetryTemplate} to use for retrying a message delivery within the + * adapter. Unlike adding retry at the container level, this can be used with an + * {@code ErrorMessageSendingRecoverer} {@link RecoveryCallback} to publish to the + * error channel after retries are exhausted. You generally should not configure an + * error channel when using retry here, use a {@link RecoveryCallback} instead. + * @param retryTemplate the template. + * @since 6.3 + * @see #recoveryCallback(RecoveryCallback) + */ + public S retryTemplate(RetryTemplate retryTemplate) { + this.target.getListener().setRetryTemplate(retryTemplate); + return _this(); + } + + /** + * Set a {@link RecoveryCallback} when using retry within the adapter. + * @param recoveryCallback the callback. + * @since 6.3 + * @see #retryTemplate(RetryTemplate) + */ + public S recoveryCallback(RecoveryCallback> recoveryCallback) { + this.target.getListener().setRecoveryCallback(recoveryCallback); + return _this(); + } + + /** + * Set to {@code false} to prevent listener container shutdown when the endpoint is stopped. * Then, if so configured, any cached consumer(s) in the container will remain. - * Otherwise the shared connection and will be closed and the listener invokers shut + * Otherwise, the shared connection and will be closed and the listener invokers shut * down; this behavior is new starting with version 5.1. Default: true. * @param shutdown false to not shutdown. * @return the spec. diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/dsl/JmsMessageDrivenChannelAdapterSpec.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/dsl/JmsMessageDrivenChannelAdapterSpec.java index ee2884ed8e..8921d480bc 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/dsl/JmsMessageDrivenChannelAdapterSpec.java +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/dsl/JmsMessageDrivenChannelAdapterSpec.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2023 the original author or authors. + * Copyright 2016-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. @@ -29,6 +29,9 @@ import org.springframework.integration.jms.JmsHeaderMapper; import org.springframework.integration.jms.JmsMessageDrivenEndpoint; import org.springframework.jms.listener.AbstractMessageListenerContainer; import org.springframework.jms.support.converter.MessageConverter; +import org.springframework.messaging.Message; +import org.springframework.retry.RecoveryCallback; +import org.springframework.retry.support.RetryTemplate; import org.springframework.util.Assert; /** @@ -93,6 +96,32 @@ public class JmsMessageDrivenChannelAdapterSpec> recoveryCallback) { + this.target.getListener().setRecoveryCallback(recoveryCallback); + return _this(); + } + /** * * @param the target {@link JmsListenerContainerSpec} implementation type. diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/support/JmsMessageHeaderErrorMessageStrategy.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/support/JmsMessageHeaderErrorMessageStrategy.java new file mode 100644 index 0000000000..df8408fe65 --- /dev/null +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/support/JmsMessageHeaderErrorMessageStrategy.java @@ -0,0 +1,63 @@ +/* + * Copyright 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. + * 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.jms.support; + +import java.util.HashMap; +import java.util.Map; + +import org.springframework.core.AttributeAccessor; +import org.springframework.integration.IntegrationMessageHeaderAccessor; +import org.springframework.integration.support.ErrorMessageStrategy; +import org.springframework.integration.support.ErrorMessageUtils; +import org.springframework.lang.Nullable; +import org.springframework.messaging.Message; +import org.springframework.messaging.support.ErrorMessage; + +/** + * An {@link ErrorMessageStrategy} extension that adds the raw JMS message as + * a header to the {@link ErrorMessage}. + * + * @author Artem Bilan + * + * @since 6.3 + * + */ +public class JmsMessageHeaderErrorMessageStrategy implements ErrorMessageStrategy { + + /** + * Header name/retry context variable for the raw received message. + */ + public static final String JMS_RAW_MESSAGE = "jms_raw_message"; + + @Override + public ErrorMessage buildErrorMessage(Throwable throwable, @Nullable AttributeAccessor context) { + Object inputMessage = context == null ? null + : context.getAttribute(ErrorMessageUtils.INPUT_MESSAGE_CONTEXT_KEY); + Map headers = new HashMap<>(); + if (context != null) { + headers.put(JMS_RAW_MESSAGE, context.getAttribute(JMS_RAW_MESSAGE)); + headers.put(IntegrationMessageHeaderAccessor.SOURCE_DATA, context.getAttribute(JMS_RAW_MESSAGE)); + } + if (inputMessage instanceof Message) { + return new ErrorMessage(throwable, headers, (Message) inputMessage); + } + else { + return new ErrorMessage(throwable, headers); + } + } + +} diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/support/package-info.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/support/package-info.java new file mode 100644 index 0000000000..8572fc09b1 --- /dev/null +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/support/package-info.java @@ -0,0 +1,6 @@ +/** + * Provides JMS Components support classes. + */ +@org.springframework.lang.NonNullApi +@org.springframework.lang.NonNullFields +package org.springframework.integration.jms.support; diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/dsl/JmsTests.java b/spring-integration-jms/src/test/java/org/springframework/integration/jms/dsl/JmsTests.java index 19066acea2..8d7a42a653 100644 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/dsl/JmsTests.java +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/dsl/JmsTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2023 the original author or authors. + * Copyright 2016-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. @@ -35,7 +35,9 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; +import org.springframework.integration.IntegrationMessageHeaderAccessor; import org.springframework.integration.MessageTimeoutException; +import org.springframework.integration.StaticMessageHeaderAccessor; import org.springframework.integration.annotation.InboundChannelAdapter; import org.springframework.integration.annotation.IntegrationComponentScan; import org.springframework.integration.annotation.MessagingGateway; @@ -74,6 +76,7 @@ import org.springframework.messaging.PollableChannel; import org.springframework.messaging.simp.SimpMessageHeaderAccessor; import org.springframework.messaging.support.ChannelInterceptor; import org.springframework.messaging.support.InterceptableChannel; +import org.springframework.retry.support.RetryTemplate; import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; import org.springframework.transaction.PlatformTransactionManager; @@ -224,6 +227,8 @@ public class JmsTests extends ActiveMQMultiContextTests { .extracting(Message::getPayload) .isEqualTo("foo"); + assertThat(StaticMessageHeaderAccessor.getDeliveryAttempt(receive).get()).isEqualTo(3); + assertThat(this.jmsOutboundFlowTemplate).isNotNull(); TestObservationRegistryAssert.assertThat(this.observationRegistry) @@ -456,7 +461,14 @@ public class JmsTests extends ActiveMQMultiContextTests { Jms.container(amqFactory, "containerSpecDestination") .pubSubDomain(false) .taskExecutor(Executors.newCachedThreadPool())) - .id("observedJmsMessageDrivenChannelAdapter")) + .id("observedJmsMessageDrivenChannelAdapter") + .retryTemplate(new RetryTemplate())) + .handle((p, h) -> { + if (h.get(IntegrationMessageHeaderAccessor.DELIVERY_ATTEMPT, AtomicInteger.class).get() < 3) { + throw new RuntimeException("intentional for retry"); + } + return p; + }) .transform(String::trim) .channel(jmsOutboundInboundReplyChannel()) .get(); diff --git a/src/reference/antora/modules/ROOT/pages/jms.adoc b/src/reference/antora/modules/ROOT/pages/jms.adoc index 5f3c399035..ac03a447d8 100644 --- a/src/reference/antora/modules/ROOT/pages/jms.adoc +++ b/src/reference/antora/modules/ROOT/pages/jms.adoc @@ -150,7 +150,7 @@ If you want the entire flow to be transactional (for example, if there is a down Alternatively, consider using a `jms-message-driven-channel-adapter` with `acknowledge` set to `transacted` (the default). [[jms-message-driven-channel-adapter]] -== Message-driven Channel Adapter +== Message Driven Channel Adapter The `message-driven-channel-adapter` requires a reference to either an instance of a Spring `MessageListener` container (any subclass of `AbstractMessageListenerContainer`) or both `ConnectionFactory` and `Destination` (a 'destinationName' can be provided in place of the 'destination' reference). The following example defines a message-driven channel adapter with a `Destination` reference: @@ -269,6 +269,9 @@ Starting with version 5.1, when the endpoint is stopped while the application re Previously, the connection and consumers remained open. To revert to the previous behavior, set the `shutdownContainerOnStop` on the `JmsMessageDrivenEndpoint` to `false`. +Starting with version 6.3, the `ChannelPublishingJmsMessageListener` can now be supplied with a `RetryTemplate` and `RecoveryCallback>` for retries on the downstream send and send-and-receive operations. +These options are also exposed into a `JmsMessageDrivenChannelAdapterSpec` for Java DSL. + [[jms-md-conversion-errors]] === Inbound Conversion Errors @@ -469,6 +472,8 @@ public IntegrationFlow jmsInboundGatewayFlow(ConnectionFactory connectionFactory } ---- +Starting with version 6.3, the `Jms.inboundGateway()` API exposes a `retryTemplate()` and `recoveryCallback()` options for retrying internal send-and-receive operations. + [[jms-outbound-gateway]] == Outbound Gateway diff --git a/src/reference/antora/modules/ROOT/pages/whats-new.adoc b/src/reference/antora/modules/ROOT/pages/whats-new.adoc index 3fb6bc06de..5ea6cffbe2 100644 --- a/src/reference/antora/modules/ROOT/pages/whats-new.adoc +++ b/src/reference/antora/modules/ROOT/pages/whats-new.adoc @@ -42,4 +42,10 @@ See xref:mqtt.adoc[MQTT Support] for more information. === 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 +See xref:testing.adoc[Testing Support] for more information. + +[[x6.3-jms]] +=== JMS Support Changes + +The `ChannelPublishingJmsMessageListener` can now be supplied with a `RetryTemplate` and `RecoveryCallback>` for retries on the downstream send operations. +See xref:jms.adoc#jms-message-driven-channel-adapter[Message Driven Channel Adapter] for more information. \ No newline at end of file