Fix stdin behaviour when piping to mvnd

common/src/main/java/org/mvndaemon/mvnd/common/Message.java

@@ -65,6 +65,8 @@ public abstract class Message {
     public static final int PRINT_ERR = 26;
     public static final int REQUEST_INPUT = 27;
     public static final int INPUT_DATA = 28;
+    public static final int REQUEST_INPUT_AVAILABLE = 29;
+    public static final int INPUT_AVAILABLE_DATA = 30;
 
     final int type;
 
@@ -122,6 +124,10 @@ public static Message read(DataInputStream input) throws IOException {
                 return RequestInput.read(input);
             case INPUT_DATA:
                 return InputData.read(input);
+            case REQUEST_INPUT_AVAILABLE:
+                return RequestInputAvailable.read(input);
+            case INPUT_AVAILABLE_DATA:
+                return InputAvailableData.read(input);
         }
         throw new IllegalStateException("Unexpected message type: " + type);
     }
@@ -1123,6 +1129,66 @@ public void write(DataOutputStream output) throws IOException {
         }
     }
 
+    public static class RequestInputAvailable extends Message {
+
+        private final String projectId;
+
+        public static RequestInputAvailable read(DataInputStream input) throws IOException {
+            String projectId = readUTF(input);
+            return new RequestInputAvailable(projectId);
+        }
+
+        RequestInputAvailable(String projectId) {
+            super(REQUEST_INPUT_AVAILABLE);
+            this.projectId = projectId;
+        }
+
+        public String getProjectId() {
+            return projectId;
+        }
+
+        @Override
+        public String toString() {
+            return "RequestInputAvailable{" + "projectId='" + projectId + "'}";
+        }
+
+        @Override
+        public void write(DataOutputStream output) throws IOException {
+            super.write(output);
+            writeUTF(output, projectId);
+        }
+    }
+
+    public static class InputAvailableData extends Message {
+
+        private final int bytesAvailable;
+
+        public static InputAvailableData read(DataInputStream input) throws IOException {
+            int bytesAvailable = input.readInt();
+            return new InputAvailableData(bytesAvailable);
+        }
+
+        InputAvailableData(int bytesAvailable) {
+            super(INPUT_AVAILABLE_DATA);
+            this.bytesAvailable = bytesAvailable;
+        }
+
+        public int getBytesAvailable() {
+            return bytesAvailable;
+        }
+
+        @Override
+        public String toString() {
+            return "InputAvailableData{" + "bytesAvailable='" + bytesAvailable + "'}";
+        }
+
+        @Override
+        public void write(DataOutputStream output) throws IOException {
+            super.write(output);
+            output.writeInt(bytesAvailable);
+        }
+    }
+
     public int getType() {
         return type;
     }
@@ -1143,6 +1209,14 @@ public static InputData inputEof() {
         return new InputData(null);
     }
 
+    public static RequestInputAvailable requestInputAvailable(String projectId) {
+        return new RequestInputAvailable(projectId);
+    }
+
+    public static InputAvailableData inputAvailableData(int bytesAvailable) {
+        return new InputAvailableData(bytesAvailable);
+    }
+
     public static StringMessage out(String message) {
         return new StringMessage(PRINT_OUT, message);
     }

common/src/main/java/org/mvndaemon/mvnd/common/logging/TerminalInputHandler.java

