Poll System.in for stdin in CommandLineUtils.executeCommandLine

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();
                     }
 

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.
+ * <p>
+ * Ported from {@code org.apache.maven.shared.utils.cli.StreamPollFeeder} of Apache Maven Shared Utils
+ * (<a href="https://issues.apache.org/jira/browse/MSHARED-1072">MSHARED-1072</a>). The input stream is never closed
+ * by this class.
+ * </p>
+ */
+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();
+        }
+    }
+}

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;
+    }
+}