Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Fix some Sonar smells

Browse Source
This commit is contained in:
Artem Bilan
2020-01-03 16:38:57 -05:00
committed by Gary Russell
parent 1d80e9ff05
commit 13bd3e5255
34 changed files with 344 additions and 407 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/AbstractAmqpChannel.java
View File

@@ -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.
@@ -208,10 +208,8 @@ public abstract class AbstractAmqpChannel extends AbstractMessageChannel impleme
@Override
protected void onInit() {
super.onInit();
if (!this.initialized && this.rabbitTemplate != null) {
if (this.connectionFactory != null) {
this.connectionFactory.addConnectionListener(this);
}
if (!this.initialized && this.rabbitTemplate != null && this.connectionFactory != null) {
this.connectionFactory.addConnectionListener(this);
}
this.initialized = true;
}
@@ -246,6 +244,6 @@ public abstract class AbstractAmqpChannel extends AbstractMessageChannel impleme
public void onClose(Connection connection) {
}
protected abstract void doDeclares();
protected abstract void doDeclares();
}

94
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpChannelParser.java
View File

@@ -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.
@@ -31,6 +31,7 @@ import org.springframework.util.StringUtils;
* @author Mark Fisher
* @author Gary Russell
* @author Artem Bilan
*
* @since 2.1
*/
public class AmqpChannelParser extends AbstractChannelParser {
@@ -50,52 +51,73 @@ public class AmqpChannelParser extends AbstractChannelParser {
builder.addPropertyValue("pubSub", "publish-subscribe-channel".equals(element.getLocalName()));
populateConsumersPerQueueIfAny(element, parserContext, builder);
String[] valuesToPopulate = {
"max-subscribers",
"acknowledge-mode",
"auto-startup",
"channel-transacted",
"template-channel-transacted",
"concurrent-consumers",
"encoding",
"expose-listener-channel",
"phase",
"prefetch-count",
"queue-name",
"receive-timeout",
"recovery-interval",
"missing-queues-fatal",
"shutdown-timeout",
"tx-size",
"default-delivery-mode",
"extract-payload",
"headers-last"
};
for (String attribute : valuesToPopulate) {
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, attribute);
}
String[] referencesToPopulate = {
"advice-chain",
"amqp-admin",
"error-handler",
"exchange",
"message-converter",
"message-properties-converter",
"task-executor",
"transaction-attribute",
"transaction-manager",
"outbound-header-mapper",
"inbound-header-mapper"
};
for (String attribute : referencesToPopulate) {
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, attribute);
}
return builder;
}
private void populateConsumersPerQueueIfAny(Element element, ParserContext parserContext,
BeanDefinitionBuilder builder) {
String consumersPerQueue = element.getAttribute("consumers-per-queue");
if (StringUtils.hasText(consumersPerQueue)) {
if (StringUtils.hasText(element.getAttribute("concurrent-consumers"))) {
parserContext.getReaderContext().error("'consumers-per-queue' and 'concurrent-consumers' are mutually "
+ "exclusive", element);
parserContext.getReaderContext()
.error("'consumers-per-queue' and 'concurrent-consumers' are mutually exclusive", element);
}
if (StringUtils.hasText(element.getAttribute("tx-size"))) {
parserContext.getReaderContext().error("'tx-size' is not allowed with 'consumers-per-queue'", element);
}
if (StringUtils.hasText(element.getAttribute("receive-timeout"))) {
parserContext.getReaderContext().error("'receive-timeout' is not allowed with 'consumers-per-queue'",
element);
parserContext.getReaderContext()
.error("'receive-timeout' is not allowed with 'consumers-per-queue'", element);
}
builder.addPropertyValue("consumersPerQueue", consumersPerQueue);
}
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "max-subscribers");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "acknowledge-mode");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "advice-chain");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "amqp-admin");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "auto-startup");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "channel-transacted");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "template-channel-transacted");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "concurrent-consumers");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "encoding");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "error-handler");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "exchange");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "expose-listener-channel");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "message-converter");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "message-properties-converter");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "phase");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "prefetch-count");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "queue-name");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "receive-timeout");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "recovery-interval");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "missing-queues-fatal");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "shutdown-timeout");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "task-executor");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "transaction-attribute");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "transaction-manager");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "tx-size");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "default-delivery-mode");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "extract-payload");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "outbound-header-mapper");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "inbound-header-mapper");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "headers-last");
return builder;
}
}

38
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundChannelAdapter.java
View File

