From d016dd5b6fb80d99df9be720e92b4b585d2f62f4 Mon Sep 17 00:00:00 2001 From: Trung Pham Date: Thu, 11 Mar 2021 16:10:30 +0700 Subject: [PATCH] GH-3507: Fix Tail producer for proper command 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 --- ...SDelegatingFileTailingMessageProducer.java | 48 +++++++++++-------- .../tail/FileTailingMessageProducerTests.java | 41 +++++++++++++++- .../integration/file/tail/TailRule.java | 37 +++++++------- 3 files changed, 87 insertions(+), 39 deletions(-) diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/tail/OSDelegatingFileTailingMessageProducer.java b/spring-integration-file/src/main/java/org/springframework/integration/file/tail/OSDelegatingFileTailingMessageProducer.java index 0ce8d0051b..7178ab70e8 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/tail/OSDelegatingFileTailingMessageProducer.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/tail/OSDelegatingFileTailingMessageProducer.java @@ -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 { diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/tail/FileTailingMessageProducerTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/tail/FileTailingMessageProducerTests.java index c2814ed7dc..2912f33119 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/tail/FileTailingMessageProducerTests.java +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/tail/FileTailingMessageProducerTests.java @@ -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 events = new ArrayList(); + final List 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(); + } + } diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/tail/TailRule.java b/spring-integration-file/src/test/java/org/springframework/integration/file/tail/TailRule.java index 3e98ae28df..58f685fab6 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/tail/TailRule.java +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/tail/TailRule.java @@ -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 c = new AtomicReference<>(); final CountDownLatch latch = new CountDownLatch(1); - Future 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 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)) {