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

Fix compatibility with the latest SF

Browse Source 
* Mostly changes are related to the `TaskScheduler` and `Trigger` APIs
* Migrate to `micrometer-tracing` dependency
* Rework `SocketTestUtils` to use a `InetAddress.getLocalHost()`
for more stability and performance on Windows
* Fix docs for new `PeriodicTrigger` API
This commit is contained in:
Artem Bilan
2022-07-08 17:36:15 -04:00
parent 64aa4d5348
commit e0f137905a
46 changed files with 565 additions and 586 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpInboundGateway.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2020 the original author or authors.
* Copyright 2002-2022 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.
@@ -16,6 +16,7 @@
package org.springframework.integration.ip.tcp;
import java.time.Duration;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledFuture;
@@ -37,6 +38,7 @@ import org.springframework.integration.ip.tcp.connection.TcpSender;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.support.ErrorMessage;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.util.Assert;
/**
@@ -91,7 +93,7 @@ public class TcpInboundGateway extends MessagingGatewaySupport implements
else {
if (isErrorMessage) {
/*
* Socket errors are sent here so they can be conveyed to any waiting thread.
* Socket errors are sent here, so they can be conveyed to any waiting thread.
* There's not one here; simply ignore.
*/
return false;
@@ -224,8 +226,9 @@ public class TcpInboundGateway extends MessagingGatewaySupport implements
ClientModeConnectionManager manager =
new ClientModeConnectionManager(this.clientConnectionFactory);
this.clientModeConnectionManager = manager;
Assert.state(getTaskScheduler() != null, "Client mode requires a task scheduler");
this.scheduledFuture = getTaskScheduler().scheduleAtFixedRate(manager, this.retryInterval);
TaskScheduler taskScheduler = getTaskScheduler();
Assert.state(taskScheduler != null, "Client mode requires a task scheduler");
this.scheduledFuture = taskScheduler.scheduleAtFixedRate(manager, Duration.ofMillis(this.retryInterval));
}
}

