diff --git a/build.gradle b/build.gradle index 693f395447..3bb4b23fea 100644 --- a/build.gradle +++ b/build.gradle @@ -610,6 +610,7 @@ project('spring-integration-jpa') { } testImplementation "com.h2database:h2:$h2Version" testImplementation "org.hibernate:hibernate-entitymanager:$hibernateVersion" + testImplementation "org.hamcrest:hamcrest-core:$hamcrestVersion" } } 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 41f1982214..738f206d7c 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-2020 the original author or authors. + * Copyright 2014-2021 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. @@ -89,7 +89,7 @@ public final class FixedSubscriberChannel implements SubscribableChannel, BeanNa @Override public boolean subscribe(MessageHandler handler) { - if (handler != this.handler && LOGGER.isDebugEnabled()) { + if (!this.handler.equals(handler) && LOGGER.isDebugEnabled()) { LOGGER.debug(getComponentName() + ": cannot be subscribed to (it has a fixed single subscriber)."); } return false; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/PublisherRegistrar.java b/spring-integration-core/src/main/java/org/springframework/integration/config/PublisherRegistrar.java index 3a57485ee4..8590741f8b 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/PublisherRegistrar.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/PublisherRegistrar.java @@ -103,7 +103,9 @@ public class PublisherRegistrar implements ImportBeanDefinitionRegistrar { if (beanFactory != null) { order = beanFactory.resolveEmbeddedValue(order); } - postProcessor.setOrder(Integer.parseInt(order)); + if (StringUtils.hasText(order)) { + postProcessor.setOrder(Integer.parseInt(order)); + } } return postProcessor; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/PointToPointChannelParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/PointToPointChannelParser.java index 465b9a6c04..752465a4e8 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/PointToPointChannelParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/PointToPointChannelParser.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2019 the original author or authors. + * Copyright 2002-2021 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. @@ -46,9 +46,9 @@ public class PointToPointChannelParser extends AbstractChannelParser { @Override protected BeanDefinitionBuilder buildBeanDefinition(Element element, ParserContext parserContext) { BeanDefinitionBuilder builder = null; - Element queueElement = null; + Element queueElement; String fixedSubscriberChannel = element.getAttribute("fixed-subscriber"); - boolean isFixedSubscriber = "true".equals(fixedSubscriberChannel.trim().toLowerCase()); + boolean isFixedSubscriber = "true".equalsIgnoreCase(fixedSubscriberChannel.trim()); // configure a queue-based channel if any queue sub-element is defined String channel = element.getAttribute(ID_ATTRIBUTE); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/mapping/AbstractHeaderMapper.java b/spring-integration-core/src/main/java/org/springframework/integration/mapping/AbstractHeaderMapper.java index 1d109c527b..11e8511fae 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/mapping/AbstractHeaderMapper.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/mapping/AbstractHeaderMapper.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2020 the original author or authors. + * Copyright 2002-2021 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. @@ -277,7 +277,7 @@ public abstract class AbstractHeaderMapper implements RequestReplyHeaderMappe if (shouldMapHeader(headerName, headerMatcher)) { Object value = entry.getValue(); target.put(headerName, value); - if (this.replyHeaderMatcher == headerMatcher && + if (this.replyHeaderMatcher.equals(headerMatcher) && JsonHeaders.TYPE_ID.equals(headerName) && value != null) { ResolvableType resolvableType = diff --git a/spring-integration-core/src/main/java/org/springframework/integration/router/PayloadTypeRouter.java b/spring-integration-core/src/main/java/org/springframework/integration/router/PayloadTypeRouter.java index 841f167e1b..16002281bf 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/router/PayloadTypeRouter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/router/PayloadTypeRouter.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2019 the original author or authors. + * Copyright 2002-2021 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. @@ -107,7 +107,7 @@ public class PayloadTypeRouter extends AbstractMappingMessageRouter { } // no match at this level, continue up the hierarchy for (Class superInterface : iface.getInterfaces()) { - int weight = this.determineTypeDifferenceWeight(candidate, superInterface, level + 3); + int weight = this.determineTypeDifferenceWeight(candidate, superInterface, level + 3); // NOSONAR if (weight < Integer.MAX_VALUE) { return weight; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/router/RecipientListRouter.java b/spring-integration-core/src/main/java/org/springframework/integration/router/RecipientListRouter.java index 2ecb7efc48..f15458ea3d 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/router/RecipientListRouter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/router/RecipientListRouter.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2020 the original author or authors. + * Copyright 2002-2021 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.router; +import java.util.ArrayList; import java.util.Collection; import java.util.Collections; import java.util.Iterator; @@ -98,13 +99,8 @@ public class RecipientListRouter extends AbstractMessageRouter implements Recipi public void setRecipients(List recipients) { Assert.notEmpty(recipients, "'recipients' must not be empty"); Queue newRecipients = new ConcurrentLinkedQueue<>(recipients); - newRecipients.forEach(this::setupRecipient); - - if (logger.isDebugEnabled()) { - logger.debug("Channel Recipients: " + this.recipients + " replaced with: " + newRecipients); - } - + logger.debug(() -> "Channel Recipients: " + this.recipients + " replaced with: " + newRecipients); this.recipients = newRecipients; } @@ -126,10 +122,7 @@ public class RecipientListRouter extends AbstractMessageRouter implements Recipi addRecipient(next.getKey(), (MessageSelector) null, newRecipients); } } - if (logger.isDebugEnabled()) { - logger.debug("Channel Recipients: " + this.recipients + " replaced with: " + newRecipients); - } - + logger.debug(() -> "Channel Recipients: " + this.recipients + " replaced with: " + newRecipients); this.recipients = newRecipients; } @@ -193,7 +186,7 @@ public class RecipientListRouter extends AbstractMessageRouter implements Recipi int counter = 0; MessageChannel channel = getChannelResolver().resolveDestination(channelName); for (Iterator it = this.recipients.iterator(); it.hasNext(); ) { - if (it.next().getChannel() == channel) { + if (channel.equals(it.next().getChannel())) { it.remove(); counter++; } @@ -236,9 +229,7 @@ public class RecipientListRouter extends AbstractMessageRouter implements Recipi addRecipient(key); } } - if (logger.isDebugEnabled()) { - logger.debug("Channel Recipients:" + originalRecipients + " replaced with:" + this.recipients); - } + logger.debug(() -> "Channel Recipients: " + originalRecipients + " replaced with: " + this.recipients); } @Override @@ -259,10 +250,13 @@ public class RecipientListRouter extends AbstractMessageRouter implements Recipi @Override protected Collection determineTargetChannels(Message message) { - return this.recipients.stream() - .filter(recipient -> recipient.accept(message)) - .map(Recipient::getChannel) - .collect(Collectors.toList()); + List result = new ArrayList<>(); + for (Recipient recipient : this.recipients) { + if (recipient.accept(message)) { + result.add(recipient.getChannel()); + } + } + return result; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/scheduling/PollerMetadata.java b/spring-integration-core/src/main/java/org/springframework/integration/scheduling/PollerMetadata.java index e9e7f3c033..b4fd2da445 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/scheduling/PollerMetadata.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/scheduling/PollerMetadata.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2019 the original author or authors. + * Copyright 2002-2021 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,11 +31,23 @@ import org.springframework.util.ErrorHandler; * @author Mark Fisher * @author Oleg Zhurakousky * @author Gary Russell + * @author Artem Bilan */ public class PollerMetadata { + /** + * The constant for unlimited number of message to poll in one cycle. + */ public static final int MAX_MESSAGES_UNBOUNDED = Integer.MIN_VALUE; + /** + * The default receive timeout as one second. + */ + public static final long DEFAULT_RECEIVE_TIMEOUT = 1000; + + /** + * The bean name for global default poller. + */ public static final String DEFAULT_POLLER_METADATA_BEAN_NAME = "org.springframework.integration.context.defaultPollerMetadata"; @@ -44,21 +56,21 @@ public class PollerMetadata { */ public static final String DEFAULT_POLLER = DEFAULT_POLLER_METADATA_BEAN_NAME; - private volatile Trigger trigger; + private Trigger trigger; - private volatile long maxMessagesPerPoll = MAX_MESSAGES_UNBOUNDED; + private long maxMessagesPerPoll = MAX_MESSAGES_UNBOUNDED; - private volatile long receiveTimeout = 1000; + private long receiveTimeout = DEFAULT_RECEIVE_TIMEOUT; - private volatile ErrorHandler errorHandler; + private ErrorHandler errorHandler; - private volatile List adviceChain; + private List adviceChain; - private volatile Executor taskExecutor; + private Executor taskExecutor; - private volatile long sendTimeout; + private long sendTimeout; - private volatile TransactionSynchronizationFactory transactionSynchronizationFactory; + private TransactionSynchronizationFactory transactionSynchronizationFactory; public void setTransactionSynchronizationFactory( @@ -91,11 +103,8 @@ public class PollerMetadata { * Set the maximum number of messages to receive for each poll. * A non-positive value indicates that polling should repeat as long * as non-null messages are being received and successfully sent. - * *

The default is unbounded. - * * @param maxMessagesPerPoll The maxMessagesPerPoll to set. - * * @see #MAX_MESSAGES_UNBOUNDED */ public void setMaxMessagesPerPoll(long maxMessagesPerPoll) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/json/Jackson2JsonObjectMapper.java b/spring-integration-core/src/main/java/org/springframework/integration/support/json/Jackson2JsonObjectMapper.java index 9742175b56..19319f653d 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/json/Jackson2JsonObjectMapper.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/json/Jackson2JsonObjectMapper.java @@ -198,44 +198,46 @@ public class Jackson2JsonObjectMapper extends AbstractJacksonJsonObjectMapper - implements ManageableLifecycle { +public class FileReadingMessageSource extends AbstractMessageSource implements ManageableLifecycle { private static final int DEFAULT_INTERNAL_QUEUE_CAPACITY = 5; @@ -124,14 +123,14 @@ public class FileReadingMessageSource extends AbstractMessageSource private WatchEventType[] watchEvents = { WatchEventType.CREATE }; /** - * Creates a FileReadingMessageSource with a naturally ordered queue of unbounded capacity. + * Create a FileReadingMessageSource with a naturally ordered queue of unbounded capacity. */ public FileReadingMessageSource() { this(null); } /** - * Creates a FileReadingMessageSource with a bounded queue of the given + * Create a FileReadingMessageSource with a bounded queue of the given * capacity. This can be used to reduce the memory footprint of this * component when reading from a large directory. * @param internalQueueCapacity @@ -150,15 +149,13 @@ public class FileReadingMessageSource extends AbstractMessageSource } /** - * Creates a FileReadingMessageSource with a {@link PriorityBlockingQueue} + * Create a FileReadingMessageSource with a {@link PriorityBlockingQueue} * ordered with the passed in {@link Comparator}. *

The size of the queue used should be large enough to hold all the files * in the input directory in order to sort all of them, so restricting the * size of the queue is mutually exclusive with ordering. No guarantees * about file delivery order can be made under concurrent access. - * @param receptionOrderComparator - * the comparator to be used to order the files in the internal - * queue + * @param receptionOrderComparator the comparator to be used to order the files in the internal queue */ public FileReadingMessageSource(@Nullable Comparator receptionOrderComparator) { this.toBeReceived = new PriorityBlockingQueue<>(DEFAULT_INTERNAL_QUEUE_CAPACITY, receptionOrderComparator); @@ -211,7 +208,7 @@ public class FileReadingMessageSource extends AbstractMessageSource } /** - * Sets a {@link FileListFilter}. + * Set a {@link FileListFilter}. * By default a {@link org.springframework.integration.file.filters.AcceptOnceFileListFilter} * with no bounds is used. In most cases a customized {@link FileListFilter} will * be needed to deal with modification and duplication concerns. @@ -227,10 +224,8 @@ public class FileReadingMessageSource extends AbstractMessageSource } /** - * Optional. Sets a {@link FileLocker} to be used to guard files against - * duplicate processing. - *

- * The supplied FileLocker must be thread safe + * Set a {@link FileLocker} to be used to guard files against duplicate processing. + *

The supplied FileLocker must be thread safe * @param locker a locker */ public void setLocker(FileLocker locker) { @@ -239,7 +234,7 @@ public class FileReadingMessageSource extends AbstractMessageSource } /** - * Optional. Set this flag if you want to make sure the internal queue is + * Set this flag if you want to make sure the internal queue is * refreshed with the latest content of the input directory on each poll. *

* By default this implementation will empty its queue before looking at the @@ -247,8 +242,7 @@ public class FileReadingMessageSource extends AbstractMessageSource * consider the effects of setting this flag. The internal * {@link java.util.concurrent.BlockingQueue} that this class is keeping * will more likely be out of sync with the file system if this flag is set - * to false, but it will change more often (causing expensive - * reordering) if it is set to true. + * to false, but it will change more often (causing expensive reordering) if it is set to true. * @param scanEachPoll * whether or not the component should re-scan (as opposed to not * rescanning until the entire backlog has been delivered) @@ -489,13 +483,13 @@ public class FileReadingMessageSource extends AbstractMessageSource while (key != null) { File parentDir = ((Path) key.watchable()).toAbsolutePath().toFile(); for (WatchEvent event : key.pollEvents()) { - if (event.kind() == StandardWatchEventKinds.ENTRY_CREATE || - event.kind() == StandardWatchEventKinds.ENTRY_MODIFY || - event.kind() == StandardWatchEventKinds.ENTRY_DELETE) { + if (StandardWatchEventKinds.ENTRY_CREATE.equals(event.kind()) || + StandardWatchEventKinds.ENTRY_MODIFY.equals(event.kind()) || + StandardWatchEventKinds.ENTRY_DELETE.equals(event.kind())) { processFilesFromNormalEvent(files, parentDir, event); } - else if (event.kind() == StandardWatchEventKinds.OVERFLOW) { + else if (StandardWatchEventKinds.OVERFLOW.equals(event.kind())) { processFilesFromOverflowEvent(files, event); } } @@ -510,7 +504,7 @@ public class FileReadingMessageSource extends AbstractMessageSource File file = new File(parentDir, item.toFile().getName()); logger.debug(() -> "Watch event [" + event.kind() + "] for file [" + file + "]"); - if (event.kind() == StandardWatchEventKinds.ENTRY_DELETE) { + if (StandardWatchEventKinds.ENTRY_DELETE.equals(event.kind())) { if (getFilter() instanceof ResettableFileListFilter) { ((ResettableFileListFilter) getFilter()).remove(file); } diff --git a/spring-integration-groovy/src/main/java/org/springframework/integration/groovy/config/GroovyAwareScriptExecutingProcessorFactory.java b/spring-integration-groovy/src/main/java/org/springframework/integration/groovy/config/GroovyAwareScriptExecutingProcessorFactory.java index 8f547b5b03..bb3e934aae 100644 --- a/spring-integration-groovy/src/main/java/org/springframework/integration/groovy/config/GroovyAwareScriptExecutingProcessorFactory.java +++ b/spring-integration-groovy/src/main/java/org/springframework/integration/groovy/config/GroovyAwareScriptExecutingProcessorFactory.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2019 the original author or authors. + * Copyright 2016-2021 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. @@ -27,6 +27,7 @@ import org.springframework.scripting.ScriptSource; * if provided {@code language == "groovy"}, otherwise delegates to the super class. * * @author Artem Bilan + * * @since 5.0 */ public class GroovyAwareScriptExecutingProcessorFactory extends ScriptExecutingProcessorFactory { @@ -34,7 +35,8 @@ public class GroovyAwareScriptExecutingProcessorFactory extends ScriptExecutingP @Override public AbstractScriptExecutingMessageProcessor createMessageProcessor(String language, ScriptSource scriptSource, ScriptVariableGenerator scriptVariableGenerator) { - if ("groovy".equals(language.toLowerCase())) { + + if ("groovy".equalsIgnoreCase(language)) { return new GroovyScriptExecutingMessageProcessor(scriptSource, scriptVariableGenerator); } else { diff --git a/spring-integration-http/src/main/java/org/springframework/integration/http/support/DefaultHttpHeaderMapper.java b/spring-integration-http/src/main/java/org/springframework/integration/http/support/DefaultHttpHeaderMapper.java index af2b51568d..33a63bb961 100644 --- a/spring-integration-http/src/main/java/org/springframework/integration/http/support/DefaultHttpHeaderMapper.java +++ b/spring-integration-http/src/main/java/org/springframework/integration/http/support/DefaultHttpHeaderMapper.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2020 the original author or authors. + * Copyright 2002-2021 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. @@ -249,10 +249,10 @@ public class DefaultHttpHeaderMapper implements HeaderMapper, BeanF * @param outboundHeaderNames The outbound header names. */ public void setOutboundHeaderNames(String... outboundHeaderNames) { - if (HTTP_REQUEST_HEADER_NAMES == outboundHeaderNames) { + if (Arrays.equals(HTTP_REQUEST_HEADER_NAMES, outboundHeaderNames)) { this.isDefaultOutboundMapper = true; } - else if (HTTP_RESPONSE_HEADER_NAMES == outboundHeaderNames) { + else if (Arrays.equals(HTTP_RESPONSE_HEADER_NAMES, outboundHeaderNames)) { this.isDefaultInboundMapper = true; } this.outboundHeaderNames = diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractClientConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractClientConnectionFactory.java index 2e75fc5b32..d79288ae8d 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractClientConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractClientConnectionFactory.java @@ -20,6 +20,7 @@ import java.io.IOException; import java.io.UncheckedIOException; import java.net.Socket; import java.time.Duration; +import java.util.Objects; import java.util.concurrent.locks.ReadWriteLock; import java.util.concurrent.locks.ReentrantReadWriteLock; import java.util.function.Predicate; @@ -53,7 +54,7 @@ public abstract class AbstractClientConnectionFactory extends AbstractConnection private volatile TcpConnectionSupport theConnection; /** - * Constructs a factory that will established connections to the host and port. + * Construct a factory that will established connections to the host and port. * @param host The host. * @param port The port. */ @@ -108,7 +109,7 @@ public abstract class AbstractClientConnectionFactory extends AbstractConnection } /** - * Obtains a connection - if {@link #setSingleUse(boolean)} was called with + * Obtain a connection - if {@link #setSingleUse(boolean)} was called with * true, a new connection is returned; otherwise a single connection is * reused for all requests while the connection remains open. * @throws InterruptedException if interrupted. @@ -198,7 +199,7 @@ public abstract class AbstractClientConnectionFactory extends AbstractConnection } /** - * Transfers attributes such as (de)serializers, singleUse etc to a new connection. + * Transfer attributes such as (de)serializers, singleUse etc to a new connection. * When the connection factory has a reference to a TCPListener (to read * responses), or for single use connections, the connection is executed. * Single use connections need to read from the connection in order to @@ -243,11 +244,10 @@ public abstract class AbstractClientConnectionFactory extends AbstractConnection /** * Force close the connection and null the field if it's * a shared connection. - * * @param connection The connection. */ public void forceClose(TcpConnection connection) { - if (this.theConnection == connection) { + if (Objects.equals(this.theConnection, connection)) { this.theConnection = null; } connection.close(); diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/ClientModeConnectionManager.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/ClientModeConnectionManager.java index 56cd208af9..663f3223e2 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/ClientModeConnectionManager.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/ClientModeConnectionManager.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2019 the original author or authors. + * Copyright 2002-2021 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,8 @@ package org.springframework.integration.ip.tcp.connection; +import java.util.Objects; + import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -28,6 +30,8 @@ import org.springframework.util.Assert; * connection factory will create a new one (if possible). * * @author Gary Russell + * @author Artem Bilan + * * @since 2.1 * */ @@ -53,7 +57,7 @@ public class ClientModeConnectionManager implements Runnable { synchronized (this.clientConnectionFactory) { try { TcpConnection connection = this.clientConnectionFactory.getConnection(); - if (connection != this.lastConnection) { + if (!Objects.equals(connection, this.lastConnection)) { if (this.logger.isDebugEnabled()) { this.logger.debug("Connection " + connection.getConnectionId() + " established"); } @@ -65,8 +69,8 @@ public class ClientModeConnectionManager implements Runnable { } } } - catch (Exception e) { - this.logger.error("Could not establish connection using " + this.clientConnectionFactory, e); + catch (Exception ex) { + this.logger.error("Could not establish connection using " + this.clientConnectionFactory, ex); } } } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionSupport.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionSupport.java index 19854c5a2a..3c403422e4 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionSupport.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionSupport.java @@ -120,7 +120,7 @@ public abstract class TcpConnectionSupport implements TcpConnection { } /** - * Creates a {@link TcpConnectionSupport} object and publishes a + * Create a {@link TcpConnectionSupport} object and publishes a * {@link TcpConnectionOpenEvent}, if an event publisher is provided. * @param socket the underlying socket. * @param server true if this connection is a server connection @@ -164,7 +164,7 @@ public abstract class TcpConnectionSupport implements TcpConnection { } /** - * Closes this connection. + * Close this connection. */ @Override public void close() { @@ -182,7 +182,6 @@ public abstract class TcpConnectionSupport implements TcpConnection { /** * If we have been intercepted, propagate the close from the outermost interceptor; * otherwise, just call close(). - * * @param isException true when this call is the result of an Exception. */ protected void closeConnection(boolean isException) { @@ -288,8 +287,7 @@ public abstract class TcpConnectionSupport implements TcpConnection { /** * Set a temporary listener to receive just the first incoming message. - * Used in conjunction with a connectionTest in a client connection - * factory. + * Used in conjunction with a connectionTest in a client connection factory. * @param tListener the test listener. * @since 5.3 */ @@ -309,7 +307,7 @@ public abstract class TcpConnectionSupport implements TcpConnection { } /** - * Registers a sender. Used on server side connections so a + * Register a sender. Used on server side connections so a * sender can determine which connection to send a reply * to. * @param senderToRegister the sender. @@ -322,7 +320,7 @@ public abstract class TcpConnectionSupport implements TcpConnection { } /** - * Registers the senders. Used on server side connections so a + * Register the senders. Used on server side connections so a * sender can determine which connection to send a reply * to. * @param sendersToRegister the sender. @@ -451,8 +449,7 @@ public abstract class TcpConnectionSupport implements TcpConnection { protected final void sendExceptionToListener(Exception e) { TcpListener listenerForException = getListener(); if (!this.exceptionSent.getAndSet(true) && listenerForException != null) { - Map headers = Collections.singletonMap(IpHeaders.CONNECTION_ID, - (Object) this.getConnectionId()); + Map headers = Collections.singletonMap(IpHeaders.CONNECTION_ID, getConnectionId()); ErrorMessage errorMessage = new ErrorMessage(e, headers); listenerForException.onMessage(errorMessage); } @@ -491,7 +488,7 @@ public abstract class TcpConnectionSupport implements TcpConnection { * @param event the event to publish. */ public void publishEvent(TcpConnectionEvent event) { - Assert.isTrue(event.getSource() == this, "Can only publish events with this as the source"); + Assert.isTrue(equals(event.getSource()), "Can only publish events with this as the source"); this.doPublish(event); } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java index 473f6c33b6..74802a099e 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java @@ -28,6 +28,7 @@ import java.nio.channels.ClosedChannelException; import java.nio.channels.SelectionKey; import java.nio.channels.Selector; import java.nio.channels.SocketChannel; +import java.util.Arrays; import java.util.concurrent.BlockingQueue; import java.util.concurrent.CountDownLatch; import java.util.concurrent.Executor; @@ -65,6 +66,8 @@ public class TcpNioConnection extends TcpConnectionSupport { private static final int SIXTY = 60; + private static final int MAX_MESSAGE_SIZE = 60 * 1024; + private static final long DEFAULT_PIPE_TIMEOUT = 60000; private static final byte[] EOF = new byte[0]; // EOF marker buffer @@ -87,8 +90,6 @@ public class TcpNioConnection extends TcpConnectionSupport { private volatile ByteBuffer rawBuffer; - private volatile int maxMessageSize = 60 * 1024; - private volatile long lastRead; private volatile long lastSend; @@ -100,7 +101,7 @@ public class TcpNioConnection extends TcpConnectionSupport { private volatile boolean timedOut; /** - * Constructs a TcpNetConnection for the SocketChannel. + * Construct a TcpNetConnection for the SocketChannel. * @param socketChannel The socketChannel. * @param server If true, this connection was created as * a result of an incoming request. @@ -211,7 +212,7 @@ public class TcpNioConnection extends TcpConnectionSupport { } /** - * Allocates a ByteBuffer of the requested length using normal or + * Allocate a ByteBuffer of the requested length using normal or * direct buffers, depending on the usingDirectBuffers field. * * @param length The buffer length. @@ -229,8 +230,8 @@ public class TcpNioConnection extends TcpConnectionSupport { } /** - * If there is no listener, - * this method exits. When there is a listener, this method assembles + * If there is no listener, this method exits. + * When there is a listener, this method assembles * data into messages by invoking convertAndSend whenever there is * data in the input Stream. Method exits when a message is complete * and there is no more data; thus freeing the thread to work on other @@ -409,7 +410,7 @@ public class TcpNioConnection extends TcpConnectionSupport { private void doRead() throws IOException { if (this.rawBuffer == null) { - this.rawBuffer = allocate(this.maxMessageSize); + this.rawBuffer = allocate(MAX_MESSAGE_SIZE); } this.writingLatch = new CountDownLatch(1); @@ -722,7 +723,7 @@ public class TcpNioConnection extends TcpConnectionSupport { } } int bite; - bite = this.currentBuffer[this.currentOffset++] & 0xff; + bite = this.currentBuffer[this.currentOffset++] & 0xff; // N0SONAR this.available.decrementAndGet(); if (this.currentOffset >= this.currentBuffer.length) { this.currentBuffer = null; @@ -736,7 +737,7 @@ public class TcpNioConnection extends TcpConnectionSupport { while (buffer == null) { try { buffer = this.buffers.poll(1, TimeUnit.SECONDS); - if (buffer == EOF || (buffer == null && this.isClosed)) { + if (Arrays.equals(EOF, buffer) || (buffer == null && this.isClosed)) { return null; } } diff --git a/spring-integration-jpa/src/test/java/org/springframework/integration/jpa/inbound/JpaPollingChannelAdapterTests.java b/spring-integration-jpa/src/test/java/org/springframework/integration/jpa/inbound/JpaPollingChannelAdapterTests.java index 94b3c07681..1fbdcd5e03 100644 --- a/spring-integration-jpa/src/test/java/org/springframework/integration/jpa/inbound/JpaPollingChannelAdapterTests.java +++ b/spring-integration-jpa/src/test/java/org/springframework/integration/jpa/inbound/JpaPollingChannelAdapterTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2019 the original author or authors. + * Copyright 2002-2021 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.jpa.inbound; import static org.assertj.core.api.Assertions.assertThat; -import static org.assertj.core.api.Assertions.fail; +import static org.awaitility.Awaitility.await; import java.util.ArrayList; import java.util.Collection; @@ -25,8 +25,7 @@ import java.util.List; import javax.persistence.EntityManager; -import org.junit.Test; -import org.junit.runner.RunWith; +import org.junit.jupiter.api.Test; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; @@ -44,8 +43,7 @@ import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.annotation.Rollback; -import org.springframework.test.context.ContextConfiguration; -import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; import org.springframework.transaction.annotation.Transactional; /** @@ -58,11 +56,10 @@ import org.springframework.transaction.annotation.Transactional; * @since 2.2 * */ -@RunWith(SpringJUnit4ClassRunner.class) -@ContextConfiguration +@SpringJUnitConfig @Rollback @Transactional("transactionManager") -@DirtiesContext +@DirtiesContext(classMode = DirtiesContext.ClassMode.AFTER_EACH_TEST_METHOD) public class JpaPollingChannelAdapterTests { @Autowired @@ -90,7 +87,6 @@ public class JpaPollingChannelAdapterTests { * to retrieve a list of records from the database. */ @Test - @DirtiesContext public void testWithEntityClass() throws Exception { testTrigger.reset(); //~~~~SETUP~~~~~ @@ -259,7 +255,6 @@ public class JpaPollingChannelAdapterTests { * will be deleted after the polling. */ @Test - @DirtiesContext public void testWithJpaQueryAndDelete() throws Exception { testTrigger.reset(); @@ -297,30 +292,11 @@ public class JpaPollingChannelAdapterTests { assertThat(students.size()).isEqualTo(3); - Long studentCount = waitForDeletes(students); - - assertThat(studentCount).isEqualTo(Long.valueOf(0)); - } - - private Long waitForDeletes(Collection students) throws InterruptedException { - Long studentCount = (long) students.size(); - - int n = 0; - - while (studentCount > 0) { - studentCount = entityManager.createQuery("select count(*) from Student", Long.class).getSingleResult(); - if (studentCount > 0) { - Thread.sleep(100); - if (n++ > 100) { - fail("Failed to delete after poll"); - } - } - } - return studentCount; + await().until(() -> entityManager.createQuery("select count(*) from Student", Long.class).getSingleResult(), + (count) -> count == 0); } @Test - @DirtiesContext public void testWithJpaQueryButNoResultsAndDelete() throws Exception { testTrigger.reset(); @@ -355,7 +331,6 @@ public class JpaPollingChannelAdapterTests { * will be deleted after the polling. */ @Test - @DirtiesContext public void testWithJpaQueryAndDeletePerRow() throws Exception { testTrigger.reset(); @@ -389,10 +364,8 @@ public class JpaPollingChannelAdapterTests { assertThat(students.size()).isEqualTo(3); - Long studentCount = waitForDeletes(students); - - assertThat(studentCount).isEqualTo(Long.valueOf(0)); - + await().until(() -> entityManager.createQuery("select count(*) from Student", Long.class).getSingleResult(), + (count) -> count == 0); } /** diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageSource.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageSource.java index 4e84f335eb..3a1db60e08 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageSource.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageSource.java @@ -1,5 +1,5 @@ /* - * Copyright 2018-2020 the original author or authors. + * Copyright 2018-2021 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. @@ -440,7 +440,7 @@ public class KafkaMessageSource extends AbstractMessageSource impl this.consumerProperties.getClientId(), null, this.consumerProperties.getKafkaConsumerProperties()); ConsumerRebalanceListener rebalanceCallback = - new ItegrationConsumerRebalanceListener(this.consumerProperties.getConsumerRebalanceListener()); + new IntegrationConsumerRebalanceListener(this.consumerProperties.getConsumerRebalanceListener()); Pattern topicPattern = this.consumerProperties.getTopicPattern(); TopicPartitionOffset[] partitions = this.consumerProperties.getTopicPartitions(); @@ -580,13 +580,13 @@ public class KafkaMessageSource extends AbstractMessageSource impl } } - private class ItegrationConsumerRebalanceListener implements ConsumerRebalanceListener { + private class IntegrationConsumerRebalanceListener implements ConsumerRebalanceListener { private final ConsumerRebalanceListener providedRebalanceListener; private final boolean isConsumerAware; - ItegrationConsumerRebalanceListener(ConsumerRebalanceListener providedRebalanceListener) { + IntegrationConsumerRebalanceListener(ConsumerRebalanceListener providedRebalanceListener) { this.providedRebalanceListener = providedRebalanceListener; this.isConsumerAware = providedRebalanceListener instanceof ConsumerAwareRebalanceListener; } @@ -594,9 +594,7 @@ public class KafkaMessageSource extends AbstractMessageSource impl @Override public void onPartitionsRevoked(Collection partitions) { KafkaMessageSource.this.assignedPartitions.removeAll(partitions); - if (KafkaMessageSource.this.logger.isInfoEnabled()) { - KafkaMessageSource.this.logger.info("Partitions revoked: " + partitions); - } + KafkaMessageSource.this.logger.info(() -> "Partitions revoked: " + partitions); if (this.providedRebalanceListener != null) { if (this.isConsumerAware) { ((ConsumerAwareRebalanceListener) this.providedRebalanceListener) @@ -625,9 +623,7 @@ public class KafkaMessageSource extends AbstractMessageSource impl KafkaMessageSource.this.logger.warn("Paused consumer resumed by Kafka due to rebalance; " + "consumer paused again, so the initial poll() will never return any records"); } - if (KafkaMessageSource.this.logger.isInfoEnabled()) { - KafkaMessageSource.this.logger.info("Partitions assigned: " + partitions); - } + KafkaMessageSource.this.logger.info(() -> "Partitions assigned: " + partitions); if (this.providedRebalanceListener != null) { if (this.isConsumerAware) { ((ConsumerAwareRebalanceListener) this.providedRebalanceListener) @@ -783,7 +779,7 @@ public class KafkaMessageSource extends AbstractMessageSource impl // see if there are any pending acks for higher offsets List> toCommit = new ArrayList<>(); for (KafkaAckInfo info : candidates) { - if (info != this.ackInfo) { + if (!this.ackInfo.equals(info)) { if (info.isAckDeferred()) { toCommit.add(info); } @@ -828,9 +824,7 @@ public class KafkaMessageSource extends AbstractMessageSource impl } } else { - if (this.logger.isDebugEnabled()) { - this.logger.debug("Deferring commit offset; earlier messages are in flight."); - } + this.logger.debug("Deferring commit offset; earlier messages are in flight."); } } } diff --git a/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/inbound/WebFluxInboundEndpoint.java b/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/inbound/WebFluxInboundEndpoint.java index 40015fd6b8..03f61aaaff 100644 --- a/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/inbound/WebFluxInboundEndpoint.java +++ b/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/inbound/WebFluxInboundEndpoint.java @@ -1,5 +1,5 @@ /* - * Copyright 2017-2019 the original author or authors. + * Copyright 2017-2021 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. @@ -35,7 +35,6 @@ import org.springframework.core.ReactiveAdapterRegistry; import org.springframework.core.ResolvableType; import org.springframework.expression.EvaluationContext; import org.springframework.expression.spel.support.StandardEvaluationContext; -import org.springframework.http.HttpHeaders; import org.springframework.http.HttpMethod; import org.springframework.http.HttpStatus; import org.springframework.http.MediaType; @@ -47,6 +46,7 @@ import org.springframework.http.codec.ServerCodecConfigurer; import org.springframework.http.server.reactive.ServerHttpRequest; import org.springframework.http.server.reactive.ServerHttpResponse; import org.springframework.integration.expression.ExpressionEvalMap; +import org.springframework.integration.http.HttpHeaders; import org.springframework.integration.http.inbound.BaseHttpInboundEndpoint; import org.springframework.integration.support.AbstractIntegrationMessageBuilder; import org.springframework.messaging.Message; @@ -281,9 +281,7 @@ public class WebFluxInboundEndpoint extends BaseHttpInboundEndpoint implements W prepareRequestMessageBuilder(request, payload, headers); return exchange.getPrincipal() - .map(principal -> - messageBuilder - .setHeader(org.springframework.integration.http.HttpHeaders.USER_PRINCIPAL, principal)) + .map(principal -> messageBuilder.setHeader(HttpHeaders.USER_PRINCIPAL, principal)) .defaultIfEmpty(messageBuilder) .map(AbstractIntegrationMessageBuilder::build) .zipWith(Mono.just(httpEntity)); @@ -308,19 +306,17 @@ public class WebFluxInboundEndpoint extends BaseHttpInboundEndpoint implements W .copyHeaders(headers); } - messageBuilder.setHeader(org.springframework.integration.http.HttpHeaders.REQUEST_URL, - request.getURI().toString()); + messageBuilder.setHeader(HttpHeaders.REQUEST_URL, request.getURI().toString()); HttpMethod httpMethod = request.getMethod(); if (httpMethod != null) { - messageBuilder.setHeader(org.springframework.integration.http.HttpHeaders.REQUEST_METHOD, - httpMethod.toString()); + messageBuilder.setHeader(HttpHeaders.REQUEST_METHOD, httpMethod.toString()); } return messageBuilder; } private EvaluationContext buildEvaluationContext(RequestEntity httpEntity, ServerWebExchange exchange) { ServerHttpRequest request = exchange.getRequest(); - HttpHeaders requestHeaders = request.getHeaders(); + org.springframework.http.HttpHeaders requestHeaders = request.getHeaders(); MultiValueMap requestParams = request.getQueryParams(); Map exchangeAttributes = exchange.getAttributes(); @@ -375,8 +371,8 @@ public class WebFluxInboundEndpoint extends BaseHttpInboundEndpoint implements W exchange.getResponse().setStatusCode(e.getStatusCode()); } - HttpHeaders entityHeaders = e.getHeaders(); - HttpHeaders responseHeaders = exchange.getResponse().getHeaders(); + org.springframework.http.HttpHeaders entityHeaders = e.getHeaders(); + org.springframework.http.HttpHeaders responseHeaders = exchange.getResponse().getHeaders(); if (!entityHeaders.isEmpty()) { entityHeaders.entrySet().stream() @@ -501,7 +497,7 @@ public class WebFluxInboundEndpoint extends BaseHttpInboundEndpoint implements W if (adapter.isNoValue()) { return ResolvableType.forClass(Void.class); } - else if (genericType != ResolvableType.NONE) { + else if (!ResolvableType.NONE.equals(genericType)) { return genericType; } else { diff --git a/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/IntegrationWebSocketContainer.java b/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/IntegrationWebSocketContainer.java index 3983e707b9..42865a95f1 100644 --- a/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/IntegrationWebSocketContainer.java +++ b/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/IntegrationWebSocketContainer.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2019 the original author or authors. + * Copyright 2014-2021 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. @@ -88,7 +88,7 @@ public abstract class IntegrationWebSocketContainer implements DisposableBean { } public void setMessageListener(WebSocketListener messageListener) { - Assert.state(this.messageListener == null || this.messageListener == messageListener, + Assert.state(this.messageListener == null || this.messageListener.equals(messageListener), "'messageListener' is already configured"); this.messageListener = messageListener; } @@ -119,7 +119,7 @@ public abstract class IntegrationWebSocketContainer implements DisposableBean { public WebSocketSession getSession(String sessionId) { WebSocketSession session = this.sessions.get(sessionId); - Assert.notNull(session, "Session not found for id '" + sessionId + "'"); + Assert.notNull(session, () -> "Session not found for id '" + sessionId + "'"); return session; } @@ -139,8 +139,8 @@ public abstract class IntegrationWebSocketContainer implements DisposableBean { try { session.close(CloseStatus.GOING_AWAY); } - catch (Exception e) { - this.logger.error("Failed to close session id '" + session.getId() + "': " + e.getMessage()); + catch (Exception ex) { + this.logger.error("Failed to close session id '" + session.getId() + "': " + ex.getMessage()); } } }