diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/AbstractAmqpChannel.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/AbstractAmqpChannel.java index afcde094bb..22a6b4a5a9 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/AbstractAmqpChannel.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/AbstractAmqpChannel.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. @@ -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(); } diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpChannelParser.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpChannelParser.java index 26fa7fb87e..58343f2132 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpChannelParser.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpChannelParser.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. @@ -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; } } diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundChannelAdapter.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundChannelAdapter.java index e258360c3f..118c55efba 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundChannelAdapter.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundChannelAdapter.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. @@ -61,17 +61,17 @@ import com.rabbitmq.client.Channel; public class AmqpInboundChannelAdapter extends MessageProducerSupport implements OrderlyShutdownCapable { - private static final ThreadLocal attributesHolder = new ThreadLocal(); + private static final ThreadLocal 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 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 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 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) 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 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(); } } diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AsyncAmqpOutboundGateway.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AsyncAmqpOutboundGateway.java index d649afd1db..d32c109a05 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AsyncAmqpOutboundGateway.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AsyncAmqpOutboundGateway.java @@ -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()) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageBarrier.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageBarrier.java index 6d8df08c04..a418cc88b5 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageBarrier.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageBarrier.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. @@ -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). *

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

@@ -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 { - private static final Log log = LogFactory.getLog(CorrelatingMessageBarrier.class); - - private volatile CorrelationStrategy correlationStrategy; - - private volatile ReleaseStrategy releaseStrategy; - - private final ConcurrentMap correlationLocks = new ConcurrentHashMap(); + private final ConcurrentMap 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> 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> 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) nextMessage; } + else { + remove(key); + } + return (Message) nextMessage; } } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/ExpressionEvaluatingCorrelationStrategy.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/ExpressionEvaluatingCorrelationStrategy.java index 0089160330..62c4066f88 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/ExpressionEvaluatingCorrelationStrategy.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/ExpressionEvaluatingCorrelationStrategy.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. @@ -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 processor; public ExpressionEvaluatingCorrelationStrategy(String expressionString) { Assert.hasText(expressionString, "expressionString must not be empty"); - Expression expression = expressionParser.parseExpression(expressionString); - this.processor = new ExpressionEvaluatingMessageProcessor(expression, Object.class); + Expression expression = EXPRESSION_PARSER.parseExpression(expressionString); + this.processor = new ExpressionEvaluatingMessageProcessor<>(expression, Object.class); } public ExpressionEvaluatingCorrelationStrategy(Expression expression) { - this.processor = new ExpressionEvaluatingMessageProcessor(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); } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/SequenceSizeReleaseStrategy.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/SequenceSizeReleaseStrategy.java index d0347bdc86..e6f4e4e7a7 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/SequenceSizeReleaseStrategy.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/SequenceSizeReleaseStrategy.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. @@ -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> 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> 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) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aop/PublisherAnnotationAdvisor.java b/spring-integration-core/src/main/java/org/springframework/integration/aop/PublisherAnnotationAdvisor.java index 67f49fe89f..408aba3f33 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aop/PublisherAnnotationAdvisor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aop/PublisherAnnotationAdvisor.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. @@ -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 annotationType; - - - /** - * Create a new AnnotationClassFilter for the given annotation type. - * @param annotationType the annotation type to look for - */ - MetaAnnotationMethodMatcher(Class 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)); - } - - } - } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractExecutorChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractExecutorChannel.java index 3517effe53..41f9d84c76 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractExecutorChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractExecutorChannel.java @@ -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; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractSubscribableChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractSubscribableChannel.java index 8f4fb51e5f..2e8ea64935 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractSubscribableChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractSubscribableChannel.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. @@ -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); } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/FixedSubscriberChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/FixedSubscriberChannel.java index af43d9dfcb..41f1982214 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/FixedSubscriberChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/FixedSubscriberChannel.java @@ -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; } } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/PriorityChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/PriorityChannel.java index 7a8e63c25b..044efa8af8 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/PriorityChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/PriorityChannel.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. @@ -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 null, the priority will be based upon the value of - * {@link IntegrationMessageHeaderAccessor#getPriority()}. - * + * {@link StaticMessageHeaderAccessor#getPriority(Message)}. * @param comparator The comparator. */ public PriorityChannel(Comparator> 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 null, 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> 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> targetComparator; - SequenceFallbackComparator(Comparator> targetComparator) { + SequenceFallbackComparator(@Nullable Comparator> 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; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/PublishSubscribeChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/PublishSubscribeChannel.java index a60c8ec5a8..2e3241d66c 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/PublishSubscribeChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/PublishSubscribeChannel.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. @@ -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) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/interceptor/WireTap.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/interceptor/WireTap.java index 09aab7eedb..75c9e7a6cc 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/interceptor/WireTap.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/interceptor/WireTap.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. @@ -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 no {@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; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/KryoClassListRegistrar.java b/spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/KryoClassListRegistrar.java index 34aa81c917..a976824fd8 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/KryoClassListRegistrar.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/KryoClassListRegistrar.java @@ -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> registeredClasses; - private int initialValue = 50; + private int initialValue = DEFAULT_INITIAL_ID; /** * @param classes the list of classes to validateRegistration */ public KryoClassListRegistrar(List> classes) { - this.registeredClasses = new ArrayList>(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 getRegistrations() { - List registrations = new ArrayList(); + List 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), diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/AbstractStandardMessageHandlerFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/config/AbstractStandardMessageHandlerFactoryBean.java index 13052316bd..e746b69377 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/AbstractStandardMessageHandlerFactoryBean.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/AbstractStandardMessageHandlerFactoryBean.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. @@ -43,9 +43,9 @@ import org.springframework.util.StringUtils; public abstract class AbstractStandardMessageHandlerFactoryBean extends AbstractSimpleMessageHandlerFactoryBean implements DisposableBean { - private static final ExpressionParser expressionParser = new SpelExpressionParser(); + private static final ExpressionParser EXPRESSION_PARSER = new SpelExpressionParser(); - private static final Set referencedReplyProducers = new HashSet<>(); + private static final Set 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; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java index 5546fad9fe..55b78f0eba 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.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. @@ -74,7 +74,7 @@ public class ConsumerEndpointFactoryBean implements FactoryBean, 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"); } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/DefaultConfiguringBeanFactoryPostProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/config/DefaultConfiguringBeanFactoryPostProcessor.java index a1cc2c3f08..8147979193 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/DefaultConfiguringBeanFactoryPostProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/DefaultConfiguringBeanFactoryPostProcessor.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-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 registriesProcessed = new HashSet<>(); + private static final Set 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."); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/ExpressionControlBusFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/config/ExpressionControlBusFactoryBean.java index 160fb0e1f1..4bd88cc77d 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/ExpressionControlBusFactoryBean.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/ExpressionControlBusFactoryBean.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. @@ -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 { - 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; } - } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/GlobalChannelInterceptorProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/config/GlobalChannelInterceptorProcessor.java index f4f72c374b..f9cf8f748f 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/GlobalChannelInterceptorProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/GlobalChannelInterceptorProcessor.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. @@ -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 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 tempInterceptors = new ArrayList<>(); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/IdGeneratorConfigurer.java b/spring-integration-core/src/main/java/org/springframework/integration/config/IdGeneratorConfigurer.java index 5275cd64db..15faf354db 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/IdGeneratorConfigurer.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/IdGeneratorConfigurer.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. @@ -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 { - private static final Set generatorContextId = new HashSet(); + private static final Set GENERATOR_CONTEXT_ID = new HashSet<>(); private static volatile IdGenerator theIdGenerator; @@ -53,19 +55,17 @@ public final class IdGeneratorConfigurer implements ApplicationListener 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, 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, 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, } } else { - logger.warn("No target supplied; falling back to SimpleSequenceSizeReleaseStrategy"); + LOGGER.warn("No target supplied; falling back to SimpleSequenceSizeReleaseStrategy"); } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/ApplicationEventMulticasterParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/ApplicationEventMulticasterParser.java index ba13df2f5e..374c2f89d7 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/ApplicationEventMulticasterParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/ApplicationEventMulticasterParser.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. @@ -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()); } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/DefaultInboundChannelAdapterParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/DefaultInboundChannelAdapterParser.java index 860ca71b48..5aca12c723 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/DefaultInboundChannelAdapterParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/DefaultInboundChannelAdapterParser.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. @@ -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,