52
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionReadTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2020 the original author or authors.
* Copyright 2002-2022 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.
@@ -19,6 +19,7 @@ package org.springframework.integration.ip.tcp.connection;
import static org.assertj.core.api.Assertions.assertThat;
import static org.awaitility.Awaitility.with;
import java.net.InetAddress;
import java.net.Socket;
import java.time.Duration;
import java.util.ArrayList;
@@ -48,18 +49,19 @@ import org.springframework.messaging.support.ErrorMessage;
*
* @since 2.0
*/
//@LongRunningTest
public class TcpNioConnectionReadTests {
private final CountDownLatch latch = new CountDownLatch(1);
private AbstractServerConnectionFactory getConnectionFactory(
AbstractByteArraySerializer serializer, TcpListener listener) throws Exception {
AbstractByteArraySerializer serializer, TcpListener listener) {
return getConnectionFactory(serializer, listener, null);
}
private AbstractServerConnectionFactory getConnectionFactory(
AbstractByteArraySerializer serializer, TcpListener listener, TcpSender sender) throws Exception {
AbstractByteArraySerializer serializer, TcpListener listener, TcpSender sender) {
TcpNioServerConnectionFactory scf = new TcpNioServerConnectionFactory(0);
scf.setUsingDirectBuffers(true);
scf.setApplicationEventPublisher(e -> {
@@ -78,7 +80,7 @@ public class TcpNioConnectionReadTests {
@Test
public void testReadLength() throws Exception {
ByteArrayLengthHeaderSerializer serializer = new ByteArrayLengthHeaderSerializer();
final List<Message<?>> responses = new ArrayList<Message<?>>();
final List<Message<?>> responses = new ArrayList<>();
final Semaphore semaphore = new Semaphore(0);
AbstractServerConnectionFactory scf = getConnectionFactory(serializer, message -> {
responses.add(message);
@@ -106,7 +108,7 @@ public class TcpNioConnectionReadTests {
@Test
public void testFragmented() throws Exception {
ByteArrayLengthHeaderSerializer serializer = new ByteArrayLengthHeaderSerializer();
final List<Message<?>> responses = new ArrayList<Message<?>>();
final List<Message<?>> responses = new ArrayList<>();
final Semaphore semaphore = new Semaphore(0);
AbstractServerConnectionFactory scf = getConnectionFactory(serializer, message -> {
responses.add(message);
@@ -137,7 +139,7 @@ public class TcpNioConnectionReadTests {
@Test
public void testReadStxEtx() throws Exception {
ByteArrayStxEtxSerializer serializer = new ByteArrayStxEtxSerializer();
final List<Message<?>> responses = new ArrayList<Message<?>>();
final List<Message<?>> responses = new ArrayList<>();
final Semaphore semaphore = new Semaphore(0);
AbstractServerConnectionFactory scf = getConnectionFactory(serializer, message -> {
responses.add(message);
@@ -164,7 +166,7 @@ public class TcpNioConnectionReadTests {
@Test
public void testReadCrLf() throws Exception {
ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer();
final List<Message<?>> responses = new ArrayList<Message<?>>();
final List<Message<?>> responses = new ArrayList<>();
final Semaphore semaphore = new Semaphore(0);
AbstractServerConnectionFactory scf = getConnectionFactory(serializer, message -> {
responses.add(message);
@@ -191,11 +193,11 @@ public class TcpNioConnectionReadTests {
public void testReadLengthOverflow() throws Exception {
ByteArrayLengthHeaderSerializer serializer = new ByteArrayLengthHeaderSerializer();
final Semaphore semaphore = new Semaphore(0);
final List<TcpConnection> added = new ArrayList<TcpConnection>();
final List<TcpConnection> removed = new ArrayList<TcpConnection>();
final List<TcpConnection> added = new ArrayList<>();
final List<TcpConnection> removed = new ArrayList<>();
final CountDownLatch errorMessageLetch = new CountDownLatch(1);
final AtomicReference<Throwable> errorMessageRef = new AtomicReference<Throwable>();
final AtomicReference<Throwable> errorMessageRef = new AtomicReference<>();
AbstractServerConnectionFactory scf = getConnectionFactory(serializer, message -> {
if (message instanceof ErrorMessage) {
@@ -299,11 +301,11 @@ public class TcpNioConnectionReadTests {
ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer();
serializer.setMaxMessageSize(1024);
final Semaphore semaphore = new Semaphore(0);
final List<TcpConnection> added = new ArrayList<TcpConnection>();
final List<TcpConnection> removed = new ArrayList<TcpConnection>();
final List<TcpConnection> added = new ArrayList<>();
final List<TcpConnection> removed = new ArrayList<>();
final CountDownLatch errorMessageLetch = new CountDownLatch(1);
final AtomicReference<Throwable> errorMessageRef = new AtomicReference<Throwable>();
final AtomicReference<Throwable> errorMessageRef = new AtomicReference<>();
AbstractServerConnectionFactory scf = getConnectionFactory(serializer, message -> {
if (message instanceof ErrorMessage) {
@@ -355,11 +357,11 @@ public class TcpNioConnectionReadTests {
ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer();
serializer.setMaxMessageSize(1024);
final Semaphore semaphore = new Semaphore(0);
final List<TcpConnection> added = new ArrayList<TcpConnection>();
final List<TcpConnection> removed = new ArrayList<TcpConnection>();
final List<TcpConnection> added = new ArrayList<>();
final List<TcpConnection> removed = new ArrayList<>();
final CountDownLatch errorMessageLetch = new CountDownLatch(1);
final AtomicReference<Throwable> errorMessageRef = new AtomicReference<Throwable>();
final AtomicReference<Throwable> errorMessageRef = new AtomicReference<>();
AbstractServerConnectionFactory scf = getConnectionFactory(serializer, message -> {
if (message instanceof ErrorMessage) {
@@ -408,11 +410,11 @@ public class TcpNioConnectionReadTests {
ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer();
serializer.setMaxMessageSize(1024);
final Semaphore semaphore = new Semaphore(0);
final List<TcpConnection> added = new ArrayList<TcpConnection>();
final List<TcpConnection> removed = new ArrayList<TcpConnection>();
final List<TcpConnection> added = new ArrayList<>();
final List<TcpConnection> removed = new ArrayList<>();
final CountDownLatch errorMessageLetch = new CountDownLatch(1);
final AtomicReference<Throwable> errorMessageRef = new AtomicReference<Throwable>();
final AtomicReference<Throwable> errorMessageRef = new AtomicReference<>();
AbstractServerConnectionFactory scf = getConnectionFactory(serializer, message -> {
if (message instanceof ErrorMessage) {
@@ -435,7 +437,7 @@ public class TcpNioConnectionReadTests {
}
});
Socket socket = SocketFactory.getDefault().createSocket("localhost", scf.getPort());
Socket socket = SocketFactory.getDefault().createSocket(InetAddress.getLocalHost(), scf.getPort());
socket.getOutputStream().write("partial".getBytes());
socket.close();
whileOpen(semaphore, added);
@@ -486,11 +488,11 @@ public class TcpNioConnectionReadTests {
private void testClosureMidMessageGuts(AbstractByteArraySerializer serializer, String shortMessage)
throws Exception {
final Semaphore semaphore = new Semaphore(0);
final List<TcpConnection> added = new ArrayList<TcpConnection>();
final List<TcpConnection> removed = new ArrayList<TcpConnection>();
final List<TcpConnection> added = new ArrayList<>();
final List<TcpConnection> removed = new ArrayList<>();
final CountDownLatch errorMessageLetch = new CountDownLatch(1);
final AtomicReference<Throwable> errorMessageRef = new AtomicReference<Throwable>();
final AtomicReference<Throwable> errorMessageRef = new AtomicReference<>();
AbstractServerConnectionFactory scf = getConnectionFactory(serializer, message -> {
if (message instanceof ErrorMessage) {
@@ -513,7 +515,7 @@ public class TcpNioConnectionReadTests {
}
});
Socket socket = SocketFactory.getDefault().createSocket("localhost", scf.getPort());
Socket socket = SocketFactory.getDefault().createSocket(InetAddress.getLocalHost(), scf.getPort());
socket.getOutputStream().write(shortMessage.getBytes());
socket.close();
whileOpen(semaphore, added);

24
spring-integration-ip/src/test/java/org/springframework/integration/ip/util/SocketTestUtils.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2020 the original author or authors.
* Copyright 2002-2022 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.
@@ -62,7 +62,7 @@ public class SocketTestUtils {
public static CountDownLatch testSendLength(final int port, final CountDownLatch latch) {
final CountDownLatch testCompleteLatch = new CountDownLatch(1);
Thread thread = new Thread(() -> {
try (Socket socket = new Socket(InetAddress.getByName("localhost"), port)) {
try (Socket socket = new Socket(InetAddress.getLocalHost(), port)) {
for (int i = 0; i < 2; i++) {
byte[] len = new byte[4];
ByteBuffer.wrap(len).putInt(TEST_STRING.length() * 2);
@@ -94,7 +94,7 @@ public class SocketTestUtils {
public static CountDownLatch testSendLengthOverflow(final int port) {
final CountDownLatch testCompleteLatch = new CountDownLatch(1);
Thread thread = new Thread(() -> {
try (Socket socket = new Socket(InetAddress.getByName("localhost"), port)) {
try (Socket socket = new Socket(InetAddress.getLocalHost(), port)) {
byte[] len = new byte[4];
ByteBuffer.wrap(len).putInt(Integer.MAX_VALUE);
socket.getOutputStream().write(len);
@@ -120,7 +120,7 @@ public class SocketTestUtils {
Socket socket = null;
try {
logger.debug("Connecting to " + port);
socket = new Socket(InetAddress.getByName("localhost"), port);
socket = new Socket(InetAddress.getLocalHost(), port);
OutputStream os = socket.getOutputStream();
for (int i = 0; i < howMany; i++) {
writeByte(os, 0, noDelay);
@@ -162,12 +162,12 @@ public class SocketTestUtils {
/**
* Sends a STX/ETX message in two chunks. Two such messages are sent.
* @param latch If not null, await until counted down before sending second chunk.
* @param latch If not null, waits until counted down before sending second chunk.
*/
public static CountDownLatch testSendStxEtx(final int port, final CountDownLatch latch) {
final CountDownLatch testCompleteLatch = new CountDownLatch(1);
Thread thread = new Thread(() -> {
try (Socket socket = new Socket(InetAddress.getByName("localhost"), port)) {
try (Socket socket = new Socket(InetAddress.getLocalHost(), port)) {
OutputStream outputStream = socket.getOutputStream();
for (int i = 0; i < 2; i++) {
writeByte(outputStream, 0x02, true);
@@ -199,7 +199,7 @@ public class SocketTestUtils {
public static CountDownLatch testSendStxEtxOverflow(final int port) {
final CountDownLatch testCompleteLatch = new CountDownLatch(1);
Thread thread = new Thread(() -> {
try (Socket socket = new Socket(InetAddress.getByName("localhost"), port)) {
try (Socket socket = new Socket(InetAddress.getLocalHost(), port)) {
OutputStream outputStream = socket.getOutputStream();
writeByte(outputStream, 0x02, true);
for (int i = 0; i < 1500; i++) {
@@ -223,7 +223,7 @@ public class SocketTestUtils {
public static CountDownLatch testSendCrLf(final int port, final CountDownLatch latch) {
final CountDownLatch testCompleteLatch = new CountDownLatch(1);
Thread thread = new Thread(() -> {
try (Socket socket = new Socket(InetAddress.getByName("localhost"), port)) {
try (Socket socket = new Socket(InetAddress.getLocalHost(), port)) {
OutputStream outputStream = socket.getOutputStream();
for (int i = 0; i < 2; i++) {
outputStream.write(TEST_STRING.getBytes());
@@ -255,7 +255,7 @@ public class SocketTestUtils {
*/
public static void testSendCrLfSingle(final int port, final CountDownLatch latch) {
Thread thread = new Thread(() -> {
try (Socket socket = new Socket(InetAddress.getByName("localhost"), port)) {
try (Socket socket = new Socket(InetAddress.getLocalHost(), port)) {
OutputStream outputStream = socket.getOutputStream();
outputStream.write(TEST_STRING.getBytes());
outputStream.write(TEST_STRING.getBytes());
@@ -278,7 +278,7 @@ public class SocketTestUtils {
*/
public static void testSendRaw(final int port) {
Thread thread = new Thread(() -> {
try (Socket socket = new Socket(InetAddress.getByName("localhost"), port)) {
try (Socket socket = new Socket(InetAddress.getLocalHost(), port)) {
OutputStream outputStream = socket.getOutputStream();
outputStream.write(TEST_STRING.getBytes());
outputStream.write(TEST_STRING.getBytes());
@@ -298,7 +298,7 @@ public class SocketTestUtils {
public static CountDownLatch testSendSerialized(final int port) {
final CountDownLatch testCompleteLatch = new CountDownLatch(1);
Thread thread = new Thread(() -> {
try (Socket socket = new Socket(InetAddress.getByName("localhost"), port)) {
try (Socket socket = new Socket(InetAddress.getLocalHost(), port)) {
OutputStream outputStream = socket.getOutputStream();
ObjectOutputStream oos = new ObjectOutputStream(outputStream);
oos.writeObject(TEST_STRING);
@@ -323,7 +323,7 @@ public class SocketTestUtils {
public static CountDownLatch testSendCrLfOverflow(final int port) {
final CountDownLatch testCompleteLatch = new CountDownLatch(1);
Thread thread = new Thread(() -> {
try (Socket socket = new Socket(InetAddress.getByName("localhost"), port)) {
try (Socket socket = new Socket(InetAddress.getLocalHost(), port)) {
OutputStream outputStream = socket.getOutputStream();
for (int i = 0; i < 1500; i++) {
writeByte(outputStream, 'x', true);