diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java index 500c63552b..2f8c2884c4 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java @@ -726,7 +726,7 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport doAccept(selector, server, now); } catch (Exception e) { - logger.error("Exception accepting new connection", e); + logger.error("Exception accepting new connection(s)", e); } } else { diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioServerConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioServerConnectionFactory.java index 7b72d19f9c..e2d9e9410c 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioServerConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioServerConnectionFactory.java @@ -46,6 +46,8 @@ import org.springframework.util.Assert; */ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFactory { + private boolean multiAccept = true; + private volatile ServerSocketChannel serverChannel; private volatile boolean usingDirectBuffers; @@ -64,6 +66,18 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto super(port); } + /** + * Set to false to only accept one connection per iteration over the + * selector keys. This might be necessary to avoid accepts overwhelming + * reads of existing sockets. By default when the {@code OP_ACCEPT} operation + * is ready, we will keep accepting connections in a loop until no more arrive. + * @param multiAccept false to accept connections one-at-a-time. + * @since 5.1.4 + */ + public void setMultiAccept(boolean multiAccept) { + this.multiAccept = multiAccept; + } + @Override public String getComponentType() { return "tcp-nio-server-connection-factory"; @@ -204,40 +218,45 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto protected void doAccept(final Selector selectorForNewSocket, ServerSocketChannel server, long now) { logger.debug("New accept"); try { - SocketChannel channel = server.accept(); - if (isShuttingDown()) { - if (logger.isInfoEnabled()) { - logger.info("New connection from " + channel.socket().getInetAddress().getHostAddress() - + ":" + channel.socket().getPort() - + " rejected; the server is in the process of shutting down."); - } - channel.close(); - } - else { - try { - channel.configureBlocking(false); - Socket socket = channel.socket(); - setSocketAttributes(socket); - TcpNioConnection connection = createTcpNioConnection(channel); - if (connection == null) { - return; + SocketChannel channel = null; + do { + channel = server.accept(); + if (channel != null) { + if (isShuttingDown()) { + if (logger.isInfoEnabled()) { + logger.info("New connection from " + channel.socket().getInetAddress().getHostAddress() + + ":" + channel.socket().getPort() + + " rejected; the server is in the process of shutting down."); + } + channel.close(); } - connection.setTaskExecutor(getTaskExecutor()); - connection.setLastRead(now); - if (getSslHandshakeTimeout() != null && connection instanceof TcpNioSSLConnection) { - ((TcpNioSSLConnection) connection).setHandshakeTimeout(getSslHandshakeTimeout()); + else { + try { + channel.configureBlocking(false); + Socket socket = channel.socket(); + setSocketAttributes(socket); + TcpNioConnection connection = createTcpNioConnection(channel); + if (connection == null) { + return; + } + connection.setTaskExecutor(getTaskExecutor()); + connection.setLastRead(now); + if (getSslHandshakeTimeout() != null && connection instanceof TcpNioSSLConnection) { + ((TcpNioSSLConnection) connection).setHandshakeTimeout(getSslHandshakeTimeout()); + } + this.channelMap.put(channel, connection); + channel.register(selectorForNewSocket, SelectionKey.OP_READ, connection); + connection.publishConnectionOpenEvent(); + } + catch (IOException e) { + logger.error("Exception accepting new connection from " + + channel.socket().getInetAddress().getHostAddress() + + ":" + channel.socket().getPort(), e); + channel.close(); + } } - this.channelMap.put(channel, connection); - channel.register(selectorForNewSocket, SelectionKey.OP_READ, connection); - connection.publishConnectionOpenEvent(); } - catch (IOException e) { - logger.error("Exception accepting new connection from " - + channel.socket().getInetAddress().getHostAddress() - + ":" + channel.socket().getPort(), e); - channel.close(); - } - } + } while (this.multiAccept && channel != null); } catch (IOException e) { throw new UncheckedIOException(e); diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionTests.java index d2be2f01f4..a6e7f79238 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionTests.java @@ -41,6 +41,7 @@ import java.nio.channels.Selector; import java.nio.channels.SocketChannel; import java.util.ArrayList; import java.util.Arrays; +import java.util.Collections; import java.util.HashMap; import java.util.HashSet; import java.util.List; @@ -73,6 +74,7 @@ import org.springframework.context.ApplicationEvent; import org.springframework.context.ApplicationEventPublisher; import org.springframework.core.task.AsyncTaskExecutor; import org.springframework.core.task.SimpleAsyncTaskExecutor; +import org.springframework.integration.ip.event.IpIntegrationEvent; import org.springframework.integration.ip.tcp.connection.TcpNioConnection.ChannelInputStream; import org.springframework.integration.ip.tcp.serializer.ByteArrayCrLfSerializer; import org.springframework.integration.ip.tcp.serializer.MapJsonSerializer; @@ -506,7 +508,7 @@ public class TcpNioConnectionTests { doAnswer(new Answer() { @Override - public Integer answer(InvocationOnMock invocation) throws Throwable { + public Integer answer(InvocationOnMock invocation) { ByteBuffer buff = invocation.getArgument(0); byte[] bytes = written.toByteArray(); buff.put(bytes); @@ -837,6 +839,57 @@ public class TcpNioConnectionTests { assertThat(watch.getLastTaskTimeMillis()).isLessThan(950L); } + @Test + public void testMultiAccept() throws InterruptedException, IOException { + testMulti(true); + } + + @Test + public void testNoMultiAccept() throws InterruptedException, IOException { + testMulti(false); + } + + private void testMulti(boolean multiAccept) throws InterruptedException, IOException { + CountDownLatch serverReadyLatch = new CountDownLatch(1); + CountDownLatch latch = new CountDownLatch(21); + List sockets = new ArrayList<>(); + TcpNioServerConnectionFactory server = new TcpNioServerConnectionFactory(0); + try { + List events = Collections.synchronizedList(new ArrayList<>()); + List> messages = Collections.synchronizedList(new ArrayList<>()); + server.setMultiAccept(multiAccept); + server.setApplicationEventPublisher(e -> { + if (e instanceof TcpConnectionServerListeningEvent) { + serverReadyLatch.countDown(); + } + events.add((IpIntegrationEvent) e); + latch.countDown(); + }); + server.registerListener(m -> { + messages.add(m); + latch.countDown(); + return false; + }); + server.afterPropertiesSet(); + server.start(); + assertThat(serverReadyLatch.await(10, TimeUnit.SECONDS)).isTrue(); + for (int i = 0; i < 10; i++) { + Socket socket = SocketFactory.getDefault().createSocket("localhost", server.getPort()); + socket.getOutputStream().write("foo\r\n".getBytes()); + sockets.add(socket); + } + assertThat(latch.await(10, TimeUnit.SECONDS)); + assertThat(events).hasSize(11); // server ready + 10 opens + assertThat(messages).hasSize(10); + } + finally { + for (Socket socket : sockets) { + socket.close(); + } + server.stop(); + } + } + private void readFully(InputStream is, byte[] buff) throws IOException { for (int i = 0; i < buff.length; i++) { buff[i] = (byte) is.read(); diff --git a/src/reference/asciidoc/ip.adoc b/src/reference/asciidoc/ip.adoc index cb9bfa3a40..8311a3044a 100644 --- a/src/reference/asciidoc/ip.adoc +++ b/src/reference/asciidoc/ip.adoc @@ -974,6 +974,10 @@ Alternatively, you can insert a resequencer downstream of the inbound endpoint t If you set `apply-sequence` to `true` on the connection factory, messages arriving on a TCP connection have `sequenceNumber` and `correlationId` headers set. The resequencer uses these headers to return the messages to their proper sequence. +IMPORTANT: Starting with version 5.1.4, priority is given to accepting new connections over reading from existing connections. +This should, generally, have little impact unless you have a very high rate of new incoming connections. +If you wish to revert to the previous behavior of giving reads priority, set the `multiAccept` property on the `TcpNioServerConnectionFactory` to `false`. + ==== Pool Size The pool size attribute is no longer used. diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 23309e19cd..2effb32a7e 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -36,3 +36,6 @@ See <> for more information. The length header used by the `ByteArrayLengthHeaderSerializer` can now include the length of the header in addition to the payload. See <> for more information. + +When using a `TcpNioServerConnectionFactory`, priority is now given to accepting new connections over reading from existing connections, but it is configurable. +See <> for more information.