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 19d9177d77..ce16c4e321 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 @@ -476,6 +476,7 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport * @see #setSslHandshakeTimeout(int) * @since 4.3.6 */ + @Nullable protected Integer getSslHandshakeTimeout() { return this.sslHandshakeTimeout; } @@ -724,7 +725,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 e8d5e8b508..ec902a050d 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 @@ -17,6 +17,7 @@ package org.springframework.integration.ip.tcp.connection; import java.io.IOException; +import java.io.UncheckedIOException; import java.net.InetAddress; import java.net.InetSocketAddress; import java.net.Socket; @@ -34,7 +35,7 @@ import org.springframework.lang.Nullable; import org.springframework.util.Assert; /** -/** + /** * Implements a server connection factory that produces {@link TcpNioConnection}s using * a {@link ServerSocketChannel}. Must have a {@link TcpListener} registered. * @@ -45,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; @@ -63,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"; @@ -193,47 +208,57 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto } /** - * @param selector The selector. + * @param selectorForNewSocket The selector. * @param server The server socket channel. * @param now The current time. - * @throws IOException Any IOException. */ @Override - protected void doAccept(final Selector selector, ServerSocketChannel server, long now) throws IOException { + protected void doAccept(final Selector selectorForNewSocket, ServerSocketChannel server, long now) { logger.debug("New accept"); - 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."); + try { + 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(); + } + 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(); + } + } + } } - channel.close(); + while (this.multiAccept && channel != null); } - 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(selector, SelectionKey.OP_READ, connection); - connection.publishConnectionOpenEvent(); - } - catch (Exception e) { - logger.error("Exception accepting new connection from " - + channel.socket().getInetAddress().getHostAddress() - + ":" + channel.socket().getPort(), e); - channel.close(); - } + catch (IOException e) { + throw new UncheckedIOException(e); } } @@ -241,7 +266,7 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto private TcpNioConnection createTcpNioConnection(SocketChannel socketChannel) { try { TcpNioConnection connection = this.tcpNioConnectionSupport.createNewConnection(socketChannel, true, - isLookupHost(), getApplicationEventPublisher(), getComponentName()); + isLookupHost(), getApplicationEventPublisher(), getComponentName()); connection.setUsingDirectBuffers(this.usingDirectBuffers); TcpConnectionSupport wrappedConnection = wrapConnection(connection); initializeConnection(wrappedConnection, socketChannel.socket()); 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 2c16a1623c..14c21de1ce 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 @@ -47,6 +47,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; @@ -79,6 +80,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; @@ -510,7 +512,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); @@ -840,6 +842,57 @@ public class TcpNioConnectionTests { assertThat(watch.getLastTaskTimeMillis(), lessThan(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(); + assertTrue(serverReadyLatch.await(10, TimeUnit.SECONDS)); + 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); + } + assertTrue(latch.await(10, TimeUnit.SECONDS)); + assertEquals(11, events.size()); // server ready + 10 opens + assertEquals(10, messages.size()); + } + 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 f3dc1c8cdb..49cc5189a6 100644 --- a/src/reference/asciidoc/ip.adoc +++ b/src/reference/asciidoc/ip.adoc @@ -971,6 +971,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.