diff --git a/spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationProperties.java b/spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationProperties.java index 4cfd27c1c4..f4e63aa7fc 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationProperties.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationProperties.java @@ -16,9 +16,11 @@ package org.springframework.integration.context; +import java.util.Arrays; import java.util.Properties; import org.springframework.integration.util.JavaUtils; +import org.springframework.util.Assert; import org.springframework.util.StringUtils; /** @@ -117,7 +119,7 @@ public final class IntegrationProperties { private boolean errorChannelIgnoreFailures = true; - private int taskSchedulerPoolSize = 10; + private int taskSchedulerPoolSize = 10; // NOSONAR private boolean messagingTemplateThrowExceptionOnLateReply = false; @@ -246,7 +248,8 @@ public final class IntegrationProperties { * @param readOnlyHeaders the value for {@link #READ_ONLY_HEADERS} option. */ public void setReadOnlyHeaders(String... readOnlyHeaders) { - this.readOnlyHeaders = readOnlyHeaders; + Assert.notNull(readOnlyHeaders, "'readOnlyHeaders' must not be null."); + this.readOnlyHeaders = Arrays.copyOf(readOnlyHeaders, readOnlyHeaders.length); } /** @@ -254,7 +257,7 @@ public final class IntegrationProperties { * @return the value of {@link #READ_ONLY_HEADERS} option. */ public String[] getReadOnlyHeaders() { - return this.readOnlyHeaders; + return Arrays.copyOf(this.readOnlyHeaders, this.readOnlyHeaders.length); } /** @@ -262,7 +265,8 @@ public final class IntegrationProperties { * @param noAutoStartupEndpoints the value for {@link #ENDPOINTS_NO_AUTO_STARTUP} option. */ public void setNoAutoStartupEndpoints(String... noAutoStartupEndpoints) { - this.noAutoStartupEndpoints = noAutoStartupEndpoints; + Assert.notNull(noAutoStartupEndpoints, "'noAutoStartupEndpoints' must not be null."); + this.noAutoStartupEndpoints = Arrays.copyOf(noAutoStartupEndpoints, noAutoStartupEndpoints.length);; } /** @@ -270,7 +274,7 @@ public final class IntegrationProperties { * @return the value of {@link #ENDPOINTS_NO_AUTO_STARTUP} option. */ public String[] getNoAutoStartupEndpoints() { - return this.noAutoStartupEndpoints; + return Arrays.copyOf(this.noAutoStartupEndpoints, this.noAutoStartupEndpoints.length); } /** diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetServerConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetServerConnectionFactory.java index d9e6c9124c..3ce5d16a95 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetServerConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetServerConnectionFactory.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. @@ -112,55 +112,7 @@ public class TcpNetServerConnectionFactory extends AbstractServerConnectionFacto try { setupServerSocket(); while (true) { - final Socket socket; - /* - * User hooks in the TcpSocketSupport may have set the server socket SO_TIMEOUT. - * Not fatal. - */ - try { - if (this.serverSocket == null) { - logger.debug(() -> this + " stopped before accept"); - throw new IOException(this + " stopped before accept"); - } - else { - socket = this.serverSocket.accept(); - } - } - catch (@SuppressWarnings("unused") SocketTimeoutException ste) { - logger.debug("Timed out on accept; continuing"); - continue; - } - if (isShuttingDown()) { - logger.info(() -> "New connection from " + socket.getInetAddress().getHostAddress() - + ":" + socket.getPort() - + " rejected; the server is in the process of shutting down."); - socket.close(); - } - else { - logger.debug(() -> "Accepted connection from " + socket.getInetAddress().getHostAddress() - + ":" + socket.getPort()); - try { - setSocketAttributes(socket); - TcpConnectionSupport connection = this.tcpNetConnectionSupport.createNewConnection(socket, true, - isLookupHost(), getApplicationEventPublisher(), getComponentName()); - connection = wrapConnection(connection); - initializeConnection(connection, socket); - getTaskExecutor().execute(connection); - harvestClosedConnections(); - connection.publishConnectionOpenEvent(); - } - catch (RuntimeException ex) { - this.logger.error(ex, () -> - "Failed to create and configure a TcpConnection for the new socket: " - + socket.getInetAddress().getHostAddress() + ":" + socket.getPort()); - try { - socket.close(); - } - catch (@SuppressWarnings("unused") IOException e1) { // NOSONAR - exception as flow control - // empty - } - } - } + acceptConnectionAndExecute(); } } catch (IOException ex) { // NOSONAR flow control via exceptions @@ -196,6 +148,58 @@ public class TcpNetServerConnectionFactory extends AbstractServerConnectionFacto publishServerListeningEvent(getPort()); } + private void acceptConnectionAndExecute() throws IOException { + final Socket socket; + /* + * User hooks in the TcpSocketSupport may have set the server socket SO_TIMEOUT. + * Not fatal. + */ + try { + if (this.serverSocket == null) { + logger.debug(() -> this + " stopped before accept"); + throw new IOException(this + " stopped before accept"); + } + else { + socket = this.serverSocket.accept(); + } + } + catch (@SuppressWarnings("unused") SocketTimeoutException ste) { + logger.debug("Timed out on accept; continuing"); + return; + } + if (isShuttingDown()) { + logger.info(() -> "New connection from " + socket.getInetAddress().getHostAddress() + + ":" + socket.getPort() + + " rejected; the server is in the process of shutting down."); + socket.close(); + } + else { + logger.debug(() -> "Accepted connection from " + socket.getInetAddress().getHostAddress() + + ":" + socket.getPort()); + try { + setSocketAttributes(socket); + TcpConnectionSupport connection = this.tcpNetConnectionSupport.createNewConnection(socket, true, + isLookupHost(), getApplicationEventPublisher(), getComponentName()); + connection = wrapConnection(connection); + initializeConnection(connection, socket); + getTaskExecutor().execute(connection); + harvestClosedConnections(); + connection.publishConnectionOpenEvent(); + } + catch (RuntimeException ex) { + this.logger.error(ex, () -> + "Failed to create and configure a TcpConnection for the new socket: " + + socket.getInetAddress().getHostAddress() + ":" + socket.getPort()); + try { + socket.close(); + } + catch (@SuppressWarnings("unused") IOException e1) { // NOSONAR - exception as flow control + // empty + } + } + } + } + /** * Create a new {@link ServerSocket}. This default implementation uses the default * {@link ServerSocketFactory}. Override to use some other mechanism diff --git a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/inbound/AbstractMongoDbMessageSource.java b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/inbound/AbstractMongoDbMessageSource.java index 797fd69653..418c3c6760 100644 --- a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/inbound/AbstractMongoDbMessageSource.java +++ b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/inbound/AbstractMongoDbMessageSource.java @@ -26,8 +26,6 @@ import org.springframework.context.ApplicationContext; import org.springframework.context.ApplicationContextAware; import org.springframework.data.mapping.IdentifierAccessor; import org.springframework.data.mapping.context.MappingContext; -import org.springframework.data.mongodb.ReactiveMongoDatabaseFactory; -import org.springframework.data.mongodb.core.ReactiveMongoTemplate; import org.springframework.data.mongodb.core.convert.MongoConverter; import org.springframework.data.mongodb.core.mapping.MongoPersistentEntity; import org.springframework.data.mongodb.core.query.BasicQuery; @@ -35,14 +33,12 @@ import org.springframework.data.mongodb.core.query.BasicUpdate; import org.springframework.data.mongodb.core.query.Criteria; import org.springframework.data.mongodb.core.query.Query; import org.springframework.data.mongodb.core.query.Update; -import org.springframework.data.mongodb.core.query.UpdateDefinition; import org.springframework.data.util.Pair; import org.springframework.expression.Expression; import org.springframework.expression.TypeLocator; import org.springframework.expression.common.LiteralExpression; import org.springframework.expression.spel.support.StandardTypeLocator; import org.springframework.integration.endpoint.AbstractMessageSource; -import org.springframework.integration.mongodb.support.MongoHeaders; import org.springframework.lang.Nullable; import org.springframework.util.Assert; import org.springframework.util.LinkedMultiValueMap; @@ -64,7 +60,7 @@ public abstract class AbstractMongoDbMessageSource extends AbstractMessageSou private static final String ID_FIELD = "_id"; - protected final Expression queryExpression; + protected final Expression queryExpression; // NOSONAR - final private Expression collectionNameExpression = new LiteralExpression("data"); @@ -86,9 +82,7 @@ public abstract class AbstractMongoDbMessageSource extends AbstractMessageSou } /** - * Allow you to set the type of the entityClass that will be passed to the - * {@link ReactiveMongoTemplate#find(Query, Class)} or {@link ReactiveMongoTemplate#findOne(Query, Class)} - * method. + * Set the type of the entityClass that will be passed to the find MongoDb template operation. * Default is {@link DBObject}. * @param entityClass The entity class. */ @@ -98,10 +92,10 @@ public abstract class AbstractMongoDbMessageSource extends AbstractMessageSou } /** - * Allow you to manage which find* method to invoke on {@link ReactiveMongoTemplate}. + * Manage which find* method to invoke. * Default is 'false', which means the {@link #receive()} method will use - * the {@link ReactiveMongoTemplate#find(Query, Class)} method. If set to 'true', - * {@link #receive()} will use {@link ReactiveMongoTemplate#findOne(Query, Class)}, + * the {@code find()} method. If set to 'true', + * {@link #receive()} will use {@code findOne(Query, Class)}, * and the payload of the returned {@link org.springframework.messaging.Message} * will be the returned target Object of type * identified by {@link #entityClass} instead of a List. @@ -114,7 +108,7 @@ public abstract class AbstractMongoDbMessageSource extends AbstractMessageSou /** * Set the SpEL {@link Expression} that should resolve to a collection name * used by the {@link Query}. The resulting collection name will be included - * in the {@link MongoHeaders#COLLECTION_NAME} header. + * in the {@link org.springframework.integration.mongodb.support.MongoHeaders#COLLECTION_NAME} header. * @param collectionNameExpression The collection name expression. */ public void setCollectionNameExpression(Expression collectionNameExpression) { @@ -123,9 +117,8 @@ public abstract class AbstractMongoDbMessageSource extends AbstractMessageSou } /** - * Allow you to provide a custom {@link MongoConverter} used to assist in deserialization - * data read from MongoDb. Only allowed if this instance was constructed with a - * {@link ReactiveMongoDatabaseFactory}. + * Provide a custom {@link MongoConverter} used to assist in deserialization + * data read from MongoDb. * @param mongoConverter The mongo converter. */ public void setMongoConverter(MongoConverter mongoConverter) { @@ -134,7 +127,8 @@ public abstract class AbstractMongoDbMessageSource extends AbstractMessageSou /** * Specify an optional {@code update} for just polled records from the collection. - * @param updateExpression SpEL expression for an {@link UpdateDefinition}. + * @param updateExpression SpEL expression for an + * {@link org.springframework.data.mongodb.core.query.UpdateDefinition}. * @since 5.5 */ public void setUpdateExpression(Expression updateExpression) {