From 97702ae7128bc3280b54f29fa47da6a360368e4d Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Wed, 8 Apr 2020 13:01:49 -0400 Subject: [PATCH] Fix some Sonar smells --- .../ApplicationEventMulticasterParser.java | 6 ++--- .../dispatcher/BroadcastingDispatcher.java | 14 ++++++----- .../RoundRobinLoadBalancingStrategy.java | 23 ++++--------------- .../endpoint/MessageProducerSupport.java | 9 ++++---- .../integration/endpoint/PollingConsumer.java | 9 ++++++-- .../endpoint/SourcePollingChannelAdapter.java | 2 +- .../MongoDbChangeStreamMessageProducer.java | 10 ++++---- 7 files changed, 31 insertions(+), 42 deletions(-) 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 374c2f89d7..966be4b46d 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 @@ -58,9 +58,9 @@ public class ApplicationEventMulticasterParser extends AbstractSingleBeanDefinit else { BeanDefinitionBuilder executorBuilder = BeanDefinitionBuilder.genericBeanDefinition(ThreadPoolTaskExecutor.class); - executorBuilder.addPropertyValue("corePoolSize", 1); - executorBuilder.addPropertyValue("maxPoolSize", 10); - executorBuilder.addPropertyValue("queueCapacity", 0); + executorBuilder.addPropertyValue("corePoolSize", 1); // NOSONAR + executorBuilder.addPropertyValue("maxPoolSize", 10); // NOSONAR + executorBuilder.addPropertyValue("queueCapacity", 0); // NOSONAR executorBuilder.addPropertyValue("threadNamePrefix", "event-multicaster-"); builder.addPropertyValue("taskExecutor", executorBuilder.getBeanDefinition()); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/BroadcastingDispatcher.java b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/BroadcastingDispatcher.java index 36be406d6f..95b3439f8e 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/BroadcastingDispatcher.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/BroadcastingDispatcher.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. @@ -65,14 +65,14 @@ public class BroadcastingDispatcher extends AbstractDispatcher implements BeanFa private volatile int minSubscribers; + private MessageHandlingTaskDecorator messageHandlingTaskDecorator = task -> task; + + private BeanFactory beanFactory; + private volatile MessageBuilderFactory messageBuilderFactory = new DefaultMessageBuilderFactory(); private volatile boolean messageBuilderFactorySet; - private volatile MessageHandlingTaskDecorator messageHandlingTaskDecorator = task -> task; - - private BeanFactory beanFactory; - public BroadcastingDispatcher() { this(null, false); @@ -226,7 +226,9 @@ public class BroadcastingDispatcher extends AbstractDispatcher implements BeanFa } catch (RuntimeException e) { if (!this.ignoreFailures) { - if (e instanceof MessagingException && ((MessagingException) e).getFailedMessage() == null) { + if (e instanceof MessagingException + && ((MessagingException) e).getFailedMessage() == null) { // NOSONAR + throw new MessagingException(message, "Failed to handle Message", e); } throw e; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/RoundRobinLoadBalancingStrategy.java b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/RoundRobinLoadBalancingStrategy.java index d2eb072363..45179c0bb8 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/RoundRobinLoadBalancingStrategy.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/RoundRobinLoadBalancingStrategy.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,6 +16,7 @@ package org.springframework.integration.dispatcher; +import java.util.Arrays; import java.util.Collection; import java.util.Iterator; import java.util.concurrent.atomic.AtomicInteger; @@ -52,7 +53,7 @@ public class RoundRobinLoadBalancingStrategy implements LoadBalancingStrategy { return handlers.iterator(); } - return this.buildHandlerIterator(size, handlers.toArray(new MessageHandler[size])); + return buildHandlerIterator(size, handlers.toArray(new MessageHandler[size])); } private Iterator buildHandlerIterator(int size, final MessageHandler[] handlers) { @@ -63,23 +64,7 @@ public class RoundRobinLoadBalancingStrategy implements LoadBalancingStrategy { System.arraycopy(handlers, nextHandlerStartIndex, reorderedHandlers, 0, size - nextHandlerStartIndex); System.arraycopy(handlers, 0, reorderedHandlers, size - nextHandlerStartIndex, nextHandlerStartIndex); - return new Iterator() { - - private int currentIndex = 0; - - public boolean hasNext() { - return this.currentIndex < reorderedHandlers.length; - } - - public MessageHandler next() { - return reorderedHandlers[this.currentIndex++]; - } - - public void remove() { - throw new UnsupportedOperationException("Remove is not supported by this Iterator"); - } - - }; + return Arrays.stream(reorderedHandlers).iterator(); } /** diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java index 160089ff2f..90d4c155d7 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java @@ -205,8 +205,7 @@ public abstract class MessageProducerSupport extends AbstractEndpoint implements } message = trackMessageIfAny(message); try { - MessageChannel outputChannel = getRequiredOutputChannel(); - this.messagingTemplate.send(outputChannel, message); + this.messagingTemplate.send(getRequiredOutputChannel(), message); } catch (RuntimeException ex) { if (!sendErrorMessageIfNecessary(message, ex)) { @@ -216,7 +215,7 @@ public abstract class MessageProducerSupport extends AbstractEndpoint implements } protected void subscribeToPublisher(Publisher> publisher) { - MessageChannel outputChannel = getRequiredOutputChannel(); + MessageChannel channelForSubscription = getRequiredOutputChannel(); Flux> messageFlux = Flux.from(publisher) @@ -225,8 +224,8 @@ public abstract class MessageProducerSupport extends AbstractEndpoint implements .doOnCancel(this::stop) .takeWhile((message) -> isRunning()); - if (outputChannel instanceof ReactiveStreamsSubscribableChannel) { - ((ReactiveStreamsSubscribableChannel) outputChannel).subscribeTo(messageFlux); + if (channelForSubscription instanceof ReactiveStreamsSubscribableChannel) { + ((ReactiveStreamsSubscribableChannel) channelForSubscription).subscribeTo(messageFlux); } else { messageFlux diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollingConsumer.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollingConsumer.java index 62f123bc33..8744edb630 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollingConsumer.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollingConsumer.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. @@ -52,13 +52,18 @@ import org.springframework.util.CollectionUtils; */ public class PollingConsumer extends AbstractPollingEndpoint implements IntegrationConsumer { + /** + * A default receive timeout as {@value DEFAULT_RECEIVE_TIMEOUT} milliseconds. + */ + public static final long DEFAULT_RECEIVE_TIMEOUT = 1000; + private final PollableChannel inputChannel; private final MessageHandler handler; private final List channelInterceptors; - private volatile long receiveTimeout = 1000; + private volatile long receiveTimeout = DEFAULT_RECEIVE_TIMEOUT; public PollingConsumer(PollableChannel inputChannel, MessageHandler handler) { Assert.notNull(inputChannel, "inputChannel must not be null"); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java index 0047000413..caebab68c9 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java @@ -236,7 +236,7 @@ public class SourcePollingChannelAdapter extends AbstractPollingEndpoint } catch (Exception e) { AckUtils.autoNack(ackCallback); - if (e instanceof MessagingException) { + if (e instanceof MessagingException) { // NOSONAR throw (MessagingException) e; } else { diff --git a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/inbound/MongoDbChangeStreamMessageProducer.java b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/inbound/MongoDbChangeStreamMessageProducer.java index 7b1eedd00b..e5e249e2b7 100644 --- a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/inbound/MongoDbChangeStreamMessageProducer.java +++ b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/inbound/MongoDbChangeStreamMessageProducer.java @@ -17,9 +17,7 @@ package org.springframework.integration.mongodb.inbound; import org.bson.Document; -import org.reactivestreams.Publisher; -import org.springframework.data.mongodb.core.ChangeStreamEvent; import org.springframework.data.mongodb.core.ChangeStreamOptions; import org.springframework.data.mongodb.core.ReactiveMongoOperations; import org.springframework.integration.endpoint.MessageProducerSupport; @@ -35,7 +33,7 @@ import reactor.core.publisher.Flux; * A {@link MessageProducerSupport} for MongoDB Change Stream implementation. * The functionality is based on the * {@link ReactiveMongoOperations#changeStream(String, ChangeStreamOptions, Class)} - * and {@link MessageProducerSupport#subscribeToPublisher(Publisher)} consumption. + * and {@link MessageProducerSupport#subscribeToPublisher(org.reactivestreams.Publisher)} consumption. * * @author Artem Bilan * @@ -97,9 +95,9 @@ public class MongoDbChangeStreamMessageProducer extends MessageProducerSupport { /** * Configure this channel adapter to build a {@link Message} to produce - * with a payload based on a {@link ChangeStreamEvent#getBody()} (by default) - * or use a whole {@link ChangeStreamEvent} as a payload. - * @param extractBody to extract {@link ChangeStreamEvent#getBody()} or not. + * with a payload based on a {@link org.springframework.data.mongodb.core.ChangeStreamEvent#getBody()} (by default) + * or use a whole {@link org.springframework.data.mongodb.core.ChangeStreamEvent} as a payload. + * @param extractBody to extract {@link org.springframework.data.mongodb.core.ChangeStreamEvent#getBody()} or not. */ public void setExtractBody(boolean extractBody) { this.extractBody = extractBody;