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 f991ef70f7..80cf3a9e44 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-2019 the original author or authors.
+ * Copyright 2002-2020 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.
@@ -44,6 +44,7 @@ import javax.jms.Topic;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
+import org.springframework.beans.factory.BeanFactory;
import org.springframework.context.Lifecycle;
import org.springframework.core.convert.ConversionService;
import org.springframework.expression.Expression;
@@ -53,6 +54,7 @@ import org.springframework.integration.handler.AbstractReplyProducingMessageHand
import org.springframework.integration.handler.ExpressionEvaluatingMessageProcessor;
import org.springframework.integration.jms.util.JmsAdapterUtils;
import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
+import org.springframework.integration.util.JavaUtils;
import org.springframework.jms.connection.ConnectionFactoryUtils;
import org.springframework.jms.listener.DefaultMessageListenerContainer;
import org.springframework.jms.support.JmsUtils;
@@ -520,61 +522,67 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp
"Exactly one of 'requestDestination', 'requestDestinationName', " +
"or 'requestDestinationExpression' is required.");
ConversionService conversionService = getConversionService();
+ BeanFactory beanFactory = getBeanFactory();
if (this.requestDestinationExpressionProcessor != null) {
- this.requestDestinationExpressionProcessor.setBeanFactory(getBeanFactory());
+ this.requestDestinationExpressionProcessor.setBeanFactory(beanFactory);
if (conversionService != null) {
this.requestDestinationExpressionProcessor.setConversionService(conversionService);
}
}
if (this.replyDestinationExpressionProcessor != null) {
- this.replyDestinationExpressionProcessor.setBeanFactory(getBeanFactory());
+ this.replyDestinationExpressionProcessor.setBeanFactory(beanFactory);
if (conversionService != null) {
this.replyDestinationExpressionProcessor.setConversionService(conversionService);
}
}
- /*
- * This is needed because there is no way to detect 2 or more gateways using the same reply queue
- * with no correlation key.
- */
- boolean hasAReplyDest = this.replyDestination != null || this.replyDestinationName != null
- || this.replyDestinationExpressionProcessor != null;
- if (this.useReplyContainer && (this.correlationKey == null && hasAReplyDest)) {
- logger.warn("The gateway cannot use a reply listener container with a specified " +
- "destination(Name/Expression) " +
- "without a 'correlation-key'; " +
- "a container will NOT be used; " +
- "to avoid this problem, set the 'correlation-key' attribute; " +
- "some consumers, including the Spring Integration , " +
- "support the use of the value 'JMSCorrelationID' " +
- "for this purpose. Alternatively, do not specify a reply destination " +
- "and a temporary queue will be used for replies.");
- this.useReplyContainer = false;
- }
- if (this.useReplyContainer) {
- Assert.state(!"JMSCorrelationID*".equals(this.correlationKey),
- "Using an existing 'JMSCorrelationID' mapped from the 'requestMessage' ('JMSCorrelationID*') " +
- "can't be used when using a 'reply-container'");
- GatewayReplyListenerContainer container = new GatewayReplyListenerContainer();
- setContainerProperties(container);
- container.afterPropertiesSet();
- this.replyContainer = container;
- if (isAsync() && this.correlationKey == null) {
- logger.warn("'async=true' requires a correlationKey; ignored");
- setAsync(false);
- }
- }
- else {
- if (isAsync()) {
- logger.warn("'async=true' is ignored when a reply container is not being used");
- setAsync(false);
- }
- }
+ initializeReplyContainer();
this.initialized = true;
}
}
+ private void initializeReplyContainer() {
+ /*
+ * This is needed because there is no way to detect 2 or more gateways using the same reply queue
+ * with no correlation key.
+ */
+ boolean hasAReplyDest = this.replyDestination != null || this.replyDestinationName != null
+ || this.replyDestinationExpressionProcessor != null;
+ if (this.useReplyContainer && (this.correlationKey == null && hasAReplyDest)) {
+ logger.warn("The gateway cannot use a reply listener container with a specified " +
+ "destination(Name/Expression) " +
+ "without a 'correlation-key'; " +
+ "a container will NOT be used; " +
+ "to avoid this problem, set the 'correlation-key' attribute; " +
+ "some consumers, including the Spring Integration , " +
+ "support the use of the value 'JMSCorrelationID' " +
+ "for this purpose. Alternatively, do not specify a reply destination " +
+ "and a temporary queue will be used for replies.");
+ this.useReplyContainer = false;
+ }
+ if (this.useReplyContainer) {
+ Assert.state(!"JMSCorrelationID*".equals(this.correlationKey),
+ "Using an existing 'JMSCorrelationID' mapped from the 'requestMessage' ('JMSCorrelationID*') " +
+ "can't be used when using a 'reply-container'");
+ GatewayReplyListenerContainer container = new GatewayReplyListenerContainer();
+ setContainerProperties(container);
+ container.afterPropertiesSet();
+ this.replyContainer = container;
+ if (isAsync() && this.correlationKey == null) {
+ logger.warn("'async=true' requires a correlationKey; ignored");
+ setAsync(false);
+ }
+ }
+ else {
+ if (isAsync()) {
+ logger.warn("'async=true' is ignored when a reply container is not being used");
+ setAsync(false);
+ }
+ }
+ }
+
private void setContainerProperties(GatewayReplyListenerContainer container) {
container.setConnectionFactory(this.connectionFactory);
+
if (this.replyDestination != null) {
container.setDestination(this.replyDestination);
}
@@ -594,61 +602,57 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp
container.setMessageSelector(messageSelector);
}
container.setMessageListener(this);
+ applyReplyContainerProperties(container);
+ }
+
+ private void applyReplyContainerProperties(GatewayReplyListenerContainer container) {
if (this.replyContainerProperties != null) {
- if (this.replyContainerProperties.isSessionTransacted() != null) {
- container.setSessionTransacted(this.replyContainerProperties.isSessionTransacted());
- }
- if (this.replyContainerProperties.getCacheLevel() != null) {
- container.setCacheLevel(this.replyContainerProperties.getCacheLevel());
- }
- if (this.replyContainerProperties.getConcurrentConsumers() != null) {
- container.setConcurrentConsumers(this.replyContainerProperties.getConcurrentConsumers());
- }
- if (this.replyContainerProperties.getIdleConsumerLimit() != null) {
- container.setIdleConsumerLimit(this.replyContainerProperties.getIdleConsumerLimit());
- }
- if (this.replyContainerProperties.getIdleTaskExecutionLimit() != null) {
- container.setIdleTaskExecutionLimit(this.replyContainerProperties.getIdleTaskExecutionLimit());
- }
- if (this.replyContainerProperties.getMaxConcurrentConsumers() != null) {
- container.setMaxConcurrentConsumers(this.replyContainerProperties.getMaxConcurrentConsumers());
- }
- if (this.replyContainerProperties.getMaxMessagesPerTask() != null) {
- container.setMaxMessagesPerTask(this.replyContainerProperties.getMaxMessagesPerTask());
- }
- if (this.replyContainerProperties.getReceiveTimeout() != null) {
- container.setReceiveTimeout(this.replyContainerProperties.getReceiveTimeout());
- }
- if (this.replyContainerProperties.getRecoveryInterval() != null) {
- container.setRecoveryInterval(this.replyContainerProperties.getRecoveryInterval());
- }
- if (StringUtils.hasText(this.replyContainerProperties.getSessionAcknowledgeModeName())) {
- Integer acknowledgeMode = JmsAdapterUtils.parseAcknowledgeMode(
- this.replyContainerProperties.getSessionAcknowledgeModeName());
- if (acknowledgeMode != null) {
- if (JmsAdapterUtils.SESSION_TRANSACTED == acknowledgeMode) {
- container.setSessionTransacted(true);
- }
- else {
- container.setSessionAcknowledgeMode(acknowledgeMode);
- }
- }
- }
- else if (this.replyContainerProperties.getSessionAcknowledgeMode() != null) {
- Integer sessionAcknowledgeMode = this.replyContainerProperties.getSessionAcknowledgeMode();
- if (Session.SESSION_TRANSACTED == sessionAcknowledgeMode) {
- container.setSessionTransacted(true);
- }
- else {
- container.setSessionAcknowledgeMode(sessionAcknowledgeMode);
- }
+ JavaUtils.INSTANCE
+ .acceptIfNotNull(this.replyContainerProperties.isSessionTransacted(),
+ container::setSessionTransacted)
+ .acceptIfNotNull(this.replyContainerProperties.getCacheLevel(),
+ container::setCacheLevel)
+ .acceptIfNotNull(this.replyContainerProperties.getConcurrentConsumers(),
+ container::setConcurrentConsumers)
+ .acceptIfNotNull(this.replyContainerProperties.getIdleConsumerLimit(),
+ container::setIdleConsumerLimit)
+ .acceptIfNotNull(this.replyContainerProperties.getIdleTaskExecutionLimit(),
+ container::setIdleTaskExecutionLimit)
+ .acceptIfNotNull(this.replyContainerProperties.getMaxConcurrentConsumers(),
+ container::setMaxConcurrentConsumers)
+ .acceptIfNotNull(this.replyContainerProperties.getMaxMessagesPerTask(),
+ container::setMaxMessagesPerTask)
+ .acceptIfNotNull(this.replyContainerProperties.getReceiveTimeout(),
+ container::setReceiveTimeout)
+ .acceptIfNotNull(this.replyContainerProperties.getRecoveryInterval(),
+ container::setRecoveryInterval)
+ .acceptIfHasText(this.replyContainerProperties.getSessionAcknowledgeModeName(),
+ acknowledgeModeName -> {
+ Integer acknowledgeMode = JmsAdapterUtils.parseAcknowledgeMode(
+ this.replyContainerProperties.getSessionAcknowledgeModeName());
+ if (acknowledgeMode != null) {
+ if (JmsAdapterUtils.SESSION_TRANSACTED == acknowledgeMode) {
+ container.setSessionTransacted(true);
+ }
+ else {
+ container.setSessionAcknowledgeMode(acknowledgeMode);
+ }
+ }
+ })
+ .acceptIfNotNull(this.replyContainerProperties.getSessionAcknowledgeMode(),
+ acknowledgeMode -> {
+ if (Session.SESSION_TRANSACTED == acknowledgeMode) {
+ container.setSessionTransacted(true);
+ }
+ else {
+ container.setSessionAcknowledgeMode(acknowledgeMode);
+ }
+ })
+ .acceptIfNotNull(this.replyContainerProperties.getTaskExecutor(),
+ container::setTaskExecutor);
- }
- if (this.replyContainerProperties.getTaskExecutor() != null) {
- container.setTaskExecutor(this.replyContainerProperties.getTaskExecutor());
- }
- else {
+ if (this.replyContainerProperties.getTaskExecutor() == null) {
// set the beanName so the default TE threads get a meaningful name
String containerBeanName = this.getComponentName();
containerBeanName = ((!StringUtils.hasText(containerBeanName)
@@ -985,8 +989,8 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp
* If the replyTo is not temporary, and the connection is lost while waiting for a reply, reconnect for
* up to receiveTimeout.
*/
- private javax.jms.Message retryableReceiveReply(Session session, Destination replyTo, String messageSelector)
- throws JMSException {
+ private javax.jms.Message retryableReceiveReply(Session session, Destination replyTo, // NOSONAR
+ String messageSelector) throws JMSException {
Connection consumerConnection = null; //NOSONAR
Session consumerSession = session;
@@ -996,17 +1000,15 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp
long replyTimeout = isTemporaryReplyTo
? Long.MIN_VALUE
: this.receiveTimeout < 0
- ? Long.MAX_VALUE
- : System.currentTimeMillis() + this.receiveTimeout;
+ ? Long.MAX_VALUE
+ : System.currentTimeMillis() + this.receiveTimeout;
try {
do {
try {
messageConsumer = consumerSession.createConsumer(replyTo, messageSelector);
javax.jms.Message reply = receiveReplyMessage(messageConsumer);
- if (reply == null) {
- if (replyTimeout > System.currentTimeMillis()) {
- throw new JMSException("Consumer closed before timeout");
- }
+ if (reply == null && replyTimeout > System.currentTimeMillis()) {
+ throw new JMSException("Consumer closed before timeout");
}
return reply;
}
@@ -1027,7 +1029,7 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp
logger.debug("Could not reconnect, retrying: " + ee.getMessage());
}
try {
- Thread.sleep(1000);
+ Thread.sleep(1000); // NOSONAR
}
catch (@SuppressWarnings("unused") InterruptedException e1) {
Thread.currentThread().interrupt();
@@ -1377,10 +1379,10 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp
}
else {
int n = 0;
- while (this.replyDestination == null && n++ < 100) {
+ while (this.replyDestination == null && n++ < 100) { // NOSONAR
logger.debug("Waiting for container to create destination");
try {
- Thread.sleep(100);
+ Thread.sleep(100); // NOSONAR
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
@@ -1488,16 +1490,15 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp
synchronized (JmsOutboundGateway.this.lifeCycleMonitor) {
if (System.currentTimeMillis() - JmsOutboundGateway.this.lastSend >
JmsOutboundGateway.this.idleReplyContainerTimeout
- && JmsOutboundGateway.this.replies.size() == 0) {
+ && JmsOutboundGateway.this.replies.size() == 0 &&
+ JmsOutboundGateway.this.replyContainer.isRunning()) {
- if (JmsOutboundGateway.this.replyContainer.isRunning()) {
- if (logger.isDebugEnabled()) {
- logger.debug(getComponentName() + ": Stopping idle reply container.");
- }
- JmsOutboundGateway.this.replyContainer.stop();
- JmsOutboundGateway.this.idleTask.cancel(false);
- JmsOutboundGateway.this.idleTask = null;
+ if (logger.isDebugEnabled()) {
+ logger.debug(getComponentName() + ": Stopping idle reply container.");
}
+ JmsOutboundGateway.this.replyContainer.stop();
+ JmsOutboundGateway.this.idleTask.cancel(false);
+ JmsOutboundGateway.this.idleTask = null;
}
}
}
diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaTemplateSpec.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaTemplateSpec.java
index 2e40d31d1f..0da850faab 100644
--- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaTemplateSpec.java
+++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaTemplateSpec.java
@@ -49,7 +49,7 @@ public class KafkaTemplateSpec
}
@Override
- public KafkaTemplateSpec id(String id) {
+ public KafkaTemplateSpec id(String id) { // NOSONAR - visibility
return super.id(id);
}
diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaInboundGateway.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaInboundGateway.java
index b4b34d3063..d301299cc3 100644
--- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaInboundGateway.java
+++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaInboundGateway.java
@@ -32,7 +32,6 @@ import org.springframework.integration.context.OrderlyShutdownCapable;
import org.springframework.integration.gateway.MessagingGatewaySupport;
import org.springframework.integration.kafka.support.RawRecordHeaderErrorMessageStrategy;
import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
-import org.springframework.integration.support.ErrorMessageStrategy;
import org.springframework.integration.support.ErrorMessageUtils;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.kafka.core.KafkaTemplate;
@@ -43,7 +42,6 @@ import org.springframework.kafka.listener.adapter.RecordMessagingMessageListener
import org.springframework.kafka.listener.adapter.RetryingMessageListenerAdapter;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
-import org.springframework.kafka.support.converter.BatchMessageConverter;
import org.springframework.kafka.support.converter.ConversionException;
import org.springframework.kafka.support.converter.KafkaMessageHeaders;
import org.springframework.kafka.support.converter.RecordMessageConverter;
@@ -110,7 +108,7 @@ public class KafkaInboundGateway extends MessagingGatewaySupport implem
/**
* Set the message converter; must be a {@link RecordMessageConverter} or
- * {@link BatchMessageConverter} depending on mode.
+ * {@link org.springframework.kafka.support.converter.BatchMessageConverter} depending on mode.
* @param messageConverter the converter.
*/
public void setMessageConverter(RecordMessageConverter messageConverter) {
@@ -216,7 +214,7 @@ public class KafkaInboundGateway extends MessagingGatewaySupport implem
* If there's a retry template, it will set the attributes holder via the listener. If
* there's no retry template, but there's an error channel, we create a new attributes
* holder here. If an attributes holder exists (by either method), we set the
- * attributes for use by the {@link ErrorMessageStrategy}.
+ * attributes for use by the {@link org.springframework.integration.support.ErrorMessageStrategy}.
* @param record the record.
* @param message the message.
*/
diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageSource.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageSource.java
index a56d7f4990..c16b458573 100644
--- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageSource.java
+++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageSource.java
@@ -428,23 +428,11 @@ public class KafkaMessageSource extends AbstractMessageSource