From 34904b3cd6929b198d780e5dd4b74010469e4da1 Mon Sep 17 00:00:00 2001 From: abilan Date: Thu, 22 Jun 2023 15:41:38 -0400 Subject: [PATCH] Upgrade to commons-io-2.13.0; downgrade sshd-sftp * The latest `sshd-sftp-2.10.0` has a bug not removing trailing `.` in the "unrooted" path when we ask to create a remote directory. It works on Windows well, but fails on UNIX file systems. * The `commons-io-2.13.0` has `Tailer` ctors deprecated and exposes a builder API. It starts a tailer process though unconditionally in its own thread. Use `setStartThread(false)` to have the tailer process managed by our own `TaskExecutor` * Add `ApacheCommonsFileTailingMessageProducer.setPollingDelayDuration(Duration)` * Deprecate a `TailerListener` impl on the `ApacheCommonsFileTailingMessageProducer` in favor of an internal instance --- build.gradle | 4 +- ...acheCommonsFileTailingMessageProducer.java | 93 ++++++++++++++++--- ...eTailInboundChannelAdapterParserTests.java | 17 ++-- 3 files changed, 89 insertions(+), 25 deletions(-) diff --git a/build.gradle b/build.gradle index e9623498a6..bf75b4e73e 100644 --- a/build.gradle +++ b/build.gradle @@ -47,7 +47,7 @@ ext { modifiedFiles = files(grgit.status().unstaged.modified).filter { f -> f.name.endsWith('.java') || f.name.endsWith('.kt') } - apacheSshdVersion = '2.10.0' + apacheSshdVersion = '2.9.2' artemisVersion = '2.29.0' aspectjVersion = '1.9.19' assertjVersion = '3.24.2' @@ -56,7 +56,7 @@ ext { awaitilityVersion = '4.2.0' camelVersion = '3.20.6' commonsDbcp2Version = '2.9.0' - commonsIoVersion = '2.11.0' + commonsIoVersion = '2.13.0' commonsNetVersion = '3.9.0' curatorVersion = '5.5.0' debeziumVersion = '2.3.0.Final' diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/tail/ApacheCommonsFileTailingMessageProducer.java b/spring-integration-file/src/main/java/org/springframework/integration/file/tail/ApacheCommonsFileTailingMessageProducer.java index b2db668638..ee7b0010d5 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/tail/ApacheCommonsFileTailingMessageProducer.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/tail/ApacheCommonsFileTailingMessageProducer.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2020 the original author or authors. + * Copyright 2002-2023 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,8 +16,11 @@ package org.springframework.integration.file.tail; +import java.time.Duration; + import org.apache.commons.io.input.Tailer; import org.apache.commons.io.input.TailerListener; +import org.apache.commons.io.input.TailerListenerAdapter; /** * File tailer that delegates to the Apache Commons Tailer. @@ -31,7 +34,9 @@ import org.apache.commons.io.input.TailerListener; public class ApacheCommonsFileTailingMessageProducer extends FileTailingMessageProducerSupport implements TailerListener { - private long pollingDelay = 1000; // NOSONAR magic number + private final TailerListener tailerListener = new IntegrationTailerListener(); + + private Duration pollingDelay = Duration.ofSeconds(1); private boolean end = true; @@ -44,6 +49,15 @@ public class ApacheCommonsFileTailingMessageProducer extends FileTailingMessageP * @param pollingDelay The delay. */ public void setPollingDelay(long pollingDelay) { + setPollingDelayDuration(Duration.ofMillis(pollingDelay)); + } + + /** + * The delay between checks of the file for new content in {@link Duration}. + * @param pollingDelay The delay duration. + * @since 6.2 + */ + public void setPollingDelayDuration(Duration pollingDelay) { this.pollingDelay = pollingDelay; } @@ -73,7 +87,15 @@ public class ApacheCommonsFileTailingMessageProducer extends FileTailingMessageP @Override protected void doStart() { super.doStart(); - Tailer theTailer = new Tailer(getFile(), this, this.pollingDelay, this.end, this.reopen); + Tailer theTailer = + Tailer.builder() + .setDelayDuration(this.pollingDelay) + .setTailFromEnd(this.end) + .setReOpen(this.reopen) + .setFile(getFile()) + .setTailerListener(this.tailerListener) + .setStartThread(false) + .get(); getTaskExecutor().execute(theTailer); this.tailer = theTailer; } @@ -81,37 +103,80 @@ public class ApacheCommonsFileTailingMessageProducer extends FileTailingMessageP @Override protected void doStop() { super.doStop(); - this.tailer.stop(); + this.tailer.close(); } + @Deprecated(since = "6.2", forRemoval = true) @Override public void init(Tailer tailer) { + tailerListenerIsDeprecatedError(); } + @Deprecated(since = "6.2", forRemoval = true) @Override public void fileNotFound() { - publish("File not found: " + getFile().getAbsolutePath()); - try { - Thread.sleep(getMissingFileDelay()); - } - catch (InterruptedException e) { - Thread.currentThread().interrupt(); - } + tailerListenerIsDeprecatedError(); + this.tailerListener.fileNotFound(); } + @Deprecated(since = "6.2", forRemoval = true) @Override public void fileRotated() { - publish("File rotated: " + getFile().getAbsolutePath()); + tailerListenerIsDeprecatedError(); + this.tailerListener.fileRotated(); } + @Deprecated(since = "6.2", forRemoval = true) @Override public void handle(String line) { - send(line); + tailerListenerIsDeprecatedError(); + this.tailerListener.handle(line); } + @Deprecated(since = "6.2", forRemoval = true) @Override public void handle(Exception ex) { - publish(ex.getMessage()); + tailerListenerIsDeprecatedError(); + this.tailerListener.handle(ex); + } + + private void tailerListenerIsDeprecatedError() { + ApacheCommonsFileTailingMessageProducer.this.logger.error( + "The 'TailerListener' implementation on the 'ApacheCommonsFileTailingMessageProducer' " + + "is deprecated (in favor of an internal instance) for removal in the next version."); + } + + private class IntegrationTailerListener extends TailerListenerAdapter { + + IntegrationTailerListener() { + } + + @Override + public void fileNotFound() { + publish("File not found: " + getFile().getAbsolutePath()); + try { + Thread.sleep(getMissingFileDelay()); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + + @Override + public void fileRotated() { + publish("File rotated: " + getFile().getAbsolutePath()); + } + + @Override + public void handle(String line) { + send(line); + } + + @Override + public void handle(Exception ex) { + publish(ex.getMessage()); + } + } } diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileTailInboundChannelAdapterParserTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileTailInboundChannelAdapterParserTests.java index c6a5450278..b2da04b940 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileTailInboundChannelAdapterParserTests.java +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileTailInboundChannelAdapterParserTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2022 the original author or authors. + * Copyright 2002-2023 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,9 +17,9 @@ package org.springframework.integration.file.config; import java.io.File; +import java.time.Duration; -import org.junit.Test; -import org.junit.runner.RunWith; +import org.junit.jupiter.api.Test; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; @@ -31,8 +31,7 @@ import org.springframework.integration.test.util.TestUtils; import org.springframework.messaging.MessageChannel; import org.springframework.scheduling.TaskScheduler; import org.springframework.test.annotation.DirtiesContext; -import org.springframework.test.context.ContextConfiguration; -import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; import static org.assertj.core.api.Assertions.assertThat; import static org.mockito.Mockito.mock; @@ -42,10 +41,10 @@ import static org.mockito.Mockito.mock; * @author Artem Bilan * @author Gavin Gray * @author Ali Shahbour + * * @since 3.0 */ -@ContextConfiguration -@RunWith(SpringJUnit4ClassRunner.class) +@SpringJUnitConfig @DirtiesContext public class FileTailInboundChannelAdapterParserTests { @@ -109,7 +108,7 @@ public class FileTailInboundChannelAdapterParserTests { String normalizedName = getNormalizedPath(fileName); assertThat(normalizedName).isEqualTo("/tmp/bar"); assertThat(TestUtils.getPropertyValue(apacheDefault, "taskExecutor")).isSameAs(exec); - assertThat(TestUtils.getPropertyValue(apacheDefault, "pollingDelay")).isEqualTo(2000L); + assertThat(TestUtils.getPropertyValue(apacheDefault, "pollingDelay")).isEqualTo(Duration.ofSeconds(2)); assertThat(TestUtils.getPropertyValue(apacheDefault, "tailAttemptsDelay")).isEqualTo(10000L); assertThat(TestUtils.getPropertyValue(apacheDefault, "idleEventInterval")).isEqualTo(10000L); assertThat(TestUtils.getPropertyValue(apacheDefault, "autoStartup", Boolean.class)).isFalse(); @@ -124,7 +123,7 @@ public class FileTailInboundChannelAdapterParserTests { String normalizedName = getNormalizedPath(fileName); assertThat(normalizedName).isEqualTo("/tmp/qux"); assertThat(TestUtils.getPropertyValue(apacheEndReopen, "taskExecutor")).isSameAs(exec); - assertThat(TestUtils.getPropertyValue(apacheEndReopen, "pollingDelay")).isEqualTo(2000L); + assertThat(TestUtils.getPropertyValue(apacheEndReopen, "pollingDelay")).isEqualTo(Duration.ofSeconds(2)); assertThat(TestUtils.getPropertyValue(apacheEndReopen, "tailAttemptsDelay")).isEqualTo(10000L); assertThat(TestUtils.getPropertyValue(apacheEndReopen, "autoStartup", Boolean.class)).isFalse(); assertThat(TestUtils.getPropertyValue(apacheEndReopen, "phase")).isEqualTo(123);