Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-3256: Support Testing TCP Connections

Browse Source 
Resolves https://github.com/spring-projects/spring-integration/issues/3256
This commit is contained in:
Gary Russell
2020-04-23 15:26:50 -04:00
committed by Artem Bilan
parent 02407f7dff
commit 1d4626f54c
6 changed files with 265 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files

62
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractClientConnectionFactory.java
View File

@@ -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<TcpConnectionSupport> 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<TcpConnectionSupport> 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<TcpConnectionSupport> 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);
}
}
/**

33
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionSupport.java
View File

@@ -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;
}

6
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpSender.java
View File

@@ -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) {
}
}

138
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionFactoryTests.java
View File

@@ -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<TcpConnection> 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 {

60
src/reference/asciidoc/ip.adoc
View File

@@ -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 <<testing-connections>>.
[[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 <<failover-cf>> 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<TcpConnectionSupport> 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 <<failover-cf>> 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<String> 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

3
src/reference/asciidoc/whats-new.adoc
View File

@@ -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