@@ -67,25 +67,36 @@ private static class InputRequest {
         final String projectId; // null for control keys
         final Message.Prompt prompt; // non-null only for prompt requests
         final boolean isControlKey; // true for control key listening
+        final boolean isAvailableLength; // true for requesting number of available bytes
         final int bytesToRead; // max number of bytes to read
 
-        private InputRequest(String projectId, Message.Prompt prompt, boolean isControlKey, int bytesToRead) {
+        private InputRequest(
+                String projectId,
+                Message.Prompt prompt,
+                boolean isControlKey,
+                boolean isAvailableLength,
+                int bytesToRead) {
             this.projectId = projectId;
             this.prompt = prompt;
             this.isControlKey = isControlKey;
+            this.isAvailableLength = isAvailableLength;
             this.bytesToRead = bytesToRead;
         }
 
         static InputRequest forProject(String projectId, int bytesToRead) {
-            return new InputRequest(projectId, null, false, bytesToRead);
+            return new InputRequest(projectId, null, false, false, bytesToRead);
         }
 
         static InputRequest forPrompt(Message.Prompt prompt) {
-            return new InputRequest(prompt.getProjectId(), prompt, false, 0);
+            return new InputRequest(prompt.getProjectId(), prompt, false, false, 0);
         }
 
         static InputRequest forControlKeys() {
-            return new InputRequest(null, null, true, 0);
+            return new InputRequest(null, null, true, false, 0);
+        }
+
+        static InputRequest forAvailable(String projectId) {
+            return new InputRequest(projectId, null, false, true, 0);
         }
     }
 
@@ -108,6 +119,8 @@ public TerminalInputHandler(Terminal terminal, boolean dumb) {
                     } else if (request.prompt != null) {
                         // Always handle prompts
                         handlePrompt(request.prompt);
+                    } else if (request.isAvailableLength) {
+                        handleAvailableLength();
                     } else if (request.projectId != null) {
                         // Always handle project input
                         handleProjectInput(request.projectId, request.bytesToRead);
@@ -138,8 +151,13 @@ private void handleProjectInput(String projectId, int bytesToRead) throws IOExce
             int c = terminal.reader().read(timeout);
             if (c < 0) {
                 // End of stream reached
+                if (idx > 0) {
+                    String data = String.valueOf(buf, 0, idx);
+                    daemonReceive.accept(Message.inputResponse(data));
+                }
+
                 daemonReceive.accept(Message.inputEof());
-                break;
+                return;
             }
             buf[idx++] = (char) c;
             timeout = idx > 0 ? 1 : 10; // Shorter timeout after first char
@@ -199,6 +217,14 @@ private void handlePrompt(Message.Prompt prompt) throws IOException {
         }
     }
 
+    private void handleAvailableLength() throws IOException {
+        // terminal.reader().available() will attempt to return the number of available characters using some math, but
+        // this method powers System.in.available() when called from a mojo, which needs to return the number of
+        // available bytes.
+        int available = System.in.available();
+        daemonReceive.accept(Message.inputAvailableData(available));
+    }
+
     private boolean isControlKey(int c) {
         return c == TerminalOutput.KEY_PLUS
                 || c == TerminalOutput.KEY_MINUS
@@ -230,6 +256,11 @@ public void requestProjectInput(String projectId, int bytesToRead) {
         inputRequests.offer(InputRequest.forProject(projectId, bytesToRead));
     }
 
+    public void requestProjectInputAvailable(String projectId) {
+        inputRequests.clear();
+        inputRequests.offer(InputRequest.forAvailable(projectId));
+    }
+
     public void requestPrompt(Message.Prompt prompt) {
         inputRequests.clear(); // Clear any pending requests
         inputRequests.offer(InputRequest.forPrompt(prompt));

common/src/main/java/org/mvndaemon/mvnd/common/logging/TerminalOutput.java

@@ -50,6 +50,7 @@
 import org.mvndaemon.mvnd.common.Message.MojoStartedEvent;
 import org.mvndaemon.mvnd.common.Message.ProjectEvent;
 import org.mvndaemon.mvnd.common.Message.RequestInput;
+import org.mvndaemon.mvnd.common.Message.RequestInputAvailable;
 import org.mvndaemon.mvnd.common.Message.StringMessage;
 import org.mvndaemon.mvnd.common.Message.TransferEvent;
 import org.mvndaemon.mvnd.common.OsUtils;
@@ -414,6 +415,15 @@ private boolean doAccept(Message entry) {
                 daemonDispatch.accept(entry);
                 break;
             }
+            case Message.REQUEST_INPUT_AVAILABLE: {
+                RequestInputAvailable ri = (RequestInputAvailable) entry;
+                inputHandler.requestProjectInputAvailable(ri.getProjectId());
+                break;
+            }
+            case Message.INPUT_AVAILABLE_DATA: {
+                daemonDispatch.accept(entry);
+                break;
+            }
             default:
                 throw new IllegalStateException("Unexpected message " + entry);
         }

daemon/src/main/java/org/mvndaemon/mvnd/daemon/DaemonInputStream.java

@@ -24,7 +24,13 @@
 import java.nio.charset.Charset;
 import java.util.LinkedList;
 import java.util.Objects;
+import java.util.Optional;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
 import java.util.function.BiConsumer;
+import java.util.function.Consumer;
 
 import org.apache.maven.logging.ProjectBuildLogAppender;
 
@@ -45,13 +51,13 @@
  * - EOF is signaled by calling addInputData with null
  *
  * The stream coordinates between multiple threads:
- * - Reader thread(s): Calling read() methods to get input
- * - Writer thread: Calling addInputData to provide input data
+ * - Reader thread(s): Calling methods to get input
+ * - Writer thread: Calling addInputData to provide input data, addInputAvailableData to provide the number of bytes available
  *
  * Synchronization:
- * - All buffer access is synchronized on the datas collection
- * - Readers wait when no data is available using datas.wait()
- * - Writers notify readers when new data arrives using datas.notifyAll()
+ * - A lock controls access to datas / available as well as internal state
+ * - Readers wait when no data is available using datasReadyCondition.await() / availableReadyCondition.await()
+ * - Writers notify readers when new data arrives datasReadyCondition.signalAll() / availableReadyCondition.signalAll()
  *
  * This implementation is particularly important for:
  * 1. Handling piped input (e.g., cat file | mvnd ...)
@@ -60,26 +66,58 @@
  */
 class DaemonInputStream extends InputStream {
     private final BiConsumer<String, Integer> startReadingFromProject;
+    private final Consumer<String> requestAvailable;
+
+    private final Lock lock = new ReentrantLock();
+    private final Condition datasReadyCondition = lock.newCondition();
+    private final Condition availableReadyCondition = lock.newCondition();
+
     private final LinkedList<byte[]> datas = new LinkedList<>();
+    private Optional<Integer> available = Optional.empty();
+
     private final Charset charset;
     private int pos = -1;
     private String projectReading = null;
     private volatile boolean eof = false;
 
-    DaemonInputStream(BiConsumer<String, Integer> startReadingFromProject) {
+    DaemonInputStream(BiConsumer<String, Integer> startReadingFromProject, Consumer<String> requestAvailable) {
         this.startReadingFromProject = startReadingFromProject;
+        this.requestAvailable = requestAvailable;
         this.charset = Charset.forName(System.getProperty("file.encoding"));
     }
 
     @Override
     public int available() throws IOException {
-        synchronized (datas) {
+        lock.lock();
+        try {
+            int buffered = datas.stream().mapToInt(a -> a.length).sum() - Math.max(pos, 0);
+
+            if (eof) {
+                return buffered;
+            }
+
             String projectId = ProjectBuildLogAppender.getProjectId();
-            if (!eof && !Objects.equals(projectId, projectReading)) {
+
+            if (!Objects.equals(projectId, projectReading)) {
                 projectReading = projectId;
-                startReadingFromProject.accept(projectId, 1);
             }
-            return datas.stream().mapToInt(a -> a.length).sum() - Math.max(pos, 0);
+
+            requestAvailable.accept(projectId);
+
+            available = Optional.empty();
+
+            try {
+                long remaining = TimeUnit.SECONDS.toNanos(1);
+                while (available.isEmpty() && remaining > 0) {
+                    remaining = availableReadyCondition.awaitNanos(remaining);
+                }
+            } catch (InterruptedException e) {
+                throw new InterruptedIOException("Interrupted");
+            }
+
+            return buffered + available.orElse(0);
+        } finally {
+            lock.unlock();
         }
     }
 
@@ -95,7 +133,8 @@ public int read() throws IOException {
 
     @Override
     public int read(byte[] b, int off, int len) throws IOException {
-        synchronized (datas) {
+        lock.lock();
+        try {
             if (eof && datas.isEmpty()) {
                 return -1; // Return EOF if we've reached the end and no more data
             }
@@ -115,7 +154,7 @@ public int read(byte[] b, int off, int len) throws IOException {
                     // Always notify we need input when waiting for data
                     startReadingFromProject.accept(projectReading, len - read);
                     try {
-                        datas.wait();
+                        datasReadyCondition.await();
                     } catch (InterruptedException e) {
                         throw new InterruptedIOException("Interrupted");
                     }
@@ -134,17 +173,32 @@ public int read(byte[] b, int off, int len) throws IOException {
                 b[off + read++] = curData[pos++];
             }
             return read;
+        } finally {
+            lock.unlock();
         }
     }
 
     public void addInputData(String data) {
-        synchronized (datas) {
+        lock.lock();
+        try {
             if (data == null) {
                 eof = true;
             } else {
                 datas.add(data.getBytes(charset));
             }
-            datas.notifyAll();
+            datasReadyCondition.signalAll();
+        } finally {
+            lock.unlock();
+        }
+    }
+
+    public void addInputAvailableData(int bytesAvailable) {
+        lock.lock();
+        try {
+            available = Optional.of(bytesAvailable);
+            availableReadyCondition.signalAll();
+        } finally {
+            lock.unlock();
         }
     }
 }

daemon/src/main/java/org/mvndaemon/mvnd/daemon/Server.java

@@ -509,7 +509,8 @@ private void handle(DaemonConnection connection, BuildRequest buildRequest) {
         final BlockingQueue<Message> recvQueue = new LinkedBlockingDeque<>();
         final BuildEventListener buildEventListener = new ClientDispatcher(sendQueue);
         final DaemonInputStream daemonInputStream = new DaemonInputStream(
-                (projectId, bytesToRead) -> sendQueue.add(Message.requestInput(projectId, bytesToRead)));
+                (projectId, bytesToRead) -> sendQueue.add(Message.requestInput(projectId, bytesToRead)),
+                (projectId) -> sendQueue.add(Message.requestInputAvailable(projectId)));
         InputStream in = System.in;
         try {
             System.setIn(daemonInputStream);
@@ -561,6 +562,9 @@ private void handle(DaemonConnection connection, BuildRequest buildRequest) {
                             return;
                         } else if (message instanceof Message.InputData) {
                             daemonInputStream.addInputData(((Message.InputData) message).getData());
+                        } else if (message instanceof Message.InputAvailableData) {
+                            daemonInputStream.addInputAvailableData(
+                                    ((Message.InputAvailableData) message).getBytesAvailable());
                         } else {
                             synchronized (recvQueue) {
                                 recvQueue.put(message);