@@ -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.
@@ -61,17 +61,17 @@ import com.rabbitmq.client.Channel;
public class AmqpInboundChannelAdapter extends MessageProducerSupport implements
OrderlyShutdownCapable {
private static final ThreadLocal<AttributeAccessor> attributesHolder = new ThreadLocal<AttributeAccessor>();
private static final ThreadLocal<AttributeAccessor> ATTRIBUTES_HOLDER = new ThreadLocal<>();
private final AbstractMessageListenerContainer messageListenerContainer;
private volatile MessageConverter messageConverter = new SimpleMessageConverter();
private MessageConverter messageConverter = new SimpleMessageConverter();
private volatile AmqpHeaderMapper headerMapper = DefaultAmqpHeaderMapper.inboundMapper();
private AmqpHeaderMapper headerMapper = DefaultAmqpHeaderMapper.inboundMapper();
private RetryTemplate retryTemplate;
private RecoveryCallback<? extends Object> recoveryCallback;
private RecoveryCallback<?> recoveryCallback;
private BatchingStrategy batchingStrategy = new SimpleBatchingStrategy(0, 0, 0L);
@@ -119,7 +119,7 @@ public class AmqpInboundChannelAdapter extends MessageProducerSupport implements
* @since 4.3.10
* @see #setRetryTemplate(RetryTemplate)
*/
public void setRecoveryCallback(RecoveryCallback<? extends Object> recoveryCallback) {
public void setRecoveryCallback(RecoveryCallback<?> recoveryCallback) {
this.recoveryCallback = recoveryCallback;
}
@@ -197,12 +197,12 @@ public class AmqpInboundChannelAdapter extends MessageProducerSupport implements
boolean needHolder = getErrorChannel() != null && this.retryTemplate == null;
boolean needAttributes = needHolder || this.retryTemplate != null;
if (needHolder) {
attributesHolder.set(ErrorMessageUtils.getAttributeAccessor(null, null));
ATTRIBUTES_HOLDER.set(ErrorMessageUtils.getAttributeAccessor(null, null));
}
if (needAttributes) {
AttributeAccessor attributes = this.retryTemplate != null
? RetrySynchronizationManager.getContext()
: attributesHolder.get();
: ATTRIBUTES_HOLDER.get();
if (attributes != null) {
attributes.setAttribute(ErrorMessageUtils.INPUT_MESSAGE_CONTEXT_KEY, message);
attributes.setAttribute(AmqpMessageHeaderErrorMessageStrategy.AMQP_RAW_MESSAGE, amqpMessage);
@@ -212,7 +212,7 @@ public class AmqpInboundChannelAdapter extends MessageProducerSupport implements
@Override
protected AttributeAccessor getErrorMessageAttributes(org.springframework.messaging.Message<?> message) {
AttributeAccessor attributes = attributesHolder.get();
AttributeAccessor attributes = ATTRIBUTES_HOLDER.get();
if (attributes == null) {
return super.getErrorMessageAttributes(message);
}
@@ -223,7 +223,6 @@ public class AmqpInboundChannelAdapter extends MessageProducerSupport implements
protected class Listener implements ChannelAwareMessageListener {
@SuppressWarnings("unchecked")
@Override
public void onMessage(final Message message, final Channel channel) {
boolean retryDisabled = AmqpInboundChannelAdapter.this.retryTemplate == null;
@@ -233,21 +232,21 @@ public class AmqpInboundChannelAdapter extends MessageProducerSupport implements
}
else {
final org.springframework.messaging.Message<Object> toSend = createMessage(message, channel);
AmqpInboundChannelAdapter.this.retryTemplate.execute(context -> {
AmqpInboundChannelAdapter.this.retryTemplate.execute(
context -> {
StaticMessageHeaderAccessor.getDeliveryAttempt(toSend).incrementAndGet();
setAttributesIfNecessary(message, toSend);
sendMessage(toSend);
return null;
},
(RecoveryCallback<Object>) AmqpInboundChannelAdapter.this.recoveryCallback);
}, AmqpInboundChannelAdapter.this.recoveryCallback);
}
}
catch (MessageConversionException e) {
if (getErrorChannel() != null) {
setAttributesIfNecessary(message, null);
getMessagingTemplate()
.send(getErrorChannel(), buildErrorMessage(null,
EndpointUtils.errorMessagePayload(message, channel, isManualAck(), e)));
.send(getErrorChannel(), buildErrorMessage(null,
EndpointUtils.errorMessagePayload(message, channel, isManualAck(), e)));
}
else {
throw e;
@@ -255,7 +254,7 @@ public class AmqpInboundChannelAdapter extends MessageProducerSupport implements
}
finally {
if (retryDisabled) {
attributesHolder.remove();
ATTRIBUTES_HOLDER.remove();
}
}
}
@@ -289,16 +288,15 @@ public class AmqpInboundChannelAdapter extends MessageProducerSupport implements
if (AmqpInboundChannelAdapter.this.bindSourceMessage) {
headers.put(IntegrationMessageHeaderAccessor.SOURCE_DATA, message);
}
final org.springframework.messaging.Message<Object> messagingMessage = getMessageBuilderFactory()
return getMessageBuilderFactory()
.withPayload(payload)
.copyHeaders(headers)
.build();
return messagingMessage;
}
private boolean isManualAck() {
return AmqpInboundChannelAdapter.this.messageListenerContainer.getAcknowledgeMode()
== AcknowledgeMode.MANUAL;
return AcknowledgeMode.MANUAL ==
AmqpInboundChannelAdapter.this.messageListenerContainer.getAcknowledgeMode();
}
}

14
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AsyncAmqpOutboundGateway.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2019 the original author or authors.
* Copyright 2016-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.
@@ -109,11 +109,11 @@ public class AsyncAmqpOutboundGateway extends AbstractAmqpOutboundEndpoint {
replyMessageBuilder = buildReply(AsyncAmqpOutboundGateway.this.messageConverter, result);
sendOutputs(replyMessageBuilder, this.requestMessage);
}
catch (Exception e) {
Exception exceptionToLogAndSend = e;
if (!(e instanceof MessagingException)) {
catch (Exception ex) {
Exception exceptionToLogAndSend = ex;
if (!(ex instanceof MessagingException)) { // NOSONAR
exceptionToLogAndSend = new MessageHandlingException(this.requestMessage,
"failed to handle a message in the [" + AsyncAmqpOutboundGateway.this + ']', e);
"failed to handle a message in the [" + AsyncAmqpOutboundGateway.this + ']', ex);
if (replyMessageBuilder != null) {
exceptionToLogAndSend =
new MessagingException(replyMessageBuilder.build(), exceptionToLogAndSend);
@@ -129,8 +129,8 @@ public class AsyncAmqpOutboundGateway extends AbstractAmqpOutboundEndpoint {
Throwable exceptionToSend = ex;
if (ex instanceof AmqpReplyTimeoutException) {
if (getRequiresReply()) {
exceptionToSend = new ReplyRequiredException(this.requestMessage, "Timeout on async request/reply",
ex);
exceptionToSend =
new ReplyRequiredException(this.requestMessage, "Timeout on async request/reply", ex);
}
else {
if (logger.isDebugEnabled()) {

62
spring-integration-core/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageBarrier.java
View File

@@ -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.
@@ -20,9 +20,6 @@ import java.util.Iterator;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.integration.core.MessageSource;
import org.springframework.integration.handler.AbstractMessageHandler;
import org.springframework.integration.store.MessageGroup;
@@ -33,11 +30,13 @@ import org.springframework.messaging.Message;
/**
* This Endpoint serves as a barrier for messages that should not be processed yet. The decision when a message can be
* processed is delegated to a {@link org.springframework.integration.aggregator.ReleaseStrategy ReleaseStrategy}.
* When a message can be processed it is up to the client to take care of the locking (potentially from the ReleaseStrategy's
* When a message can be processed it is up to the client to take care of the locking (potentially from the
* ReleaseStrategy's
* {@link org.springframework.integration.aggregator.ReleaseStrategy#canRelease(org.springframework.integration.store.MessageGroup) canRelease(..)}
* method).
* <p>
* This class differs from AbstractCorrelatingMessageHandler in that it completely decouples the receiver and the sender. It can
* This class differs from AbstractCorrelatingMessageHandler in that it completely decouples the receiver and the
* sender. It can
* be applied in scenarios where completion of a message group is not well defined but only a certain amount of messages
* for any given correlation key may be processed at a time.
* <p>
@@ -47,34 +46,32 @@ import org.springframework.messaging.Message;
* @author Iwein Fuld
* @author Oleg Zhurakousky
* @author Gary Russell
* @author Artem Bilan
*
* @see AbstractCorrelatingMessageHandler
*/
public class CorrelatingMessageBarrier extends AbstractMessageHandler implements MessageSource<Object> {
private static final Log log = LogFactory.getLog(CorrelatingMessageBarrier.class);
private volatile CorrelationStrategy correlationStrategy;
private volatile ReleaseStrategy releaseStrategy;
private final ConcurrentMap<Object, Object> correlationLocks = new ConcurrentHashMap<Object, Object>();
private final ConcurrentMap<Object, Object> correlationLocks = new ConcurrentHashMap<>();
private final MessageGroupStore store;
private CorrelationStrategy correlationStrategy;
private ReleaseStrategy releaseStrategy;
public CorrelatingMessageBarrier(MessageGroupStore store) {
this.store = store;
}
public CorrelatingMessageBarrier() {
this(new SimpleMessageStore(0));
}
public CorrelatingMessageBarrier(MessageGroupStore store) {
this.store = store;
}
/**
* Set the CorrelationStrategy to be used to determine the correlation key for incoming messages
*
* @param correlationStrategy The correlation strategy.
*/
public void setCorrelationStrategy(CorrelationStrategy correlationStrategy) {
@@ -83,7 +80,6 @@ public class CorrelatingMessageBarrier extends AbstractMessageHandler implements
/**
* Set the ReleaseStrategy that should be used when deciding if a group in this barrier may be released.
*
* @param releaseStrategy The release strategy.
*/
public void setReleaseStrategy(ReleaseStrategy releaseStrategy) {
@@ -97,8 +93,8 @@ public class CorrelatingMessageBarrier extends AbstractMessageHandler implements
synchronized (lock) {
this.store.addMessagesToGroup(correlationKey, message);
}
if (log.isDebugEnabled()) {
log.debug(String.format("Handled message for key [%s]: %s.", correlationKey, message));
if (logger.isDebugEnabled()) {
logger.debug(String.format("Handled message for key [%s]: %s.", correlationKey, message));
}
}
@@ -116,23 +112,21 @@ public class CorrelatingMessageBarrier extends AbstractMessageHandler implements
synchronized (lock) {
MessageGroup group = this.store.getMessageGroup(key);
//group might be removed by another thread
if (group != null) {
if (this.releaseStrategy.canRelease(group)) {
Message<?> nextMessage = null;
if (group != null && this.releaseStrategy.canRelease(group)) {
Message<?> nextMessage = null;
Iterator<Message<?>> messages = group.getMessages().iterator();
if (messages.hasNext()) {
nextMessage = messages.next();
this.store.removeMessagesFromGroup(key, nextMessage);
if (log.isDebugEnabled()) {
log.debug(String.format("Released message for key [%s]: %s.", key, nextMessage));
}
Iterator<Message<?>> messages = group.getMessages().iterator();
if (messages.hasNext()) {
nextMessage = messages.next();
this.store.removeMessagesFromGroup(key, nextMessage);
if (logger.isDebugEnabled()) {
logger.debug(String.format("Released message for key [%s]: %s.", key, nextMessage));
}
else {
remove(key);
}
return (Message<Object>) nextMessage;
}
else {
remove(key);
}
return (Message<Object>) nextMessage;
}
}
}

15
spring-integration-core/src/main/java/org/springframework/integration/aggregator/ExpressionEvaluatingCorrelationStrategy.java
View File

@@ -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.
@@ -36,19 +36,20 @@ import org.springframework.util.Assert;
*/
public class ExpressionEvaluatingCorrelationStrategy implements CorrelationStrategy, BeanFactoryAware {
private static final ExpressionParser expressionParser = new SpelExpressionParser(new SpelParserConfiguration(true, true));
private static final ExpressionParser EXPRESSION_PARSER =
new SpelExpressionParser(new SpelParserConfiguration(true, true));
private final ExpressionEvaluatingMessageProcessor<Object> processor;
public ExpressionEvaluatingCorrelationStrategy(String expressionString) {
Assert.hasText(expressionString, "expressionString must not be empty");
Expression expression = expressionParser.parseExpression(expressionString);
this.processor = new ExpressionEvaluatingMessageProcessor<Object>(expression, Object.class);
Expression expression = EXPRESSION_PARSER.parseExpression(expressionString);
this.processor = new ExpressionEvaluatingMessageProcessor<>(expression, Object.class);
}
public ExpressionEvaluatingCorrelationStrategy(Expression expression) {
this.processor = new ExpressionEvaluatingMessageProcessor<Object>(expression, Object.class);
this.processor = new ExpressionEvaluatingMessageProcessor<>(expression, Object.class);
}
public Object getCorrelationKey(Message<?> message) {
@@ -56,9 +57,7 @@ public class ExpressionEvaluatingCorrelationStrategy implements CorrelationStrat
}
public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
if (beanFactory != null) {
this.processor.setBeanFactory(beanFactory);
}
this.processor.setBeanFactory(beanFactory);
}
}

16
spring-integration-core/src/main/java/org/springframework/integration/aggregator/SequenceSizeReleaseStrategy.java
View File

@@ -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.
@@ -23,7 +23,7 @@ import java.util.Comparator;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.integration.IntegrationMessageHeaderAccessor;
import org.springframework.integration.StaticMessageHeaderAccessor;
import org.springframework.integration.store.MessageGroup;
import org.springframework.messaging.Message;
@@ -38,11 +38,11 @@ import org.springframework.messaging.Message;
* @author Iwein Fuld
* @author Oleg Zhurakousky
* @author Artem Bilan
* @author Enrique Rodr?guez
* @author Enrique Rodriguez
*/
public class SequenceSizeReleaseStrategy implements ReleaseStrategy {
private static final Log logger = LogFactory.getLog(SequenceSizeReleaseStrategy.class);
private static final Log LOGGER = LogFactory.getLog(SequenceSizeReleaseStrategy.class);
private final Comparator<Message<?>> comparator = new MessageSequenceComparator();
@@ -78,18 +78,16 @@ public class SequenceSizeReleaseStrategy implements ReleaseStrategy {
@Override
public boolean canRelease(MessageGroup messageGroup) {
boolean canRelease = false;
int size = messageGroup.size();
if (this.releasePartialSequences && size > 0) {
if (logger.isTraceEnabled()) {
logger.trace("Considering partial release of group [" + messageGroup + "]");
if (LOGGER.isTraceEnabled()) {
LOGGER.trace("Considering partial release of group [" + messageGroup + "]");
}
Collection<Message<?>> messages = messageGroup.getMessages();
Message<?> minMessage = Collections.min(messages, this.comparator);
int nextSequenceNumber = new IntegrationMessageHeaderAccessor(minMessage).getSequenceNumber();
int nextSequenceNumber = StaticMessageHeaderAccessor.getSequenceNumber(minMessage);
int lastReleasedMessageSequence = messageGroup.getLastReleasedMessageSequenceNumber();
if (nextSequenceNumber - lastReleasedMessageSequence == 1) {

41
spring-integration-core/src/main/java/org/springframework/integration/aop/PublisherAnnotationAdvisor.java
View File

@@ -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.
@@ -17,7 +17,6 @@
package org.springframework.integration.aop;
import java.lang.annotation.Annotation;
import java.lang.reflect.Method;
import java.util.Arrays;
import java.util.stream.Collectors;
@@ -27,13 +26,11 @@ import org.springframework.aop.ClassFilter;
import org.springframework.aop.MethodMatcher;
import org.springframework.aop.Pointcut;
import org.springframework.aop.support.AbstractPointcutAdvisor;
import org.springframework.aop.support.AopUtils;
import org.springframework.aop.support.ComposablePointcut;
import org.springframework.aop.support.annotation.AnnotationClassFilter;
import org.springframework.aop.support.annotation.AnnotationMethodMatcher;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.integration.annotation.Publisher;
import org.springframework.util.Assert;
@@ -48,10 +45,11 @@ import org.springframework.util.Assert;
*
* @since 2.0
*/
@SuppressWarnings("serial")
public class PublisherAnnotationAdvisor extends AbstractPointcutAdvisor implements BeanFactoryAware {
private final MessagePublishingInterceptor interceptor;
private static final long serialVersionUID = -5387975397101845222L;
private final transient MessagePublishingInterceptor interceptor;
private final Pointcut pointcut;
@@ -150,7 +148,7 @@ public class PublisherAnnotationAdvisor extends AbstractPointcutAdvisor implemen
}
if (methodAnnotationType != null) {
this.methodMatcher = new MetaAnnotationMethodMatcher(methodAnnotationType);
this.methodMatcher = new AnnotationMethodMatcher(methodAnnotationType, true);
}
else {
this.methodMatcher = MethodMatcher.TRUE;
@@ -170,33 +168,4 @@ public class PublisherAnnotationAdvisor extends AbstractPointcutAdvisor implemen
}
private static final class MetaAnnotationMethodMatcher extends AnnotationMethodMatcher {
private final Class<? extends Annotation> annotationType;
/**
* Create a new AnnotationClassFilter for the given annotation type.
* @param annotationType the annotation type to look for
*/
MetaAnnotationMethodMatcher(Class<? extends Annotation> annotationType) {
super(annotationType);
this.annotationType = annotationType;
}
@Override
public boolean matches(Method method, Class<?> targetClass) {
if (AnnotationUtils.getAnnotation(method, this.annotationType) != null) {
return true;
}
// The method may be on an interface, so let's check on the target class as well.
Method specificMethod = AopUtils.getMostSpecificMethod(method, targetClass);
return (!specificMethod.equals(method) &&
(AnnotationUtils.getAnnotation(specificMethod, this.annotationType) != null));
}
}
}

10
spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractExecutorChannel.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2015-2019 the original author or authors.
* Copyright 2015-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.
@@ -57,13 +57,13 @@ import org.springframework.util.CollectionUtils;
public abstract class AbstractExecutorChannel extends AbstractSubscribableChannel
implements ExecutorChannelInterceptorAware {
protected volatile Executor executor; // NOSONAR
protected Executor executor; // NOSONAR
protected volatile AbstractDispatcher dispatcher; // NOSONAR
protected AbstractDispatcher dispatcher; // NOSONAR
protected volatile Integer maxSubscribers; // NOSONAR
protected Integer maxSubscribers; // NOSONAR
protected volatile int executorInterceptorsSize; // NOSONAR
protected int executorInterceptorsSize; // NOSONAR
public AbstractExecutorChannel(@Nullable Executor executor) {
this.executor = executor;

17
spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractSubscribableChannel.java
View File

@@ -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.
@@ -33,6 +33,7 @@ import org.springframework.util.Assert;
* @author Mark Fisher
* @author Oleg Zhurakousky
* @author Gary Russell
* @author Artme Bilan
*/
public abstract class AbstractSubscribableChannel extends AbstractMessageChannel
implements SubscribableChannel, SubscribableChannelManagement {
@@ -59,11 +60,9 @@ public abstract class AbstractSubscribableChannel extends AbstractMessageChannel
}
private void adjustCounterIfNecessary(MessageDispatcher dispatcher, int delta) {
if (delta != 0) {
if (logger.isInfoEnabled()) {
logger.info("Channel '" + this.getFullChannelName() + "' has " + dispatcher.getHandlerCount()
+ " subscriber(s).");
}
if (delta != 0 && logger.isInfoEnabled()) {
logger.info("Channel '" + getFullChannelName() + "' has " + dispatcher.getHandlerCount()
+ " subscriber(s).");
}
}
@@ -72,9 +71,9 @@ public abstract class AbstractSubscribableChannel extends AbstractMessageChannel
try {
return getRequiredDispatcher().dispatch(message);
}
catch (MessageDispatchingException e) {
String description = e.getMessage() + " for channel '" + this.getFullChannelName() + "'.";
throw new MessageDeliveryException(message, description, e);
catch (MessageDispatchingException ex) {
String description = ex.getMessage() + " for channel '" + getFullChannelName() + "'.";
throw new MessageDeliveryException(message, description, ex);
}
}

11
spring-integration-core/src/main/java/org/springframework/integration/channel/FixedSubscriberChannel.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2014-2019 the original author or authors.
* Copyright 2014-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.
@@ -77,13 +77,12 @@ public final class FixedSubscriberChannel implements SubscribableChannel, BeanNa
this.handler.handleMessage(message);
return true;
}
catch (RuntimeException e) {
if (e instanceof MessagingException &&
((MessagingException) e).getFailedMessage() == null) {
throw new MessagingException(message, "Failed to handle Message", e);
catch (MessagingException ex) {
if (ex.getFailedMessage() == null) {
throw new MessagingException(message, "Failed to handle Message", ex);
}
else {
throw e;
throw ex;
}
}
}

30
spring-integration-core/src/main/java/org/springframework/integration/channel/PriorityChannel.java
View File

@@ -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.
@@ -20,7 +20,7 @@ import java.util.Comparator;
import java.util.concurrent.PriorityBlockingQueue;
import java.util.concurrent.atomic.AtomicLong;
import org.springframework.integration.IntegrationMessageHeaderAccessor;
import org.springframework.integration.StaticMessageHeaderAccessor;
import org.springframework.integration.store.MessageGroupQueue;
import org.springframework.integration.store.PriorityCapableChannelMessageStore;
import org.springframework.integration.util.UpperBound;
@@ -39,6 +39,11 @@ import org.springframework.messaging.MessageHeaders;
*/
public class PriorityChannel extends QueueChannel {
/**
* PriorityBlockingQueue#DEFAULT_INITIAL_CAPACITY is private
*/
private static final int DEFAULT_INITIAL_CAPACITY = 11;
private final UpperBound upperBound;
private final AtomicLong sequenceCounter = new AtomicLong();
@@ -47,7 +52,7 @@ public class PriorityChannel extends QueueChannel {
/**
* Create a channel with an unbounded queue. Message priority will be
* based on the value of {@link IntegrationMessageHeaderAccessor#getPriority()}.
* based on the value of {@link StaticMessageHeaderAccessor#getPriority(Message)}.
*/
public PriorityChannel() {
this(0, null);
@@ -55,8 +60,7 @@ public class PriorityChannel extends QueueChannel {
/**
* Create a channel with the specified queue capacity. Message priority
* will be based upon the value of {@link IntegrationMessageHeaderAccessor#getPriority()}.
*
* will be based upon the value of {@link StaticMessageHeaderAccessor#getPriority(Message)}.
* @param capacity The queue capacity.
*/
public PriorityChannel(int capacity) {
@@ -67,8 +71,7 @@ public class PriorityChannel extends QueueChannel {
* Create a channel with an unbounded queue. Message priority will be
* determined by the provided {@link Comparator}. If the comparator
* is <code>null</code>, the priority will be based upon the value of
* {@link IntegrationMessageHeaderAccessor#getPriority()}.
*
* {@link StaticMessageHeaderAccessor#getPriority(Message)}.
* @param comparator The comparator.
*/
public PriorityChannel(Comparator<Message<?>> comparator) {
@@ -80,13 +83,12 @@ public class PriorityChannel extends QueueChannel {
* is a non-positive value, the queue will be unbounded. Message priority
* will be determined by the provided {@link Comparator}. If the comparator
* is <code>null</code>, the priority will be based upon the value of
* {@link IntegrationMessageHeaderAccessor#getPriority()}.
*
* {@link StaticMessageHeaderAccessor#getPriority(Message)}.
* @param capacity The capacity.
* @param comparator The comparator.
*/
public PriorityChannel(int capacity, @Nullable Comparator<Message<?>> comparator) {
super(new PriorityBlockingQueue<>(11, new SequenceFallbackComparator(comparator)));
super(new PriorityBlockingQueue<>(DEFAULT_INITIAL_CAPACITY, new SequenceFallbackComparator(comparator)));
this.upperBound = new UpperBound(capacity);
this.useMessageStore = false;
}
@@ -147,19 +149,19 @@ public class PriorityChannel extends QueueChannel {
private final Comparator<Message<?>> targetComparator;
SequenceFallbackComparator(Comparator<Message<?>> targetComparator) {
SequenceFallbackComparator(@Nullable Comparator<Message<?>> targetComparator) {
this.targetComparator = targetComparator;
}
@Override
public int compare(Message<?> message1, Message<?> message2) {
int compareResult = 0;
int compareResult;
if (this.targetComparator != null) {
compareResult = this.targetComparator.compare(message1, message2);
}
else {
Integer priority1 = new IntegrationMessageHeaderAccessor(message1).getPriority();
Integer priority2 = new IntegrationMessageHeaderAccessor(message2).getPriority();
Integer priority1 = StaticMessageHeaderAccessor.getPriority(message1);
Integer priority2 = StaticMessageHeaderAccessor.getPriority(message2);
priority1 = priority1 != null ? priority1 : 0;
priority2 = priority2 != null ? priority2 : 0;

13
spring-integration-core/src/main/java/org/springframework/integration/channel/PublishSubscribeChannel.java
View File

@@ -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.
@@ -122,7 +122,6 @@ public class PublishSubscribeChannel extends AbstractExecutorChannel {
* If at least this number of subscribers receive the message,
* {@link #send(org.springframework.messaging.Message)}
* will return true. Default: 0.
*
* @param minSubscribers The minimum number of subscribers.
*/
public void setMinSubscribers(int minSubscribers) {
@@ -154,12 +153,10 @@ public class PublishSubscribeChannel extends AbstractExecutorChannel {
dispatcherToUse.setMinSubscribers(this.minSubscribers);
this.dispatcher = dispatcherToUse;
}
else if (this.errorHandler != null) {
if (this.logger.isWarnEnabled()) {
this.logger.warn("The 'errorHandler' is ignored for the '" + getComponentName() +
"' (an 'executor' is not provided) and exceptions will be thrown " +
"directly within the sending Thread");
}
else if (this.errorHandler != null && this.logger.isWarnEnabled()) {
this.logger.warn("The 'errorHandler' is ignored for the '" + getComponentName() +
"' (an 'executor' is not provided) and exceptions will be thrown " +
"directly within the sending Thread");
}
if (this.maxSubscribers == null) {

26
spring-integration-core/src/main/java/org/springframework/integration/channel/interceptor/WireTap.java
View File

@@ -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.
@@ -45,20 +45,20 @@ import org.springframework.util.Assert;
@ManagedResource
public class WireTap implements ChannelInterceptor, Lifecycle, VetoCapableInterceptor, BeanFactoryAware {
private static final Log logger = LogFactory.getLog(WireTap.class);
private volatile MessageChannel channel;
private volatile String channelName;
private volatile long timeout = 0;
private static final Log LOGGER = LogFactory.getLog(WireTap.class);
private final MessageSelector selector;
private volatile boolean running = true;
private MessageChannel channel;
private String channelName;
private long timeout = 0;
private BeanFactory beanFactory;
private volatile boolean running = true;
/**
* Create a new wire tap with <em>no</em> {@link MessageSelector}.
@@ -156,8 +156,8 @@ public class WireTap implements ChannelInterceptor, Lifecycle, VetoCapableInterc
public Message<?> preSend(Message<?> message, MessageChannel channel) {
MessageChannel wireTapChannel = getChannel();
if (wireTapChannel.equals(channel)) {
if (logger.isDebugEnabled()) {
logger.debug("WireTap is refusing to intercept its own channel '" + wireTapChannel + "'");
if (LOGGER.isDebugEnabled()) {
LOGGER.debug("WireTap is refusing to intercept its own channel '" + wireTapChannel + "'");
}
return message;
}
@@ -165,8 +165,8 @@ public class WireTap implements ChannelInterceptor, Lifecycle, VetoCapableInterc
boolean sent = (this.timeout >= 0)
? wireTapChannel.send(message, this.timeout)
: wireTapChannel.send(message);
if (!sent && logger.isWarnEnabled()) {
logger.warn("failed to send message to WireTap channel '" + wireTapChannel + "'");
if (!sent && LOGGER.isWarnEnabled()) {
LOGGER.warn("failed to send message to WireTap channel '" + wireTapChannel + "'");
}
}
return message;

14
spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/KryoClassListRegistrar.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2015-2019 the original author or authors.
* Copyright 2015-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.
@@ -31,23 +31,27 @@ import com.esotericsoftware.kryo.Registration;
* list in the same order.
*
* @author David Turanski
* @author Artem Bilan
*
* @since 4.2
*/
public class KryoClassListRegistrar extends AbstractKryoRegistrar {
private static final int DEFAULT_INITIAL_ID = 50;
private final List<Class<?>> registeredClasses;
private int initialValue = 50;
private int initialValue = DEFAULT_INITIAL_ID;
/**
* @param classes the list of classes to validateRegistration
*/
public KryoClassListRegistrar(List<Class<?>> classes) {
this.registeredClasses = new ArrayList<Class<?>>(classes);
this.registeredClasses = new ArrayList<>(classes);
}
/**
* Set the inital ID value. Classes in the list will be sequentially assigned an ID starting with this value
* Set the initial ID value. Classes in the list will be sequentially assigned an ID starting with this value
* (default is 50).
* @param initialValue the initial value
*/
@@ -60,7 +64,7 @@ public class KryoClassListRegistrar extends AbstractKryoRegistrar {
@Override
public List<Registration> getRegistrations() {
List<Registration> registrations = new ArrayList<Registration>();
List<Registration> registrations = new ArrayList<>();
if (!CollectionUtils.isEmpty(this.registeredClasses)) {
for (int i = 0; i < this.registeredClasses.size(); i++) {
registrations.add(new Registration(this.registeredClasses.get(i),

14
spring-integration-core/src/main/java/org/springframework/integration/config/AbstractStandardMessageHandlerFactoryBean.java
View File

@@ -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.
@@ -43,9 +43,9 @@ import org.springframework.util.StringUtils;
public abstract class AbstractStandardMessageHandlerFactoryBean
extends AbstractSimpleMessageHandlerFactoryBean<MessageHandler> implements DisposableBean {
private static final ExpressionParser expressionParser = new SpelExpressionParser();
private static final ExpressionParser EXPRESSION_PARSER = new SpelExpressionParser();
private static final Set<MessageHandler> referencedReplyProducers = new HashSet<>();
private static final Set<MessageHandler> REFERENCED_REPLY_PRODUCERS = new HashSet<>();
private Boolean requiresReply;
@@ -80,7 +80,7 @@ public abstract class AbstractStandardMessageHandlerFactoryBean
* @param expressionString the expression as a String.
*/
public void setExpressionString(String expressionString) {
this.expression = expressionParser.parseExpression(expressionString);
this.expression = EXPRESSION_PARSER.parseExpression(expressionString);
}
/**
@@ -106,7 +106,7 @@ public abstract class AbstractStandardMessageHandlerFactoryBean
@Override
public void destroy() {
if (this.replyHandler != null) {
referencedReplyProducers.remove(this.replyHandler);
REFERENCED_REPLY_PRODUCERS.remove(this.replyHandler);
}
}
@@ -165,10 +165,10 @@ public abstract class AbstractStandardMessageHandlerFactoryBean
}
private void checkReuse(AbstractMessageProducingHandler replyHandler) {
Assert.isTrue(!referencedReplyProducers.contains(replyHandler),
Assert.isTrue(!REFERENCED_REPLY_PRODUCERS.contains(replyHandler),
"An AbstractMessageProducingMessageHandler may only be referenced once (" +
replyHandler.getBeanName() + ") - use scope=\"prototype\"");
referencedReplyProducers.add(replyHandler);
REFERENCED_REPLY_PRODUCERS.add(replyHandler);
this.replyHandler = replyHandler;
}

14
spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java
View File

@@ -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.
@@ -74,7 +74,7 @@ public class ConsumerEndpointFactoryBean
implements FactoryBean<AbstractEndpoint>, BeanFactoryAware, BeanNameAware, BeanClassLoaderAware,
InitializingBean, SmartLifecycle, DisposableBean {
private static final Log logger = LogFactory.getLog(ConsumerEndpointFactoryBean.class);
private static final Log LOGGER = LogFactory.getLog(ConsumerEndpointFactoryBean.class);
private final Object initializationMonitor = new Object();
@@ -184,7 +184,7 @@ public class ConsumerEndpointFactoryBean
@Override
public void afterPropertiesSet() {
if (this.beanName == null) {
logger.error("The MessageHandler [" + this.handler + "] will be created without a 'componentName'. " +
LOGGER.error("The MessageHandler [" + this.handler + "] will be created without a 'componentName'. " +
"Consider specifying the 'beanName' property on this ConsumerEndpointFactoryBean.");
}
else {
@@ -203,8 +203,8 @@ public class ConsumerEndpointFactoryBean
}
}
catch (Exception e) {
if (logger.isDebugEnabled()) {
logger.debug("Could not set component name for handler "
if (LOGGER.isDebugEnabled()) {
LOGGER.debug("Could not set component name for handler "
+ this.handler + " for " + this.beanName + " :" + e.getMessage());
}
}
@@ -301,10 +301,10 @@ public class ConsumerEndpointFactoryBean
() -> "A poller should not be specified for endpoint '" + this.beanName
+ "', since '" + channel + "' is a SubscribableChannel (not pollable).");
this.endpoint = new EventDrivenConsumer((SubscribableChannel) channel, this.handler);
if (logger.isWarnEnabled()
if (LOGGER.isWarnEnabled()
&& Boolean.FALSE.equals(this.autoStartup)
&& channel instanceof FixedSubscriberChannel) {
logger.warn("'autoStartup=\"false\"' has no effect when using a FixedSubscriberChannel");
LOGGER.warn("'autoStartup=\"false\"' has no effect when using a FixedSubscriberChannel");
}
}

46
spring-integration-core/src/main/java/org/springframework/integration/config/DefaultConfiguringBeanFactoryPostProcessor.java
View File

@@ -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.
@@ -92,12 +92,12 @@ import org.springframework.util.ClassUtils;
class DefaultConfiguringBeanFactoryPostProcessor
implements BeanFactoryPostProcessor, BeanClassLoaderAware, SmartInitializingSingleton {
private static final Log logger = LogFactory.getLog(DefaultConfiguringBeanFactoryPostProcessor.class);
private static final Log LOGGER = LogFactory.getLog(DefaultConfiguringBeanFactoryPostProcessor.class);
private static final IntegrationConverterInitializer INTEGRATION_CONVERTER_INITIALIZER =
new IntegrationConverterInitializer();
private static final Set<Integer> registriesProcessed = new HashSet<>();
private static final Set<Integer> REGISTRIES_PROCESSED = new HashSet<>();
private ClassLoader classLoader;
@@ -134,22 +134,22 @@ class DefaultConfiguringBeanFactoryPostProcessor
registerMessageHandlerMethodFactory();
registerListMessageHandlerMethodFactory();
}
else if (logger.isWarnEnabled()) {
logger.warn("BeanFactory is not a BeanDefinitionRegistry. " +
else if (LOGGER.isWarnEnabled()) {
LOGGER.warn("BeanFactory is not a BeanDefinitionRegistry. " +
"The default Spring Integration infrastructure beans are not going to be registered");
}
}
@Override
public void afterSingletonsInstantiated() {
if (logger.isDebugEnabled()) {
if (LOGGER.isDebugEnabled()) {
Properties integrationProperties = IntegrationContextUtils.getIntegrationProperties(this.beanFactory);
StringWriter writer = new StringWriter();
integrationProperties.list(new PrintWriter(writer));
StringBuffer propertiesBuffer = writer.getBuffer()
.delete(0, "-- listing properties --".length());
logger.debug("\nSpring Integration global properties:\n" + propertiesBuffer);
LOGGER.debug("\nSpring Integration global properties:\n" + propertiesBuffer);
}
}
@@ -209,8 +209,8 @@ class DefaultConfiguringBeanFactoryPostProcessor
*/
private void registerErrorChannel() {
if (!this.beanFactory.containsBean(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME)) {
if (logger.isInfoEnabled()) {
logger.info("No bean named '" + IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME +
if (LOGGER.isInfoEnabled()) {
LOGGER.info("No bean named '" + IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME +
"' has been explicitly defined. " +
"Therefore, a default PublishSubscribeChannel will be created.");
}
@@ -280,8 +280,8 @@ class DefaultConfiguringBeanFactoryPostProcessor
for (String definitionName : definitionNames) {
BeanDefinition definition = this.registry.getBeanDefinition(definitionName);
if (className.equals(definition.getBeanClassName())) {
if (logger.isInfoEnabled()) {
logger.info(className + " is already registered and will be used");
if (LOGGER.isInfoEnabled()) {
LOGGER.info(className + " is already registered and will be used");
}
return;
}
@@ -296,8 +296,8 @@ class DefaultConfiguringBeanFactoryPostProcessor
*/
private void registerTaskScheduler() {
if (!this.beanFactory.containsBean(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME)) {
if (logger.isInfoEnabled()) {
logger.info("No bean named '" + IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME +
if (LOGGER.isInfoEnabled()) {
LOGGER.info("No bean named '" + IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME +
"' has been explicitly defined. " +
"Therefore, a default ThreadPoolTaskScheduler will be created.");
}
@@ -337,7 +337,7 @@ class DefaultConfiguringBeanFactoryPostProcessor
integrationPropertiesBuilder.getBeanDefinition());
}
catch (IOException e) {
logger.warn("Cannot load 'spring.integration.properties' Resources.", e);
LOGGER.warn("Cannot load 'spring.integration.properties' Resources.", e);
}
}
}
@@ -350,18 +350,18 @@ class DefaultConfiguringBeanFactoryPostProcessor
jsonPath(registryId);
xpath(registryId);
jsonNodeToString(registryId);
registriesProcessed.add(registryId);
REGISTRIES_PROCESSED.add(registryId);
}
private void jsonPath(int registryId) throws LinkageError {
String jsonPathBeanName = "jsonPath";
if (!this.beanFactory.containsBean(jsonPathBeanName) && !registriesProcessed.contains(registryId)) {
if (!this.beanFactory.containsBean(jsonPathBeanName) && !REGISTRIES_PROCESSED.contains(registryId)) {
Class<?> jsonPathClass = null;
try {
jsonPathClass = ClassUtils.forName("com.jayway.jsonpath.JsonPath", this.classLoader);
}
catch (@SuppressWarnings("unused") ClassNotFoundException e) {
logger.debug("The '#jsonPath' SpEL function cannot be registered: " +
LOGGER.debug("The '#jsonPath' SpEL function cannot be registered: " +
"there is no jayway json-path.jar on the classpath.");
}
@@ -371,7 +371,7 @@ class DefaultConfiguringBeanFactoryPostProcessor
}
catch (ClassNotFoundException e) {
jsonPathClass = null;
logger.warn("The '#jsonPath' SpEL function cannot be registered. " +
LOGGER.warn("The '#jsonPath' SpEL function cannot be registered. " +
"An old json-path.jar version is detected in the classpath." +
"At least 2.4.0 is required; see version information at: " +
"https://github.com/jayway/JsonPath/releases", e);
@@ -388,14 +388,14 @@ class DefaultConfiguringBeanFactoryPostProcessor
private void xpath(int registryId) throws LinkageError {
String xpathBeanName = "xpath";
if (!this.beanFactory.containsBean(xpathBeanName) && !registriesProcessed.contains(registryId)) {
if (!this.beanFactory.containsBean(xpathBeanName) && !REGISTRIES_PROCESSED.contains(registryId)) {
Class<?> xpathClass = null;
try {
xpathClass = ClassUtils.forName(IntegrationConfigUtils.BASE_PACKAGE + ".xml.xpath.XPathUtils",
this.classLoader);
}
catch (@SuppressWarnings("unused") ClassNotFoundException e) {
logger.debug("SpEL function '#xpath' isn't registered: " +
LOGGER.debug("SpEL function '#xpath' isn't registered: " +
"there is no spring-integration-xml.jar on the classpath.");
}
@@ -409,7 +409,7 @@ class DefaultConfiguringBeanFactoryPostProcessor
private void jsonNodeToString(int registryId) {
if (!this.beanFactory.containsBean(
IntegrationContextUtils.TO_STRING_FRIENDLY_JSON_NODE_TO_STRING_CONVERTER_BEAN_NAME) &&
!registriesProcessed.contains(registryId) && JacksonPresent.isJackson2Present()) {
!REGISTRIES_PROCESSED.contains(registryId) && JacksonPresent.isJackson2Present()) {
this.registry.registerBeanDefinition(
IntegrationContextUtils.TO_STRING_FRIENDLY_JSON_NODE_TO_STRING_CONVERTER_BEAN_NAME,
@@ -456,8 +456,8 @@ class DefaultConfiguringBeanFactoryPostProcessor
*/
private void registerHeaderChannelRegistry() {
if (!this.beanFactory.containsBean(IntegrationContextUtils.INTEGRATION_HEADER_CHANNEL_REGISTRY_BEAN_NAME)) {
if (logger.isInfoEnabled()) {
logger.info("No bean named '" + IntegrationContextUtils.INTEGRATION_HEADER_CHANNEL_REGISTRY_BEAN_NAME +
if (LOGGER.isInfoEnabled()) {
LOGGER.info("No bean named '" + IntegrationContextUtils.INTEGRATION_HEADER_CHANNEL_REGISTRY_BEAN_NAME +
"' has been explicitly defined. " +
"Therefore, a default DefaultHeaderChannelRegistry will be created.");
}

12
spring-integration-core/src/main/java/org/springframework/integration/config/ExpressionControlBusFactoryBean.java
View File

@@ -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.
@@ -28,14 +28,15 @@ import org.springframework.messaging.MessageHandler;
* @author Dave Syer
* @author Mark Fisher
* @author Oleg Zhurakousky
* @author Artem Bilan
*
* @since 2.0
*/
public class ExpressionControlBusFactoryBean extends AbstractSimpleMessageHandlerFactoryBean<MessageHandler> {
private static final MethodFilter methodFilter = new ControlBusMethodFilter();
private volatile Long sendTimeout;
private static final MethodFilter METHOD_FILTER = new ControlBusMethodFilter();
private Long sendTimeout;
public void setSendTimeout(Long sendTimeout) {
this.sendTimeout = sendTimeout;
@@ -44,7 +45,7 @@ public class ExpressionControlBusFactoryBean extends AbstractSimpleMessageHandle
@Override
protected MessageHandler createHandler() {
ExpressionCommandMessageProcessor processor =
new ExpressionCommandMessageProcessor(methodFilter, this.getBeanFactory());
new ExpressionCommandMessageProcessor(METHOD_FILTER, getBeanFactory());
ServiceActivatingHandler handler = new ServiceActivatingHandler(processor);
if (this.sendTimeout != null) {
handler.setSendTimeout(this.sendTimeout);
@@ -52,5 +53,4 @@ public class ExpressionControlBusFactoryBean extends AbstractSimpleMessageHandle
return handler;
}
}

12
spring-integration-core/src/main/java/org/springframework/integration/config/GlobalChannelInterceptorProcessor.java
View File

@@ -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.
@@ -56,8 +56,7 @@ import org.springframework.util.StringUtils;
public final class GlobalChannelInterceptorProcessor
implements BeanFactoryAware, SmartInitializingSingleton, BeanPostProcessor {
private static final Log logger = LogFactory.getLog(GlobalChannelInterceptorProcessor.class);
private static final Log LOGGER = LogFactory.getLog(GlobalChannelInterceptorProcessor.class);
private final OrderComparator comparator = new OrderComparator();
@@ -76,12 +75,11 @@ public final class GlobalChannelInterceptorProcessor
}
@Override
@SuppressWarnings("deprecation")
public void afterSingletonsInstantiated() {
Collection<GlobalChannelInterceptorWrapper> interceptors =
this.beanFactory.getBeansOfType(GlobalChannelInterceptorWrapper.class).values();
if (CollectionUtils.isEmpty(interceptors)) {
logger.debug("No global channel interceptors.");
LOGGER.debug("No global channel interceptors.");
}
else {
interceptors.forEach(interceptor -> {
@@ -115,8 +113,8 @@ public final class GlobalChannelInterceptorProcessor
*/
public void addMatchingInterceptors(InterceptableChannel channel, String beanName) {
if (logger.isDebugEnabled()) {
logger.debug("Applying global interceptors on channel '" + beanName + "'");
if (LOGGER.isDebugEnabled()) {
LOGGER.debug("Applying global interceptors on channel '" + beanName + "'");
}
List<GlobalChannelInterceptorWrapper> tempInterceptors = new ArrayList<>();

30
spring-integration-core/src/main/java/org/springframework/integration/config/IdGeneratorConfigurer.java
View File

@@ -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.
@@ -38,11 +38,13 @@ import org.springframework.util.ReflectionUtils;
* @author Oleg Zhurakousky
* @author Mark Fisher
* @author Gary Russell
* @author Artem Bilan
*
* @since 2.0.4
*/
public final class IdGeneratorConfigurer implements ApplicationListener<ApplicationContextEvent> {
private static final Set<String> generatorContextId = new HashSet<String>();
private static final Set<String> GENERATOR_CONTEXT_ID = new HashSet<>();
private static volatile IdGenerator theIdGenerator;
@@ -53,19 +55,17 @@ public final class IdGeneratorConfigurer implements ApplicationListener<Applicat
ApplicationContext context = event.getApplicationContext();
if (event instanceof ContextRefreshedEvent) {
boolean contextHasIdGenerator = context.getBeanNamesForType(IdGenerator.class).length > 0;
if (contextHasIdGenerator) {
if (this.setIdGenerator(context)) {
IdGeneratorConfigurer.generatorContextId.add(context.getId());
}
if (contextHasIdGenerator && setIdGenerator(context)) {
IdGeneratorConfigurer.GENERATOR_CONTEXT_ID.add(context.getId());
}
}
else if (event instanceof ContextClosedEvent) {
if (IdGeneratorConfigurer.generatorContextId.contains(context.getId())) {
if (IdGeneratorConfigurer.generatorContextId.size() == 1) {
this.unsetIdGenerator();
}
IdGeneratorConfigurer.generatorContextId.remove(context.getId());
else if (event instanceof ContextClosedEvent
&& IdGeneratorConfigurer.GENERATOR_CONTEXT_ID.contains(context.getId())) {
if (IdGeneratorConfigurer.GENERATOR_CONTEXT_ID.size() == 1) {
unsetIdGenerator();
}
IdGeneratorConfigurer.GENERATOR_CONTEXT_ID.remove(context.getId());
}
}
@@ -93,12 +93,14 @@ public final class IdGeneratorConfigurer implements ApplicationListener<Applicat
}
else {
// different instance has been set, not legal
throw new BeanDefinitionStoreException("'MessageHeaders.idGenerator' has already been set and can not be set again");
throw new BeanDefinitionStoreException("'MessageHeaders.idGenerator' has already been set and " +
"can not be set again");
}
}
}
if (this.logger.isInfoEnabled()) {
this.logger.info("Message IDs will be generated using custom IdGenerator [" + idGeneratorBean.getClass() + "]");
this.logger.info("Message IDs will be generated using custom IdGenerator [" + idGeneratorBean
.getClass() + "]");
}
ReflectionUtils.setField(idGeneratorField, null, idGeneratorBean);
IdGeneratorConfigurer.theIdGenerator = idGeneratorBean;

12
spring-integration-core/src/main/java/org/springframework/integration/config/PublisherRegistrar.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2014-2019 the original author or authors.
* Copyright 2014-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.
@@ -42,7 +42,7 @@ import org.springframework.util.StringUtils;
*/
public class PublisherRegistrar implements ImportBeanDefinitionRegistrar {
private static final Log logger = LogFactory.getLog(PublisherRegistrar.class);
private static final Log LOGGER = LogFactory.getLog(PublisherRegistrar.class);
@Override
public void registerBeanDefinitions(AnnotationMetadata importingClassMetadata, BeanDefinitionRegistry registry) {
@@ -60,8 +60,8 @@ public class PublisherRegistrar implements ImportBeanDefinitionRegistrar {
if (StringUtils.hasText(defaultChannel)) {
builder.addPropertyValue("defaultChannelName", defaultChannel);
if (logger.isInfoEnabled()) {
logger.info("Setting '@Publisher' default-output-channel to '" + defaultChannel + "'.");
if (LOGGER.isInfoEnabled()) {
LOGGER.info("Setting '@Publisher' default-output-channel to '" + defaultChannel + "'.");
}
}
@@ -82,8 +82,8 @@ public class PublisherRegistrar implements ImportBeanDefinitionRegistrar {
if (StringUtils.hasText(defaultChannel)) {
if (defaultChannelPropertyValue == null) {
propertyValues.addPropertyValue("defaultChannelName", defaultChannel);
if (logger.isInfoEnabled()) {
logger.info("Setting '@Publisher' default-output-channel to '" + defaultChannel + "'.");
if (LOGGER.isInfoEnabled()) {
LOGGER.info("Setting '@Publisher' default-output-channel to '" + defaultChannel + "'.");
}
}
else if (!defaultChannel.equals(defaultChannelPropertyValue.getValue())) {

10
spring-integration-core/src/main/java/org/springframework/integration/config/ReleaseStrategyFactoryBean.java
View File

@@ -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.
@@ -40,7 +40,7 @@ import org.springframework.util.StringUtils;
*/
public class ReleaseStrategyFactoryBean implements FactoryBean<ReleaseStrategy>, InitializingBean {
private static final Log logger = LogFactory.getLog(ReleaseStrategyFactoryBean.class);
private static final Log LOGGER = LogFactory.getLog(ReleaseStrategyFactoryBean.class);
private Object target;
@@ -73,8 +73,8 @@ public class ReleaseStrategyFactoryBean implements FactoryBean<ReleaseStrategy>,
this.strategy = new MethodInvokingReleaseStrategy(this.target, method);
}
else {
if (logger.isWarnEnabled()) {
logger.warn("No ReleaseStrategy annotated method found on "
if (LOGGER.isWarnEnabled()) {
LOGGER.warn("No ReleaseStrategy annotated method found on "
+ this.target.getClass().getSimpleName()
+ "; falling back to SimpleSequenceSizeReleaseStrategy, target: "
+ this.target + ", methodName: " + this.methodName);
@@ -83,7 +83,7 @@ public class ReleaseStrategyFactoryBean implements FactoryBean<ReleaseStrategy>,
}
}
else {
logger.warn("No target supplied; falling back to SimpleSequenceSizeReleaseStrategy");
LOGGER.warn("No target supplied; falling back to SimpleSequenceSizeReleaseStrategy");
}
}

26
spring-integration-core/src/main/java/org/springframework/integration/config/xml/ApplicationEventMulticasterParser.java
View File

@@ -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.
@@ -16,18 +16,16 @@
package org.springframework.integration.config.xml;
import java.util.concurrent.CopyOnWriteArraySet;
import org.w3c.dom.Element;
import org.springframework.beans.factory.BeanDefinitionStoreException;
import org.springframework.beans.factory.support.AbstractBeanDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
import org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.context.event.SimpleApplicationEventMulticaster;
import org.springframework.context.support.AbstractApplicationContext;
import org.springframework.core.SpringVersion;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.util.StringUtils;
/**
@@ -35,17 +33,19 @@ import org.springframework.util.StringUtils;
* integration namespace.
*
* @author Mark Fisher
* @author Artem Bilan
*/
public class ApplicationEventMulticasterParser extends AbstractSingleBeanDefinitionParser {
@Override
protected String getBeanClassName(Element element) {
return "org.springframework.context.event.SimpleApplicationEventMulticaster";
protected Class<?> getBeanClass(Element element) {
return SimpleApplicationEventMulticaster.class;
}
@Override
protected String resolveId(Element element, AbstractBeanDefinition definition, ParserContext parserContext)
throws BeanDefinitionStoreException {
return AbstractApplicationContext.APPLICATION_EVENT_MULTICASTER_BEAN_NAME;
}
@@ -56,19 +56,13 @@ public class ApplicationEventMulticasterParser extends AbstractSingleBeanDefinit
builder.addPropertyReference("taskExecutor", taskExecutorRef);
}
else {
BeanDefinitionBuilder executorBuilder = BeanDefinitionBuilder.genericBeanDefinition(
"org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor");
BeanDefinitionBuilder executorBuilder =
BeanDefinitionBuilder.genericBeanDefinition(ThreadPoolTaskExecutor.class);
executorBuilder.addPropertyValue("corePoolSize", 1);
executorBuilder.addPropertyValue("maxPoolSize", 10);
executorBuilder.addPropertyValue("queueCapacity", 0);
executorBuilder.addPropertyValue("threadNamePrefix", "event-multicaster-");
String executorBeanName = BeanDefinitionReaderUtils.registerWithGeneratedName(
executorBuilder.getBeanDefinition(), parserContext.getRegistry());
builder.addPropertyReference("taskExecutor", executorBeanName);
}
String springVersion = SpringVersion.getVersion();
if (springVersion != null && springVersion.startsWith("2")) {
builder.addPropertyValue("collectionClass", CopyOnWriteArraySet.class);
builder.addPropertyValue("taskExecutor", executorBuilder.getBeanDefinition());
}
}

18
spring-integration-core/src/main/java/org/springframework/integration/config/xml/DefaultInboundChannelAdapterParser.java
View File

@@ -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.
@@ -47,7 +47,7 @@ import org.springframework.util.xml.DomUtils;
public class DefaultInboundChannelAdapterParser extends AbstractPollingInboundChannelAdapterParser {
@Override // NOSONAR complexity
protected BeanMetadataElement parseSource(Element element, ParserContext parserContext) {
protected BeanMetadataElement parseSource(Element element, ParserContext parserContext) { // NOSONAR
Object source = parserContext.extractSource(element);
BeanMetadataElement result = null;
BeanComponentDefinition innerBeanDef =
@@ -69,7 +69,6 @@ public class DefaultInboundChannelAdapterParser extends AbstractPollingInboundCh
parserContext.getReaderContext().error(
"Exactly one of the 'ref', 'expression', inner bean, <script> or <expression> is required.", element);
}
if (hasInnerDef) {
if (hasRef || hasExpression) {
parserContext.getReaderContext().error(
@@ -78,7 +77,7 @@ public class DefaultInboundChannelAdapterParser extends AbstractPollingInboundCh
return null;
}
if (hasMethod) {
result = this.parseMethodInvokingSource(innerBeanDef, methodName, element, parserContext);
result = parseMethodInvokingSource(innerBeanDef, methodName, element, parserContext);
}
else {
result = innerBeanDef.getBeanDefinition();
@@ -96,7 +95,7 @@ public class DefaultInboundChannelAdapterParser extends AbstractPollingInboundCh
BeanDefinitionBuilder sourceBuilder = BeanDefinitionBuilder.genericBeanDefinition(
IntegrationConfigUtils.BASE_PACKAGE + ".scripting.ScriptExecutingMessageSource");
sourceBuilder.addConstructorArgValue(scriptBeanDefinition);
this.parseHeaderExpressions(sourceBuilder, element, parserContext);
parseHeaderExpressions(sourceBuilder, element, parserContext);
result = sourceBuilder.getBeanDefinition();
}
else if (hasExpression || hasExpressionElement) {
@@ -111,18 +110,17 @@ public class DefaultInboundChannelAdapterParser extends AbstractPollingInboundCh
"Exactly one of the 'expression' attribute or inner <expression> is required.", element);
return null;
}
result = this.parseExpression(expressionString, expressionElement, element, parserContext);
result = parseExpression(expressionString, expressionElement, element, parserContext);
}
else if (hasRef) {
BeanMetadataElement sourceValue = new RuntimeBeanReference(sourceRef);
if (hasMethod) {
result = this.parseMethodInvokingSource(sourceValue, methodName, element, parserContext);
result = parseMethodInvokingSource(sourceValue, methodName, element, parserContext);
}
else {
result = sourceValue;
}
}
return result;
}
@@ -143,7 +141,7 @@ public class DefaultInboundChannelAdapterParser extends AbstractPollingInboundCh
BeanDefinitionBuilder sourceBuilder = BeanDefinitionBuilder
.genericBeanDefinition(ExpressionEvaluatingMessageSource.class);
BeanDefinition expressionDef = null;
BeanDefinition expressionDef;
if (StringUtils.hasText(expressionString)) {
expressionDef = new RootBeanDefinition(ExpressionFactoryBean.class);
@@ -169,7 +167,7 @@ public class DefaultInboundChannelAdapterParser extends AbstractPollingInboundCh
private void parseHeaderExpressions(BeanDefinitionBuilder builder, Element element, ParserContext parserContext) {
List<Element> headerElements = DomUtils.getChildElementsByTagName(element, "header");
if (!CollectionUtils.isEmpty(headerElements)) {
ManagedMap<String, Object> headerExpressions = new ManagedMap<String, Object>();
ManagedMap<String, Object> headerExpressions = new ManagedMap<>();
for (Element headerElement : headerElements) {
String headerName = headerElement.getAttribute("name");
BeanDefinition expressionDef = IntegrationNamespaceUtils

12
spring-integration-core/src/main/java/org/springframework/integration/config/xml/DefaultOutboundChannelAdapterParser.java
View File

@@ -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.
@@ -61,12 +61,10 @@ public class DefaultOutboundChannelAdapterParser extends AbstractOutboundChannel
"Exactly one of the 'ref', 'expression', <script> or inner bean is required.", source);
}
if (hasScript) {
if (isRef | isExpression) {
parserContext.getReaderContext().error(
"Neither 'ref' nor 'expression' are permitted when an inner script element is configured.",
source);
}
if (hasScript && (isRef | isExpression)) {
parserContext.getReaderContext().error(
"Neither 'ref' nor 'expression' are permitted when an inner script element is configured.",
source);
}
if (hasMethod & isExpression) {

9
spring-integration-core/src/main/java/org/springframework/integration/config/xml/IdempotentReceiverInterceptorParser.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2014-2019 the original author or authors.
* Copyright 2014-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.
@@ -38,12 +38,13 @@ import org.springframework.util.StringUtils;
*
* @author Artem Bilan
* @author Gary Russell
*
* @since 4.1
*/
public class IdempotentReceiverInterceptorParser extends AbstractBeanDefinitionParser {
@Override // NOSONAR complexity
protected AbstractBeanDefinition parseInternal(Element element, ParserContext parserContext) {
protected AbstractBeanDefinition parseInternal(Element element, ParserContext parserContext) { // NOSONAR
Object source = parserContext.extractSource(element);
String selector = element.getAttribute("selector");
@@ -87,14 +88,14 @@ public class IdempotentReceiverInterceptorParser extends AbstractBeanDefinitionP
parserContext.getReaderContext().error("The 'endpoint' attribute is required", source);
}
BeanMetadataElement selectorBeanDefinition = null;
BeanMetadataElement selectorBeanDefinition;
if (hasSelector) {
selectorBeanDefinition = new RuntimeBeanReference(selector);
}
else {
BeanDefinitionBuilder selectorBuilder =
BeanDefinitionBuilder.genericBeanDefinition(MetadataStoreSelector.class);
BeanMetadataElement keyStrategyBeanDefinition = null;
BeanMetadataElement keyStrategyBeanDefinition;
if (hasKeyStrategy) {
keyStrategyBeanDefinition = new RuntimeBeanReference(keyStrategy);
}

4
spring-integration-core/src/main/java/org/springframework/integration/config/xml/IntegrationNamespaceHandler.java
View File

@@ -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.
@@ -29,7 +29,7 @@ package org.springframework.integration.config.xml;
public class IntegrationNamespaceHandler extends AbstractIntegrationNamespaceHandler {
@Override
public void init() {
public void init() { // NOSONAR
registerBeanDefinitionParser("channel", new PointToPointChannelParser());
registerBeanDefinitionParser("publish-subscribe-channel", new PublishSubscribeChannelParser());
registerBeanDefinitionParser("service-activator", new ServiceActivatorParser());

58
spring-integration-core/src/main/java/org/springframework/integration/config/xml/IntegrationNamespaceUtils.java
View File

@@ -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.
@@ -66,7 +66,6 @@ import org.springframework.util.xml.DomUtils;
* @author Gary Russell
* @author Artem Bilan
* @author Gunnar Hillert
*
*/
public abstract class IntegrationNamespaceUtils {
@@ -89,7 +88,6 @@ public abstract class IntegrationNamespaceUtils {
/**
* Configures the provided bean definition builder with a property value corresponding to the attribute whose name
* is provided if that attribute is defined in the given element.
*
* @param builder the bean definition builder to be configured
* @param element the XML element where the attribute should be defined
* @param attributeName the name of the attribute whose value will be used to populate the property
@@ -104,16 +102,12 @@ public abstract class IntegrationNamespaceUtils {
/**
* Configures the provided bean definition builder with a property value corresponding to the attribute whose name
* is provided if that attribute is defined in the given element.
*
* <p>
* The property name will be the camel-case equivalent of the lower case hyphen separated attribute (e.g. the
* <p> The property name will be the camel-case equivalent of the lower case hyphen separated attribute (e.g. the
* "foo-bar" attribute would match the "fooBar" property).
*
* @see Conventions#attributeNameToPropertyName(String)
*
* @param builder the bean definition builder to be configured
* @param element - the XML element where the attribute should be defined
* @param attributeName - the name of the attribute whose value will be set on the property
* @see Conventions#attributeNameToPropertyName(String)
*/
public static void setValueIfAttributeDefined(BeanDefinitionBuilder builder, Element element,
String attributeName) {
@@ -124,7 +118,6 @@ public abstract class IntegrationNamespaceUtils {
/**
* Configures the provided bean definition builder with a property value corresponding to the attribute whose name
* is provided if that attribute is defined in the given element.
*
* @param builder the bean definition builder to be configured
* @param element the XML element where the attribute should be defined
* @param attributeName the name of the attribute whose value will be used to populate the property
@@ -144,19 +137,15 @@ public abstract class IntegrationNamespaceUtils {
/**
* Configures the provided bean definition builder with a property value corresponding to the attribute whose name
* is provided if that attribute is defined in the given element.
*
* <p>
* The property name will be the camel-case equivalent of the lower case hyphen separated attribute (e.g. the
* <p> The property name will be the camel-case equivalent of the lower case hyphen separated attribute (e.g. the
* "foo-bar" attribute would match the "fooBar" property).
*
* @see Conventions#attributeNameToPropertyName(String)
*
* @param builder the bean definition builder to be configured
* @param element - the XML element where the attribute should be defined
* @param attributeName - the name of the attribute whose value will be set on the property
* @param emptyStringAllowed - if true, the value is set, even if an empty String (""); if false, an empty
* String is treated as if the attribute wasn't provided.
*
* @see Conventions#attributeNameToPropertyName(String)
*/
public static void setValueIfAttributeDefined(BeanDefinitionBuilder builder, Element element, String attributeName,
boolean emptyStringAllowed) {
@@ -169,7 +158,6 @@ public abstract class IntegrationNamespaceUtils {
* Configures the provided bean definition builder with a property reference to a bean. The bean reference is
* identified by the value from the attribute whose name is provided if that attribute is defined in the given
* element.
*
* @param builder the bean definition builder to be configured
* @param element the XML element where the attribute should be defined
* @param attributeName the name of the attribute whose value will be used as a bean reference to populate the
@@ -200,18 +188,12 @@ public abstract class IntegrationNamespaceUtils {
* Configures the provided bean definition builder with a property reference to a bean. The bean reference is
* identified by the value from the attribute whose name is provided if that attribute is defined in the given
* element.
*
* <p>
* The property name will be the camel-case equivalent of the lower case hyphen separated attribute (e.g. the
* <p> The property name will be the camel-case equivalent of the lower case hyphen separated attribute (e.g. the
* "foo-bar" attribute would match the "fooBar" property).
*
* @see Conventions#attributeNameToPropertyName(String)
*
* @param builder the bean definition builder to be configured
* @param element - the XML element where the attribute should be defined
* @param attributeName - the name of the attribute whose value will be used as a bean reference to populate the
* property
*
* @see Conventions#attributeNameToPropertyName(String)
*/
public static void setReferenceIfAttributeDefined(BeanDefinitionBuilder builder, Element element,
@@ -230,7 +212,6 @@ public abstract class IntegrationNamespaceUtils {
/**
* Provides a user friendly description of an element based on its node name and, if available, its "id" attribute
* value. This is useful for creating error messages from within bean definition parsers.
*
* @param element The element.
* @return The description.
*/
@@ -247,7 +228,6 @@ public abstract class IntegrationNamespaceUtils {
* Parse a "poller" element to provide a reference for the target BeanDefinitionBuilder. If the poller element does
* not contain a "ref" attribute, this will create and register a PollerMetadata instance and then add it as a
* property reference of the target builder.
*
* @param pollerElement the "poller" element to parse
* @param targetBuilder the builder that expects the "trigger" property
* @param parserContext the parserContext for the target builder
@@ -257,17 +237,15 @@ public abstract class IntegrationNamespaceUtils {
if (pollerElement.hasAttribute(REF_ATTRIBUTE)) {
int numberOfAttributes = pollerElement.getAttributes().getLength();
if (numberOfAttributes != 1) {
/*
* When importing the core namespace, e.g. into jdbc, we get a 'default="false"' attribute,
* even if not explicitly declared.
*/
if (!(numberOfAttributes == 2 &&
pollerElement.hasAttribute("default") &&
pollerElement.getAttribute("default").equals("false"))) {
parserContext.getReaderContext().error(
"A 'poller' element that provides a 'ref' must have no other attributes.", pollerElement);
}
/*
* When importing the core namespace, e.g. into jdbc, we get a 'default="false"' attribute,
* even if not explicitly declared.
*/
if (numberOfAttributes != 1 && !(numberOfAttributes == 2 &&
pollerElement.hasAttribute("default") &&
pollerElement.getAttribute("default").equals("false"))) {
parserContext.getReaderContext().error(
"A 'poller' element that provides a 'ref' must have no other attributes.", pollerElement);
}
if (pollerElement.getChildNodes().getLength() != 0) {
parserContext.getReaderContext().error(
@@ -288,7 +266,6 @@ public abstract class IntegrationNamespaceUtils {
/**
* Get a text value from a named attribute if it exists, otherwise check for a nested element of the same name.
* If both are specified it is an error, but if neither is specified, just returns null.
*
* @param element a DOM node
* @param name the name of the property (attribute or child element)
* @param parserContext the current context
@@ -314,7 +291,7 @@ public abstract class IntegrationNamespaceUtils {
// parses out the inner bean definition for concrete implementation if defined
List<Element> childElements = DomUtils.getChildElementsByTagName(element, "bean");
BeanComponentDefinition innerComponentDefinition = null;
if (childElements != null && childElements.size() == 1) {
if (childElements.size() == 1) {
Element beanElement = childElements.get(0);
BeanDefinitionParserDelegate delegate = parserContext.getDelegate();
BeanDefinitionHolder bdHolder = delegate.parseBeanDefinitionElement(beanElement);
@@ -336,7 +313,6 @@ public abstract class IntegrationNamespaceUtils {
/**
* Utility method to configure a HeaderMapper for Inbound and Outbound channel adapters/gateway.
*
* @param element The element.
* @param rootBuilder The root builder.
* @param parserContext The parser context.
@@ -352,7 +328,6 @@ public abstract class IntegrationNamespaceUtils {
/**
* Utility method to configure a HeaderMapper for Inbound and Outbound channel adapters/gateway.
*
* @param element The element.
* @param rootBuilder The root builder.
* @param parserContext The parser context.
@@ -434,7 +409,6 @@ public abstract class IntegrationNamespaceUtils {
/**
* Parse attributes of "transactional" element and configure a {@link DefaultTransactionAttribute}
* with provided "transactionDefinition" properties.
*
* @param txElement The transactional element.
* @return The bean definition.
*/

7
spring-integration-core/src/main/java/org/springframework/integration/config/xml/ScatterGatherParser.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2014-2019 the original author or authors.
* Copyright 2014-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.
@@ -37,6 +37,7 @@ import org.springframework.util.xml.DomUtils;
*
* @author Artem Bilan
* @author Gary Russell
*
* @since 4.1
*/
public class ScatterGatherParser extends AbstractConsumerEndpointParser {
@@ -45,7 +46,7 @@ public class ScatterGatherParser extends AbstractConsumerEndpointParser {
private static final AggregatorParser GATHERER_PARSER = new AggregatorParser();
private static final DocumentBuilderFactory documentBuilderFactory = DocumentBuilderFactory.newInstance();
private static final DocumentBuilderFactory DOCUMENT_BUILDER_FACTORY = DocumentBuilderFactory.newInstance();
@Override
protected BeanDefinitionBuilder parseHandler(Element element, ParserContext parserContext) {
@@ -115,7 +116,7 @@ public class ScatterGatherParser extends AbstractConsumerEndpointParser {
BeanDefinition gathererDefinition = null;
if (gatherer == null) {
try {
gatherer = documentBuilderFactory.newDocumentBuilder().newDocument().createElement("aggregator");
gatherer = DOCUMENT_BUILDER_FACTORY.newDocumentBuilder().newDocument().createElement("aggregator");
}
catch (ParserConfigurationException e) {
parserContext.getReaderContext().error(e.getMessage(), element);

16
spring-integration-core/src/main/java/org/springframework/integration/config/xml/SelectorChainParser.java
View File

@@ -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.
@@ -36,12 +36,13 @@ import org.springframework.util.StringUtils;
* @author Mark Fisher
* @author Iwein Fuld
* @author Gary Russell
* @author Artem Bilan
*/
public class SelectorChainParser extends AbstractSingleBeanDefinitionParser {
@Override
protected String getBeanClassName(Element element) {
return MessageSelectorChain.class.getName();
protected Class<?> getBeanClass(Element element) {
return MessageSelectorChain.class;
}
@Override
@@ -49,7 +50,7 @@ public class SelectorChainParser extends AbstractSingleBeanDefinitionParser {
if (!StringUtils.hasText(element.getAttribute("id"))) {
parserContext.getReaderContext().error("id is required", element);
}
this.parseSelectorChain(builder, element, parserContext);
parseSelectorChain(builder, element, parserContext);
}
@SuppressWarnings({ "unchecked", "rawtypes" })
@@ -88,11 +89,12 @@ public class SelectorChainParser extends AbstractSingleBeanDefinitionParser {
}
private RuntimeBeanReference buildMethodInvokingSelector(ParserContext parserContext, String ref, String method) {
BeanDefinitionBuilder methodInvokingSelectorBuilder = BeanDefinitionBuilder.genericBeanDefinition(MethodInvokingSelector.class);
BeanDefinitionBuilder methodInvokingSelectorBuilder =
BeanDefinitionBuilder.genericBeanDefinition(MethodInvokingSelector.class);
methodInvokingSelectorBuilder.addConstructorArgValue(new RuntimeBeanReference(ref));
methodInvokingSelectorBuilder.addConstructorArgValue(method);
RuntimeBeanReference selector = new RuntimeBeanReference(BeanDefinitionReaderUtils.registerWithGeneratedName(
return new RuntimeBeanReference(BeanDefinitionReaderUtils.registerWithGeneratedName(
methodInvokingSelectorBuilder.getBeanDefinition(), parserContext.getRegistry()));
return selector;
}
}

2
spring-integration-core/src/main/java/org/springframework/integration/history/MessageHistoryConfigurer.java
View File

@@ -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.

36
spring-integration-core/src/test/java/org/springframework/integration/config/xml/DelegatingConsumerParserTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2013-2019 the original author or authors.
* Copyright 2013-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.
@@ -17,7 +17,7 @@
package org.springframework.integration.config.xml;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.fail;
import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
import static org.mockito.Mockito.mock;
import java.util.ArrayList;
@@ -25,8 +25,7 @@ import java.util.Collection;
import java.util.List;
import java.util.Set;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.annotation.Autowired;
@@ -46,8 +45,7 @@ import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
/**
* @author Gary Russell
@@ -56,8 +54,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
* @since 3.0
*
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
@SpringJUnitConfig
public class DelegatingConsumerParserTests {
@Autowired
@@ -191,8 +188,7 @@ public class DelegatingConsumerParserTests {
}
@Test
@SuppressWarnings("unchecked")
public void testOneRefOnly() throws Exception {
public void testOneRefOnly() {
ServiceActivatorFactoryBean fb = new ServiceActivatorFactoryBean();
fb.setBeanFactory(mock(BeanFactory.class));
MyServiceARPMH service = new MyServiceARPMH();
@@ -200,24 +196,18 @@ public class DelegatingConsumerParserTests {
fb.setTargetObject(service);
fb.getObject();
assertThat(TestUtils.getPropertyValue(fb, "referencedReplyProducers", Set.class).contains(service)).isTrue();
assertThat(TestUtils.getPropertyValue(fb, "REFERENCED_REPLY_PRODUCERS", Set.class).contains(service)).isTrue();
ServiceActivatorFactoryBean fb2 = new ServiceActivatorFactoryBean();
fb2.setBeanFactory(mock(BeanFactory.class));
fb2.setTargetObject(service);
try {
fb2.getObject();
fail("expected exception");
}
catch (Exception e) {
assertThat(e.getMessage())
.isEqualTo("An AbstractMessageProducingMessageHandler may only be referenced once (foo) - "
+ "use scope=\"prototype\"");
}
assertThatExceptionOfType(Exception.class)
.isThrownBy(fb2::getObject)
.withMessage("An AbstractMessageProducingMessageHandler may only be referenced once (foo) - "
+ "use scope=\"prototype\"");
fb.destroy();
assertThat(TestUtils.getPropertyValue(fb, "referencedReplyProducers", Set.class).contains(service)).isFalse();
assertThat(TestUtils.getPropertyValue(fb, "REFERENCED_REPLY_PRODUCERS", Set.class).contains(service)).isFalse();
}
private void testHandler(MessageHandler handler) {
@@ -256,7 +246,7 @@ public class DelegatingConsumerParserTests {
@Override
protected Collection<MessageChannel> determineTargetChannels(Message<?> message) {
List<MessageChannel> channels = new ArrayList<MessageChannel>();
List<MessageChannel> channels = new ArrayList<>();
channels.add(replyChannel);
return channels;
}

2
spring-integration-core/src/test/java/org/springframework/integration/dsl/manualflow/ManualFlowTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2019 the original author or authors.
* Copyright 2016-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.