diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/AbstractJmsChannel.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/AbstractJmsChannel.java index 4efc99655c..67b2741aae 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/AbstractJmsChannel.java +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/AbstractJmsChannel.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. @@ -23,9 +23,15 @@ import org.springframework.messaging.Message; import org.springframework.util.Assert; /** + * A base {@link AbstractMessageChannel} implementation for JMS-backed message channels. + * * @author Mark Fisher * @author Gary Russell + * * @since 2.0 + * + * @see PollableJmsChannel + * @see SubscribableJmsChannel */ public abstract class AbstractJmsChannel extends AbstractMessageChannel { 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 cc3cfc0592..e7a209b9d5 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-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. @@ -406,8 +406,8 @@ public class ChannelPublishingJmsMessageListener headers.put(IntegrationMessageHeaderAccessor.DELIVERY_ATTEMPT, new AtomicInteger()); } requestMessage = - (result instanceof Message) ? - this.messageBuilderFactory.fromMessage((Message) result).copyHeaders(headers).build() : + result instanceof Message message ? + this.messageBuilderFactory.fromMessage(message).copyHeaders(headers).build() : this.messageBuilderFactory.withPayload(result).copyHeaders(headers).build(); } catch (RuntimeException e) { diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/DefaultJmsHeaderMapper.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/DefaultJmsHeaderMapper.java index a31bb36b3d..97b85485e4 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/DefaultJmsHeaderMapper.java +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/DefaultJmsHeaderMapper.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. @@ -159,9 +159,9 @@ public class DefaultJmsHeaderMapper extends JmsHeaderMapper { if (jmsCorrelationId instanceof Number) { jmsCorrelationId = jmsCorrelationId.toString(); } - if (jmsCorrelationId instanceof String) { + if (jmsCorrelationId instanceof String jmsCorrelationIdStr) { try { - jmsMessage.setJMSCorrelationID((String) jmsCorrelationId); + jmsMessage.setJMSCorrelationID(jmsCorrelationIdStr); } catch (Exception ex) { LOGGER.info("Failed to set JMSCorrelationID, skipping", ex); @@ -171,9 +171,9 @@ public class DefaultJmsHeaderMapper extends JmsHeaderMapper { private void populateReplyToPropertyFromHeaders(MessageHeaders headers, jakarta.jms.Message jmsMessage) { Object jmsReplyTo = headers.get(JmsHeaders.REPLY_TO); - if (jmsReplyTo instanceof Destination) { + if (jmsReplyTo instanceof Destination destination) { try { - jmsMessage.setJMSReplyTo((Destination) jmsReplyTo); + jmsMessage.setJMSReplyTo(destination); } catch (Exception ex) { LOGGER.info("Failed to set JMSReplyTo, skipping", ex); @@ -183,9 +183,9 @@ public class DefaultJmsHeaderMapper extends JmsHeaderMapper { private void populateTypePropertyFromHeaders(MessageHeaders headers, jakarta.jms.Message jmsMessage) { Object jmsType = headers.get(JmsHeaders.TYPE); - if (jmsType instanceof String) { + if (jmsType instanceof String jmsTypeStr) { try { - jmsMessage.setJMSType((String) jmsType); + jmsMessage.setJMSType(jmsTypeStr); } catch (Exception ex) { LOGGER.info("Failed to set JMSType, skipping", ex); diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/DynamicJmsTemplate.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/DynamicJmsTemplate.java index 9560230f6f..6bed9bd79a 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/DynamicJmsTemplate.java +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/DynamicJmsTemplate.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2021 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,6 +24,15 @@ import org.springframework.jms.support.destination.JmsDestinationAccessor; import org.springframework.util.Assert; /** + * A {@code JmsTemplate} implementation used by {@link JmsSendingMessageHandler} for + * propagating QoS properties from the request message into the underlying {@code + * producer.send(message, getDeliveryMode(), getPriority(), getTimeToLive())} API. + * Propagation is only applied when {@code explicitQosEnabled} is true. + *

+ * Starting with version 5.0.8, a default value of the receive-timeout is -1 (no wait) + * for the {@link CachingConnectionFactory} and {@code cacheConsumers}, otherwise + * it is 1 second. + * * @author Mark Fisher * @author Artem Bilan * @@ -45,8 +54,8 @@ public class DynamicJmsTemplate extends JmsTemplate { public void setConnectionFactory(ConnectionFactory connectionFactory) { super.setConnectionFactory(connectionFactory); if (!this.receiveTimeoutExplicitlySet) { - if (connectionFactory instanceof CachingConnectionFactory && - ((CachingConnectionFactory) connectionFactory).isCacheConsumers()) { + if (connectionFactory instanceof CachingConnectionFactory cachingConnectionFactory && + cachingConnectionFactory.isCacheConsumers()) { super.setReceiveTimeout(JmsDestinationAccessor.RECEIVE_TIMEOUT_NO_WAIT); } else { diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsDestinationPollingSource.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsDestinationPollingSource.java index e4f4d49588..90559bc3e3 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsDestinationPollingSource.java +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsDestinationPollingSource.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. @@ -137,8 +137,8 @@ public class JmsDestinationPollingSource extends AbstractMessageSource { } } AbstractIntegrationMessageBuilder builder = - (object instanceof Message) - ? getMessageBuilderFactory().fromMessage((Message) object) + (object instanceof Message message) + ? getMessageBuilderFactory().fromMessage(message) : getMessageBuilderFactory().withPayload(object); return builder.copyHeadersIfAbsent(mappedHeaders); } @@ -148,7 +148,7 @@ public class JmsDestinationPollingSource extends AbstractMessageSource { } private jakarta.jms.Message doReceiveJmsMessage() { - jakarta.jms.Message jmsMessage = null; + jakarta.jms.Message jmsMessage; if (this.destination != null) { jmsMessage = this.jmsTemplate.receiveSelected(this.destination, this.messageSelector); } diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsInboundGateway.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsInboundGateway.java index a5d7e9fb7d..d277fbcf39 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsInboundGateway.java +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsInboundGateway.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2023 the original author or authors. + * Copyright 2016-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. @@ -109,7 +109,7 @@ public class JmsInboundGateway extends MessagingGatewaySupport implements Orderl /** * 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 will be closed and the listener invokers shut * down; this behavior is new starting with version 5.1. Default: true. * @param shutdownContainerOnStop false to not shutdown. * @since 5.1 diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsMessageDrivenEndpoint.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsMessageDrivenEndpoint.java index 845a5ffae3..5f9a58a4df 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsMessageDrivenEndpoint.java +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsMessageDrivenEndpoint.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. @@ -140,7 +140,7 @@ public class JmsMessageDrivenEndpoint extends MessageProducerSupport implements /** * 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 will be closed and the listener invokers shut * down; this behavior is new starting with version 5.1. Default: true. * @param shutdownContainerOnStop false to not shutdown. * @since 5.1 diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsOutboundGateway.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsOutboundGateway.java index 7bcf11849f..18141a309f 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsOutboundGateway.java +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsOutboundGateway.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. @@ -467,11 +467,11 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler } if (this.requestDestinationExpressionProcessor != null) { Object result = this.requestDestinationExpressionProcessor.processMessage(message); - if (result instanceof Destination) { - return (Destination) result; + if (result instanceof Destination destination) { + return destination; } - if (result instanceof String) { - return resolveRequestDestination((String) result, session); + if (result instanceof String destinationName) { + return resolveRequestDestination(destinationName, session); } throw new MessageDeliveryException(message, "Evaluation of requestDestinationExpression failed " + @@ -496,11 +496,11 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler } if (this.replyDestinationExpressionProcessor != null) { Object result = this.replyDestinationExpressionProcessor.processMessage(message); - if (result instanceof Destination) { - return (Destination) result; + if (result instanceof Destination destination) { + return destination; } - if (result instanceof String) { - return resolveReplyDestination((String) result, session); + if (result instanceof String destinationName) { + return resolveReplyDestination(destinationName, session); } throw new MessageDeliveryException(message, "Evaluation of replyDestinationExpression failed to produce a Destination or destination name. " + @@ -771,8 +771,8 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler } } - if (reply instanceof jakarta.jms.Message) { - return buildReply((jakarta.jms.Message) reply); + if (reply instanceof jakarta.jms.Message jmsMessage) { + return buildReply(jmsMessage); } else { return reply; @@ -799,8 +799,8 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler // do not propagate back the gateway's internal correlation id jmsReplyHeaders.remove(this.correlationKey); } - if (result instanceof Message) { - return getMessageBuilderFactory().fromMessage((Message) result).copyHeaders(jmsReplyHeaders); + if (result instanceof Message message) { + return getMessageBuilderFactory().fromMessage(message).copyHeaders(jmsReplyHeaders); } else { return getMessageBuilderFactory().withPayload(result).copyHeaders(jmsReplyHeaders); @@ -850,8 +850,8 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler * Remove the gateway's internal correlation Id to avoid conflicts with an upstream * gateway. */ - if (reply instanceof jakarta.jms.Message) { - ((jakarta.jms.Message) reply).setJMSCorrelationID(null); + if (reply instanceof jakarta.jms.Message jmsMessage) { + jmsMessage.setJMSCorrelationID(null); } return reply; } @@ -1239,11 +1239,11 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler */ private void deleteDestinationIfTemporary(Destination destination) { try { - if (destination instanceof TemporaryQueue) { - ((TemporaryQueue) destination).delete(); + if (destination instanceof TemporaryQueue temporaryQueue) { + temporaryQueue.delete(); } - else if (destination instanceof TemporaryTopic) { - ((TemporaryTopic) destination).delete(); + else if (destination instanceof TemporaryTopic temporaryTopic) { + temporaryTopic.delete(); } } catch (@SuppressWarnings("unused") JMSException e) { diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsSendingMessageHandler.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsSendingMessageHandler.java index e192db5936..342d3cdb7b 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsSendingMessageHandler.java +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsSendingMessageHandler.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. @@ -212,11 +212,11 @@ public class JmsSendingMessageHandler extends AbstractMessageHandler { } private void send(Object destination, Object objectToSend, MessagePostProcessor messagePostProcessor) { - if (destination instanceof Destination) { - this.jmsTemplate.convertAndSend((Destination) destination, objectToSend, messagePostProcessor); + if (destination instanceof Destination destinationObj) { + this.jmsTemplate.convertAndSend(destinationObj, objectToSend, messagePostProcessor); } - else if (destination instanceof String) { - this.jmsTemplate.convertAndSend((String) destination, objectToSend, messagePostProcessor); + else if (destination instanceof String destinationStr) { + this.jmsTemplate.convertAndSend(destinationStr, objectToSend, messagePostProcessor); } else { // fallback to default destination of the template this.jmsTemplate.convertAndSend(objectToSend, messagePostProcessor); diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/PollableJmsChannel.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/PollableJmsChannel.java index 1085c960f7..63ffb81971 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/PollableJmsChannel.java +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/PollableJmsChannel.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. @@ -31,6 +31,8 @@ import org.springframework.messaging.support.ChannelInterceptor; import org.springframework.messaging.support.ExecutorChannelInterceptor; /** + * A JMS-backed channel from which messages can be received through polling. + * * @author Mark Fisher * @author Oleg Zhurakousky * @author Gary Russell @@ -75,9 +77,7 @@ public class PollableJmsChannel extends AbstractJmsChannel Deque interceptorStack = null; boolean counted = false; try { - if (isLoggingEnabled()) { - logger.trace(() -> "preReceive on channel '" + this + "'"); - } + logger.trace(() -> "preReceive on channel '" + this + "'"); if (!interceptorList.getInterceptors().isEmpty()) { interceptorStack = new ArrayDeque<>(); @@ -118,26 +118,15 @@ public class PollableJmsChannel extends AbstractJmsChannel } if (object == null) { - if (isLoggingEnabled()) { - logger.trace(() -> "postReceive on channel '" + this + "', message is null"); - } + logger.trace(() -> "postReceive on channel '" + this + "', message is null"); return null; } - else { - Message message; - if (object instanceof Message) { - message = (Message) object; - } - else { - message = getMessageBuilderFactory() - .withPayload(object) - .build(); - } - if (isLoggingEnabled()) { - logger.debug(() -> "postReceive on channel '" + this + "', message: " + message); - } - return message; - } + Message message = object instanceof Message msg + ? msg + : getMessageBuilderFactory().withPayload(object).build(); + logger.debug(() -> "postReceive on channel '" + this + "', message: " + message); + + return message; } private void incrementReceiveCounter() { diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/SubscribableJmsChannel.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/SubscribableJmsChannel.java index 126e5fd9cc..f45a3e2434 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/SubscribableJmsChannel.java +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/SubscribableJmsChannel.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. @@ -218,8 +218,8 @@ public class SubscribableJmsChannel extends AbstractJmsChannel } if (converted != null) { messageToSend = - converted instanceof Message - ? (Message) converted + converted instanceof Message convertedMessage + ? convertedMessage : this.messageBuilderFactory.withPayload(converted).build(); this.dispatcher.dispatch(messageToSend); } diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsChannelFactoryBean.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsChannelFactoryBean.java index f2248c69d4..5eee8c4cdf 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsChannelFactoryBean.java +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsChannelFactoryBean.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. @@ -48,6 +48,8 @@ import org.springframework.util.ErrorHandler; import org.springframework.util.StringUtils; /** + * An {@link AbstractFactoryBean} implementation for creating {@link AbstractJmsChannel} instances. + * * @author Mark Fisher * @author Oleg Zhurakousky * @author Gary Russell @@ -434,8 +436,7 @@ public class JmsChannelFactoryBean extends AbstractFactoryBean) inputMessage); + if (inputMessage instanceof Message message) { + return new ErrorMessage(throwable, headers, message); } else { return new ErrorMessage(throwable, headers); diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/SubscribableJmsChannelTests.java b/spring-integration-jms/src/test/java/org/springframework/integration/jms/SubscribableJmsChannelTests.java index 8be22df496..2733e3dda3 100644 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/SubscribableJmsChannelTests.java +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/SubscribableJmsChannelTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2022 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. @@ -338,9 +338,7 @@ public class SubscribableJmsChannelTests extends ActiveMQMultiContextTests { private static boolean waitUntilRegisteredWithDestination(SubscribableJmsChannel channel, long timeout) { AbstractMessageListenerContainer container = (AbstractMessageListenerContainer) new DirectFieldAccessor(channel).getPropertyValue("container"); - if (container instanceof DefaultMessageListenerContainer) { - DefaultMessageListenerContainer listenerContainer = - (DefaultMessageListenerContainer) container; + if (container instanceof DefaultMessageListenerContainer listenerContainer) { if (listenerContainer.getCacheLevel() != DefaultMessageListenerContainer.CACHE_CONSUMER) { return true; }