From 1d4626f54c1c9669bc4fcbd3048a5dee1b3326e5 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Thu, 23 Apr 2020 15:26:50 -0400 Subject: [PATCH] GH-3256: Support Testing TCP Connections Resolves https://github.com/spring-projects/spring-integration/issues/3256 --- .../AbstractClientConnectionFactory.java | 62 ++++++-- .../tcp/connection/TcpConnectionSupport.java | 33 +++++ .../ip/tcp/connection/TcpSender.java | 6 +- .../connection/ConnectionFactoryTests.java | 138 +++++++++++++++--- src/reference/asciidoc/ip.adoc | 60 ++++++++ src/reference/asciidoc/whats-new.adoc | 3 + 6 files changed, 265 insertions(+), 37 deletions(-) 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 b4b5ac2718..00fa032e7a 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 @@ -16,10 +16,13 @@ package org.springframework.integration.ip.tcp.connection; +import java.io.IOException; +import java.io.UncheckedIOException; import java.net.Socket; import java.time.Duration; import java.util.concurrent.locks.ReadWriteLock; import java.util.concurrent.locks.ReentrantReadWriteLock; +import java.util.function.Predicate; import org.springframework.context.ApplicationEventPublisher; import org.springframework.lang.Nullable; @@ -44,6 +47,9 @@ public abstract class AbstractClientConnectionFactory extends AbstractConnection private Duration connectTimeout = Duration.ofSeconds(DEFAULT_CONNECT_TIMEOUT); + @Nullable + private Predicate connectionTest; + private volatile TcpConnectionSupport theConnection; /** @@ -68,6 +74,7 @@ public abstract class AbstractClientConnectionFactory extends AbstractConnection return this.connectTimeout; } + /** * Set whether to automatically (default) or manually add a {@link TcpListener} to the * connections created by this factory. By default, the factory automatically configures @@ -79,6 +86,27 @@ public abstract class AbstractClientConnectionFactory extends AbstractConnection this.manualListenerRegistration = true; } + /** + * Get a {@link Predicate} that will be invoked to test a new connection; return true + * to accept the connection, false the reject. + * @return the predicate. + * @since 5.3 + */ + @Nullable + protected Predicate getConnectionTest() { + return this.connectionTest; + } + + /** + * Set a {@link Predicate} that will be invoked to test a new connection; return true + * to accept the connection, false the reject. + * @param connectionTest the predicate. + * @since 5.3 + */ + public void setConnectionTest(@Nullable Predicate connectionTest) { + this.connectionTest = connectionTest; + } + /** * Obtains a connection - if {@link #setSingleUse(boolean)} was called with * true, a new connection is returned; otherwise a single connection is @@ -126,21 +154,12 @@ public abstract class AbstractClientConnectionFactory extends AbstractConnection if (!singleUse) { // Another write lock holder might have created a new one by now. connection = obtainSharedConnection(); - if (connection != null) { + if (connection != null && connection.isOpen()) { return connection; } } - if (logger.isDebugEnabled()) { - logger.debug("Opening new socket connection to " + getHost() + ":" + getPort()); - } - - connection = buildNewConnection(); - if (!singleUse) { - setTheConnection(connection); - } - connection.publishConnectionOpenEvent(); - return connection; + return doObtain(singleUse); } catch (RuntimeException e) { ApplicationEventPublisher applicationEventPublisher = getApplicationEventPublisher(); @@ -156,6 +175,24 @@ public abstract class AbstractClientConnectionFactory extends AbstractConnection } } + private TcpConnectionSupport doObtain(boolean singleUse) { + TcpConnectionSupport connection; + if (logger.isDebugEnabled()) { + logger.debug("Opening new socket connection to " + getHost() + ":" + getPort()); + } + + connection = buildNewConnection(); + if (this.connectionTest != null && !this.connectionTest.test(connection)) { + connection.close(); + throw new UncheckedIOException(new IOException("Connection test failed for " + connection)); + } + if (!singleUse) { + setTheConnection(connection); + } + connection.publishConnectionOpenEvent(); + return connection; + } + protected TcpConnectionSupport buildNewConnection() { throw new UnsupportedOperationException( "Factories that don't override this class' obtainConnection() must implement this method"); @@ -187,6 +224,9 @@ public abstract class AbstractClientConnectionFactory extends AbstractConnection connection.setMapper(getMapper()); connection.setDeserializer(getDeserializer()); connection.setSerializer(getSerializer()); + if (this.connectionTest != null) { + connection.setNeedsTest(true); + } } /** 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 1132a55713..aecb068fd4 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 @@ -78,6 +78,8 @@ public abstract class TcpConnectionSupport implements TcpConnection { private TcpListener listener; + private volatile TcpListener testListener; + private TcpSender sender; private String connectionId; @@ -92,6 +94,13 @@ public abstract class TcpConnectionSupport implements TcpConnection { private boolean manualListenerRegistration; + /* + * This boolean is to avoid looking for a temporary listener when not needed + * to avoid a CPU cache flush. This does not have to be volatile because it + * is reset by the thread that checks for the temporary listener. + */ + private boolean needsTest; + public TcpConnectionSupport() { this(null); } @@ -238,6 +247,15 @@ public abstract class TcpConnectionSupport implements TcpConnection { } } + /** + * Set to true to use a temporary listener for just the first incoming message. + * @param needsTest true for a temporary listener. + * @since 5.3 + */ + public void setNeedsTest(boolean needsTest) { + this.needsTest = needsTest; + } + /** * Set the listener that will receive incoming Messages. * @param listener The listener. @@ -247,6 +265,17 @@ public abstract class TcpConnectionSupport implements TcpConnection { this.listenerRegisteredLatch.countDown(); } + /** + * Set a temporary listener to receive just the first incoming message. + * Used in conjunction with a connectionTest in a client connection + * factory. + * @param tListener the test listener. + * @since 5.3 + */ + public void registerTestListener(TcpListener tListener) { + this.testListener = tListener; + } + /** * Set whether or not automatic or manual registration of the {@link TcpListener} is to be * used. (Default automatic). When manual registration is in place, incoming messages will @@ -282,6 +311,10 @@ public abstract class TcpConnectionSupport implements TcpConnection { } waitForListenerRegistration(); } + if (this.needsTest && this.testListener != null) { + this.needsTest = false; + return this.testListener; + } return this.listener; } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpSender.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpSender.java index f2986b473e..635bd67fec 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpSender.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpSender.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. @@ -23,6 +23,7 @@ package org.springframework.integration.ip.tcp.connection; * @since 2.0 * */ +@FunctionalInterface public interface TcpSender { /** @@ -37,6 +38,7 @@ public interface TcpSender { * method is called each time a connection is closed. * @param connection The connection. */ - void removeDeadConnection(TcpConnection connection); + default void removeDeadConnection(TcpConnection connection) { + } } diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionFactoryTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionFactoryTests.java index f1634bcf3f..e57eea78df 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionFactoryTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionFactoryTests.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. @@ -17,6 +17,7 @@ package org.springframework.integration.ip.tcp.connection; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatExceptionOfType; import static org.assertj.core.api.Assertions.fail; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.contains; @@ -30,29 +31,36 @@ import static org.mockito.Mockito.when; import java.net.InetSocketAddress; import java.net.SocketAddress; import java.util.ArrayList; +import java.util.Arrays; import java.util.Collections; import java.util.List; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; -import org.junit.Rule; -import org.junit.Test; +import org.junit.jupiter.api.Test; import org.mockito.ArgumentCaptor; import org.springframework.beans.DirectFieldAccessor; import org.springframework.beans.factory.BeanFactory; -import org.springframework.context.ApplicationEvent; import org.springframework.context.ApplicationEventPublisher; import org.springframework.core.task.SimpleAsyncTaskExecutor; import org.springframework.integration.channel.NullChannel; +import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.context.IntegrationContextUtils; import org.springframework.integration.ip.config.TcpConnectionFactoryFactoryBean; import org.springframework.integration.ip.event.IpIntegrationEvent; +import org.springframework.integration.ip.tcp.TcpOutboundGateway; import org.springframework.integration.ip.tcp.TcpReceivingChannelAdapter; -import org.springframework.integration.test.rule.Log4j2LevelAdjuster; import org.springframework.integration.test.util.TestUtils; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessagingException; +import org.springframework.messaging.support.ErrorMessage; +import org.springframework.messaging.support.GenericMessage; import org.springframework.scheduling.TaskScheduler; import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; @@ -65,9 +73,6 @@ import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; */ public class ConnectionFactoryTests { - @Rule - public Log4j2LevelAdjuster adjuster = Log4j2LevelAdjuster.trace(); - @Test public void factoryBeanTests() { TcpConnectionFactoryFactoryBean fb = new TcpConnectionFactoryFactoryBean("client"); @@ -98,23 +103,13 @@ public class ConnectionFactoryTests { : 5; // Listening, + OPEN, CLOSE (but we *might* get exceptions, depending on timing). final CountDownLatch serverListeningLatch = new CountDownLatch(1); final CountDownLatch eventLatch = new CountDownLatch(expectedEvents); - ApplicationEventPublisher publisher = new ApplicationEventPublisher() { - - @Override - public void publishEvent(ApplicationEvent event) { - LogFactory.getLog(this.getClass()).trace("Received: " + event); - events.add((IpIntegrationEvent) event); - if (event instanceof TcpConnectionServerListeningEvent) { - serverListeningLatch.countDown(); - } - eventLatch.countDown(); + ApplicationEventPublisher publisher = event -> { + LogFactory.getLog(this.getClass()).trace("Received: " + event); + events.add((IpIntegrationEvent) event); + if (event instanceof TcpConnectionServerListeningEvent) { + serverListeningLatch.countDown(); } - - @Override - public void publishEvent(Object event) { - - } - + eventLatch.countDown(); }; serverFactory.setBeanName("serverFactory"); serverFactory.setApplicationEventPublisher(publisher); @@ -247,6 +242,101 @@ public class ConnectionFactoryTests { factory.stop(); } + @Test + void healthCheckSuccessNet() throws InterruptedException { + healthCheckSuccess(new TcpNetServerConnectionFactory(0), false); + } + + @Test + void healthCheckSuccessNio() throws InterruptedException { + healthCheckSuccess(new TcpNioServerConnectionFactory(0), false); + } + + @Test + void healthCheckFailureNet() throws InterruptedException { + healthCheckSuccess(new TcpNetServerConnectionFactory(0), true); + } + + @Test + void healthCheckFailureNio() throws InterruptedException { + healthCheckSuccess(new TcpNioServerConnectionFactory(0), true); + } + + private void healthCheckSuccess(AbstractServerConnectionFactory server, boolean fail) throws InterruptedException { + CountDownLatch serverUp = new CountDownLatch(1); + server.setApplicationEventPublisher(event -> { + if (event instanceof TcpConnectionServerListeningEvent) { + serverUp.countDown(); + } + }); + server.setBeanFactory(mock(BeanFactory.class)); + AtomicReference connection = new AtomicReference<>(); + server.registerSender(conn -> { + connection.set(conn); + }); + AtomicInteger tested = new AtomicInteger(); + server.registerListener(msg -> { + if (!(msg instanceof ErrorMessage)) { + String payload = new String((byte[]) msg.getPayload()); + if (payload.equals("PING")) { + tested.incrementAndGet(); + connection.get().send(new GenericMessage<>(fail ? "PANG" : "PONG")); + } + else { + connection.get().send(msg); + } + } + return false; + }); + server.start(); + assertThat(serverUp.await(10, TimeUnit.SECONDS)).isTrue(); + TcpNetClientConnectionFactory clientFactory = new TcpNetClientConnectionFactory("localhost", server.getPort()); + clientFactory.setApplicationEventPublisher(event -> { }); + clientFactory.setConnectionTest(conn -> { + CountDownLatch latch = new CountDownLatch(1); + AtomicBoolean result = new AtomicBoolean(); + conn.registerTestListener(msg -> { + if (Arrays.equals("PONG".getBytes(), (byte[]) msg.getPayload())) { + result.set(true); + } + latch.countDown(); + return false; + }); + conn.send(new GenericMessage<>("PING")); + try { + latch.await(10, TimeUnit.SECONDS); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + return result.get(); + }); + TcpOutboundGateway gateway = new TcpOutboundGateway(); + gateway.setRemoteTimeout(60000); + gateway.setConnectionFactory(clientFactory); + QueueChannel outputChannel = new QueueChannel(); + gateway.setOutputChannel(outputChannel); + gateway.start(); + if (fail) { + assertThatExceptionOfType(MessagingException.class).isThrownBy(() -> + gateway.handleMessage(new GenericMessage<>("test1"))) + .withMessageContaining("Connection test failed for"); + } + else { + gateway.handleMessage(new GenericMessage<>("test1")); + Message received = outputChannel.receive(0); + assertThat(received).isNotNull(); + assertThat(received.getPayload()).isEqualTo("test1".getBytes()); + gateway.handleMessage(new GenericMessage<>("test2")); + received = outputChannel.receive(0); + assertThat(received).isNotNull(); + assertThat(received.getPayload()).isEqualTo("test2".getBytes()); + assertThat(tested.get()).isEqualTo(1); + } + gateway.stop(); + server.stop(); + } + @SuppressWarnings("serial") private class FooEvent extends TcpConnectionOpenEvent { diff --git a/src/reference/asciidoc/ip.adoc b/src/reference/asciidoc/ip.adoc index d293458d52..70eaaaaf5b 100644 --- a/src/reference/asciidoc/ip.adoc +++ b/src/reference/asciidoc/ip.adoc @@ -555,6 +555,8 @@ IMPORTANT: These properties do not apply if any of the delegate factories is a ` Starting with version 5.3, these default to `Long.MAX_VALUE` and `true` so the factory only attempts to fail back when the current connection fails. To revert to the default behavior of previous versions, set them to `0` and `false`. +Also see <>. + [[tcp-affinity-cf]] ==== TCP Thread Affinity Connection Factory @@ -592,6 +594,64 @@ public TcpOutboundGateway outGate() { ---- ==== +[[testing-connections]] +=== Testing Connections + +In some scenarios, it can be useful to send some kind of health-check request when a connection is first opened. +One such scenario might be when using a <> so that we can fail over if the selected server allowed a connection to be opened but reports that it is not healthy. + +In order to support this feature, add a `connectionTest` to the client connection factory. + +==== +[source, java] +---- +/** + * Set a {@link Predicate} that will be invoked to test a new connection; return true + * to accept the connection, false the reject. + * @param connectionTest the predicate. + * @since 5.3 + */ +public void setConnectionTest(@Nullable Predicate connectionTest) { + this.connectionTest = connectionTest; +} +---- +==== + +To test the connection, attach a temporary listener to the connection within the test. +If the test fails, the connection is closed and an exception thrown. +When used with the <> this triggers trying the next server. + +IMPORTANT: Only the first reply from the server will go to the test listener. + +In the following example, the server is considered healthy if the server replies `PONG` when we send `PING`. + +==== +[source, java] +---- +Message ping = new GenericMessage<>("PING"); +byte[] pong = "PONG".getBytes(); +clientFactory.setConnectionTest(conn -> { + CountDownLatch latch = new CountDownLatch(1); + AtomicBoolean result = new AtomicBoolean(); + conn.registerTestListener(msg -> { + if (Arrays.equals(pong, (byte[]) msg.getPayload())) { + result.set(true); + } + latch.countDown(); + return false; + }); + conn.send(ping); + try { + latch.await(10, TimeUnit.SECONDS); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + return result.get(); +}); +---- +==== + [[ip-interceptors]] === TCP Connection Interceptors diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index e9e7ad221f..0906e4a9fe 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -128,6 +128,9 @@ See <<./ip.adoc#failover-cf,TCP Failover Client Connection Factory>> for more in The `TcpOutboundGateway` now supports asynchronous request/reply. See <<./ip.adoc#tcp-gateways,TCP Gateways>> for more information. +You can now configure client connections to perform some arbitrary test on new connections. +See <<./ip.adoc#testing-connections,Testing Connections>> for more information. + [[x5.3-rsocket]] === RSocket Changes