diff --git a/src/main/java/org/codehaus/plexus/util/cli/CommandLineUtils.java b/src/main/java/org/codehaus/plexus/util/cli/CommandLineUtils.java index b2ac489e..5ed4c9c9 100644 --- a/src/main/java/org/codehaus/plexus/util/cli/CommandLineUtils.java +++ b/src/main/java/org/codehaus/plexus/util/cli/CommandLineUtils.java @@ -82,7 +82,8 @@ public static int executeCommandLine( /** * @param cl The command line to execute - * @param systemIn The input to read from, must be thread safe + * @param systemIn The input to read from, must be thread safe. {@link System#in} is polled for available data, so + * the call does not wait for it once the process has ended * @param systemOut A consumer that receives output, must be thread safe * @param systemErr A consumer that receives system error stream output, must be thread safe * @param timeoutInSeconds Positive integer to specify timeout, zero and negative integers for no timeout. @@ -105,7 +106,8 @@ public static int executeCommandLine( * Immediately forks a process, returns a callable that will block until process is complete. * * @param cl The command line to execute - * @param systemIn The input to read from, must be thread safe + * @param systemIn The input to read from, must be thread safe. {@link System#in} is polled for available data, so + * the call does not wait for it once the process has ended * @param systemOut A consumer that receives output, must be thread safe * @param systemErr A consumer that receives system error stream output, must be thread safe * @param timeoutInSeconds Positive integer to specify timeout, zero and negative integers for no timeout. @@ -152,7 +154,12 @@ public Integer call() throws CommandLineException { boolean success = false; try { if (systemIn != null) { - inputFeeder = new StreamFeeder(systemIn, p.getOutputStream()); + if (systemIn == System.in) { + // may never deliver data nor end, so poll it instead of blocking in a read + inputFeeder = new StreamPollFeeder(systemIn, p.getOutputStream()); + } else { + inputFeeder = new StreamFeeder(systemIn, p.getOutputStream()); + } inputFeeder.start(); } diff --git a/src/main/java/org/codehaus/plexus/util/cli/StreamPollFeeder.java b/src/main/java/org/codehaus/plexus/util/cli/StreamPollFeeder.java new file mode 100644 index 00000000..db30e44e --- /dev/null +++ b/src/main/java/org/codehaus/plexus/util/cli/StreamPollFeeder.java @@ -0,0 +1,144 @@ +package org.codehaus.plexus.util.cli; + +/* + * Copyright The Codehaus Foundation. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.util.Objects; + +/** + * Polls an {@link InputStream} for available data and writes it to an {@link OutputStream}. Unlike + * {@link StreamFeeder} it never blocks in a read, so it can be stopped at any time even when the input is a stream + * such as {@code System.in} that has no data and no end. + *

+ * Ported from {@code org.apache.maven.shared.utils.cli.StreamPollFeeder} of Apache Maven Shared Utils + * (MSHARED-1072). The input stream is never closed + * by this class. + *

+ */ +class StreamPollFeeder extends StreamFeeder { + + private static final int BUF_LEN = 80; + + private static final long POLL_INTERVAL_MILLIS = 100; + + private final InputStream pollInput; + + private final OutputStream pollOutput; + + private final Object lock = new Object(); + + private volatile Throwable pollException; + + private volatile boolean stop; + + /** + * Create a new StreamPollFeeder. + * + * @param input stream to read from + * @param output stream to write to + */ + StreamPollFeeder(InputStream input, OutputStream output) { + super(input, output); + this.pollInput = Objects.requireNonNull(input); + this.pollOutput = Objects.requireNonNull(output); + setDaemon(true); + } + + @Override + public void run() { + byte[] buf = new byte[BUF_LEN]; + + try { + while (!stop) { + if (pollInput.available() > 0) { + int i = pollInput.read(buf); + if (i > 0) { + if (!isDisabled()) { + pollOutput.write(buf, 0, i); + pollOutput.flush(); + } + } else { + stop = true; + } + } else { + synchronized (lock) { + if (!stop) { + lock.wait(POLL_INTERVAL_MILLIS); + } + } + } + } + } catch (IOException e) { + pollException = e; + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } finally { + closeOutput(); + synchronized (this) { + setDone(); + notifyAll(); + } + } + } + + private void closeOutput() { + try { + pollOutput.close(); + } catch (IOException e) { + if (pollException == null) { + pollException = e; + } + } + } + + /** + * Stops polling and waits for the feeder thread to finish. + */ + @Override + public void waitUntilDone() throws InterruptedException { + stopPolling(); + join(); + } + + @Override + public void disable() { + super.disable(); + stopPolling(); + } + + /** + * The output stream is closed when the thread ends, and the input stream is left open, so there is nothing to do. + */ + @Override + public void close() { + // nothing to close here + } + + @Override + public Throwable getException() { + return pollException; + } + + private void stopPolling() { + synchronized (lock) { + stop = true; + lock.notifyAll(); + } + } +} diff --git a/src/test/java/org/codehaus/plexus/util/cli/StreamPollFeederTest.java b/src/test/java/org/codehaus/plexus/util/cli/StreamPollFeederTest.java new file mode 100644 index 00000000..19514920 --- /dev/null +++ b/src/test/java/org/codehaus/plexus/util/cli/StreamPollFeederTest.java @@ -0,0 +1,187 @@ +package org.codehaus.plexus.util.cli; + +/* + * Copyright The Codehaus Foundation. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.File; +import java.io.IOException; +import java.io.InputStream; +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.util.concurrent.CountDownLatch; + +import org.codehaus.plexus.util.Os; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.condition.DisabledOnOs; +import org.junit.jupiter.api.condition.OS; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTimeoutPreemptively; +import static org.junit.jupiter.api.Assertions.assertTrue; + +class StreamPollFeederTest { + + /** An input that never has data and never ends, like an idle terminal. */ + private static final class IdleInputStream extends InputStream { + private final CountDownLatch never = new CountDownLatch(1); + + @Override + public int available() { + return 0; + } + + @Override + public int read() throws IOException { + try { + never.await(); + } catch (InterruptedException e) { + throw new IOException(e); + } + return -1; + } + } + + private static final class CountingOutputStream extends ByteArrayOutputStream { + private volatile boolean closed; + + @Override + public void close() { + closed = true; + } + } + + @Test + void waitUntilDoneWithIdleInput() { + CountingOutputStream out = new CountingOutputStream(); + StreamPollFeeder feeder = new StreamPollFeeder(new IdleInputStream(), out); + + assertTimeoutPreemptively(Duration.ofSeconds(5), () -> { + feeder.start(); + Thread.sleep(150); + feeder.waitUntilDone(); + }); + + assertNull(feeder.getException()); + assertTrue(feeder.isDone()); + assertTrue(out.closed, "output must be closed when feeding ends"); + } + + @Test + void waitUntilDoneWithContinuousData() { + InputStream continuous = new InputStream() { + @Override + public int available() { + return 1; + } + + @Override + public int read() { + return 0; + } + + @Override + public int read(byte[] b, int off, int len) { + return 1; + } + }; + StreamPollFeeder feeder = new StreamPollFeeder(continuous, new ByteArrayOutputStream() { + @Override + public synchronized void write(byte[] b, int off, int len) { + // discard + } + }); + + assertTimeoutPreemptively(Duration.ofSeconds(5), () -> { + feeder.start(); + feeder.waitUntilDone(); + }); + assertNull(feeder.getException()); + } + + @Test + void dataIsCopied() throws Exception { + StringBuilder data = new StringBuilder(); + for (int i = 0; i < 100; i++) { + data.append("TestData"); + } + ByteArrayOutputStream out = new ByteArrayOutputStream(); + StreamPollFeeder feeder = + new StreamPollFeeder(new ByteArrayInputStream(data.toString().getBytes(StandardCharsets.UTF_8)), out); + + assertTimeoutPreemptively(Duration.ofSeconds(10), () -> { + feeder.start(); + while (out.size() < data.length()) { + Thread.sleep(20); + } + feeder.waitUntilDone(); + }); + + assertNull(feeder.getException()); + assertEquals(data.toString(), out.toString("UTF-8")); + } + + @Test + @DisabledOnOs(OS.WINDOWS) + void executeCommandLineDoesNotWaitForIdleSystemIn() throws Exception { + InputStream originalIn = System.in; + System.setIn(new IdleInputStream()); + try { + Commandline cli = new Commandline(); + cli.setExecutable(javaExecutable().getAbsolutePath()); + cli.createArg().setValue("-version"); + + int exitCode = assertTimeoutPreemptively( + Duration.ofSeconds(30), + () -> CommandLineUtils.executeCommandLine( + cli, + System.in, + new CommandLineUtils.StringStreamConsumer(), + new CommandLineUtils.StringStreamConsumer())); + assertEquals(0, exitCode); + } finally { + System.setIn(originalIn); + } + } + + @Test + @DisabledOnOs(OS.WINDOWS) + void executeCommandLineStillDeliversEndOfInputToFiniteStream() throws Exception { + Commandline cli = new Commandline(); + cli.setExecutable("cat"); + CommandLineUtils.StringStreamConsumer out = new CommandLineUtils.StringStreamConsumer(); + + int exitCode = assertTimeoutPreemptively( + Duration.ofSeconds(30), + () -> CommandLineUtils.executeCommandLine( + cli, + new ByteArrayInputStream("hello".getBytes(StandardCharsets.UTF_8)), + out, + new CommandLineUtils.StringStreamConsumer())); + + assertEquals(0, exitCode); + assertEquals("hello", out.getOutput().trim()); + } + + private static File javaExecutable() { + File java = + new File(System.getProperty("java.home"), Os.isFamily(Os.FAMILY_WINDOWS) ? "bin/java.exe" : "bin/java"); + assertTrue(java.exists(), java + " doesn't exist"); + return java; + } +}