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

GH-3507: Fix Tail producer for proper command

Browse Source 
Fixes https://github.com/spring-projects/spring-integration/issues/3507

The `OSDelegatingFileTailingMessageProducer` passing command string to `Runtime.getRuntime().exec()`
may cause problems if spaces (and other special characters) are used in the filename.

* Use an array for command and its options to let the target `Runtime` to parse and
execute it properly

**Cherry-pick to 5.4.x, 5.3.x & 5.2.x**

# Conflicts:
#	spring-integration-file/src/main/java/org/springframework/integration/file/tail/OSDelegatingFileTailingMessageProducer.java
This commit is contained in:
Trung Pham
2021-03-11 16:10:30 +07:00
committed by Artem Bilan
parent b7d9b87201
commit d016dd5b6f
3 changed files with 87 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files

48
spring-integration-file/src/main/java/org/springframework/integration/file/tail/OSDelegatingFileTailingMessageProducer.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2019 the original author or authors.
* Copyright 2002-2021 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.
@@ -33,22 +33,28 @@ import org.springframework.util.Assert;
* @author Gary Russell
* @author Gavin Gray
* @author Ali Shahbour
* @author Artem Bilan
* @author Trung Pham
*
* @since 3.0
*
*/
public class OSDelegatingFileTailingMessageProducer extends FileTailingMessageProducerSupport
implements SchedulingAwareRunnable {
private volatile Process nativeTailProcess;
private volatile String options = "-F -n 0";
private volatile String command = "ADAPTER_NOT_INITIALIZED";
private volatile boolean enableStatusReader = true;
private volatile String[] tailCommand;
private volatile Process nativeTailProcess;
private volatile BufferedReader stdOutReader;
private volatile boolean enableStatusReader = true;
public void setOptions(String options) {
if (options == null) {
this.options = "";
@@ -92,8 +98,13 @@ public class OSDelegatingFileTailingMessageProducer extends FileTailingMessagePr
protected void doStart() {
super.doStart();
destroyProcess();
this.command = "tail " + this.options + " " + this.getFile().getAbsolutePath();
this.getTaskExecutor().execute(this::runExec);
String[] tailOptions = this.options.split("\\s+");
this.tailCommand = new String[tailOptions.length + 2];
this.tailCommand[0] = "tail";
this.tailCommand[this.tailCommand.length - 1] = getFile().getAbsolutePath();
System.arraycopy(tailOptions, 0, this.tailCommand, 1, tailOptions.length);
this.command = String.join(" ", this.tailCommand);
getTaskExecutor().execute(this::runExec);
}
@Override
@@ -114,20 +125,18 @@ public class OSDelegatingFileTailingMessageProducer extends FileTailingMessagePr
* Exec the native tail process.
*/
private void runExec() {
this.destroyProcess();
if (logger.isInfoEnabled()) {
logger.info("Starting tail process");
}
destroyProcess();
logger.info("Starting tail process");
try {
Process process = Runtime.getRuntime().exec(this.command);
Process process = Runtime.getRuntime().exec(this.tailCommand);
BufferedReader reader = new BufferedReader(new InputStreamReader(process.getInputStream()));
this.nativeTailProcess = process;
this.startProcessMonitor();
startProcessMonitor();
if (this.enableStatusReader) {
startStatusReader();
}
this.stdOutReader = reader;
this.getTaskExecutor().execute(this);
getTaskExecutor().execute(this);
}
catch (IOException e) {
throw new MessagingException("Failed to exec tail command: '" + this.command + "'", e);
@@ -139,14 +148,13 @@ public class OSDelegatingFileTailingMessageProducer extends FileTailingMessagePr
* Runs a thread that waits for the Process result.
*/
private void startProcessMonitor() {
this.getTaskExecutor().execute(() -> {
Process process = OSDelegatingFileTailingMessageProducer.this.nativeTailProcess;
if (process == null) {
if (logger.isDebugEnabled()) {
logger.debug("Process destroyed before starting process monitor");
}
return;
}
getTaskExecutor()
.execute(() -> {
Process process = OSDelegatingFileTailingMessageProducer.this.nativeTailProcess;
if (process == null) {
logger.debug("Process destroyed before starting process monitor");
return;
}
int result = Integer.MIN_VALUE;
try {

41
spring-integration-file/src/test/java/org/springframework/integration/file/tail/FileTailingMessageProducerTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2019 the original author or authors.
* Copyright 2002-2021 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.
@@ -53,6 +53,7 @@ import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
* @author Gavin Gray
* @author Artem Bilan
* @author Ali Shahbour
* @author Trung Pham
*
* @since 3.0
*/
@@ -183,7 +184,7 @@ public class FileTailingMessageProducerTests {
ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
taskScheduler.afterPropertiesSet();
adapter.setTaskScheduler(taskScheduler);
final List<FileTailingEvent> events = new ArrayList<FileTailingEvent>();
final List<FileTailingEvent> events = new ArrayList<>();
adapter.setApplicationEventPublisher(event -> {
FileTailingEvent tailEvent = (FileTailingEvent) event;
logger.debug(event);
@@ -232,6 +233,8 @@ public class FileTailingMessageProducerTests {
}
assertThat(events.size()).isGreaterThanOrEqualTo(1);
taskScheduler.destroy();
}
private void waitForField(FileTailingMessageProducerSupport adapter, String field) throws Exception {
@@ -248,4 +251,38 @@ public class FileTailingMessageProducerTests {
fail("adapter failed to start");
}
@Test
@TailAvailable
public void canHandleFilenameHavingSpecialCharacters() throws Exception {
File file = File.createTempFile("foo bar", " -c 1");
file.delete();
OSDelegatingFileTailingMessageProducer adapter = new OSDelegatingFileTailingMessageProducer();
adapter.setOptions(TAIL_OPTIONS_FOLLOW_NAME_ALL_LINES);
adapter.setFile(file);
ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
taskScheduler.afterPropertiesSet();
adapter.setTaskScheduler(taskScheduler);
QueueChannel outputChannel = new QueueChannel();
adapter.setOutputChannel(outputChannel);
adapter.setTailAttemptsDelay(500);
adapter.setBeanFactory(mock(BeanFactory.class));
adapter.afterPropertiesSet();
adapter.start();
waitForField(adapter, "stdOutReader");
FileOutputStream fos = new FileOutputStream(file);
fos.write(("hello foobar\n").getBytes());
fos.close();
Message<?> message = outputChannel.receive(10000);
assertThat(message).as("expected a non-null message").isNotNull();
assertThat(message.getPayload()).isEqualTo("hello foobar");
adapter.stop();
file.delete();
taskScheduler.destroy();
}
}

37
spring-integration-file/src/test/java/org/springframework/integration/file/tail/TailRule.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2019 the original author or authors.
* Copyright 2002-2021 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.
@@ -24,6 +24,7 @@ import java.io.IOException;
import java.io.OutputStream;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
@@ -75,9 +76,6 @@ public class TailRule extends TestWatcher {
}
private boolean tailWorksOnThisMachine() {
if (tmpDir.contains(":")) {
return false;
}
File testDir = new File(tmpDir, "FileTailingMessageProducerTests");
testDir.mkdir();
final File file = new File(testDir, "foo");
@@ -88,19 +86,24 @@ public class TailRule extends TestWatcher {
fos.close();
final AtomicReference<Integer> c = new AtomicReference<>();
final CountDownLatch latch = new CountDownLatch(1);
Future<Process> future = Executors.newSingleThreadExecutor().submit(() -> {
final Process process = Runtime.getRuntime().exec(commandToTest + " " + file.getAbsolutePath());
Executors.newSingleThreadExecutor().execute(() -> {
try {
c.set(process.getInputStream().read());
latch.countDown();
}
catch (IOException e) {
logger.error("Error reading test stream", e);
}
});
return process;
});
ExecutorService newSingleThreadExecutor = Executors.newSingleThreadExecutor();
Future<Process> future =
newSingleThreadExecutor.submit(() -> {
final Process process = Runtime.getRuntime().exec(commandToTest + " " + file.getAbsolutePath());
ExecutorService executorService = Executors.newSingleThreadExecutor();
executorService.execute(() -> {
try {
c.set(process.getInputStream().read());
latch.countDown();
}
catch (IOException e) {
logger.error("Error reading test stream", e);
}
});
executorService.shutdown();
return process;
});
newSingleThreadExecutor.shutdown();
try {
Process process = future.get(10, TimeUnit.SECONDS);
if (latch.await(10, TimeUnit.SECONDS)) {