diff --git a/common/src/main/java/org/mvndaemon/mvnd/common/Message.java b/common/src/main/java/org/mvndaemon/mvnd/common/Message.java index 2878f3e30..790e45097 100644 --- a/common/src/main/java/org/mvndaemon/mvnd/common/Message.java +++ b/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); } diff --git a/common/src/main/java/org/mvndaemon/mvnd/common/logging/TerminalInputHandler.java b/common/src/main/java/org/mvndaemon/mvnd/common/logging/TerminalInputHandler.java index c4071cbef..c5f0e5850 100644 --- a/common/src/main/java/org/mvndaemon/mvnd/common/logging/TerminalInputHandler.java +++ b/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)); diff --git a/common/src/main/java/org/mvndaemon/mvnd/common/logging/TerminalOutput.java b/common/src/main/java/org/mvndaemon/mvnd/common/logging/TerminalOutput.java index ea4164982..1c0b9a28b 100644 --- a/common/src/main/java/org/mvndaemon/mvnd/common/logging/TerminalOutput.java +++ b/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); } diff --git a/daemon/src/main/java/org/mvndaemon/mvnd/daemon/DaemonInputStream.java b/daemon/src/main/java/org/mvndaemon/mvnd/daemon/DaemonInputStream.java index 6843a9106..e4c14bb2f 100644 --- a/daemon/src/main/java/org/mvndaemon/mvnd/daemon/DaemonInputStream.java +++ b/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 startReadingFromProject; + private final Consumer requestAvailable; + + private final Lock lock = new ReentrantLock(); + private final Condition datasReadyCondition = lock.newCondition(); + private final Condition availableReadyCondition = lock.newCondition(); + private final LinkedList datas = new LinkedList<>(); + private Optional available = Optional.empty(); + private final Charset charset; private int pos = -1; private String projectReading = null; private volatile boolean eof = false; - DaemonInputStream(BiConsumer startReadingFromProject) { + DaemonInputStream(BiConsumer startReadingFromProject, Consumer 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(); } } } diff --git a/daemon/src/main/java/org/mvndaemon/mvnd/daemon/Server.java b/daemon/src/main/java/org/mvndaemon/mvnd/daemon/Server.java index 87e18a156..3e386692c 100644 --- a/daemon/src/main/java/org/mvndaemon/mvnd/daemon/Server.java +++ b/daemon/src/main/java/org/mvndaemon/mvnd/daemon/Server.java @@ -509,7 +509,8 @@ private void handle(DaemonConnection connection, BuildRequest buildRequest) { final BlockingQueue 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); diff --git a/integration-tests/src/test/java/org/mvndaemon/mvnd/it/InputStreamNativeIT.java b/integration-tests/src/test/java/org/mvndaemon/mvnd/it/InputStreamNativeIT.java index 78df741eb..e3aa0c560 100644 --- a/integration-tests/src/test/java/org/mvndaemon/mvnd/it/InputStreamNativeIT.java +++ b/integration-tests/src/test/java/org/mvndaemon/mvnd/it/InputStreamNativeIT.java @@ -61,4 +61,37 @@ public void accept(Message message) { client.execute(output, "org.mvndaemon.mvnd.test.input-stream:echo-maven-plugin:echo"); } + + @Test + void installPluginAndCount() throws IOException, InterruptedException { + final TestClientOutput output = new TestClientOutput() { + boolean sentInput = false; + String data = "0123456789"; + + @Override + public void accept(Message message) { + if (message instanceof Message.RequestInput) { + if (!sentInput) { + daemonDispatch.accept(Message.inputResponse(data)); + sentInput = true; + } else { + daemonDispatch.accept(Message.inputEof()); + } + } else if (message instanceof Message.RequestInputAvailable) { + daemonDispatch.accept(Message.inputAvailableData(data.length())); + } + + if (!(message instanceof Message.TransferEvent)) { + super.accept(message); + } + } + }; + client.execute(output, "install").assertSuccess(); + + client.execute(output, "org.mvndaemon.mvnd.test.input-stream:echo-maven-plugin:countbytes") + .assertSuccess(); + + output.assertContainsMatchingSubsequence( + "Saw 10 bytes available from stdin. Actually read 10 bytes from stdin."); + } } diff --git a/integration-tests/src/test/projects/input-stream/src/main/java/org/apache/maven/its/CountBytesMojo.java b/integration-tests/src/test/projects/input-stream/src/main/java/org/apache/maven/its/CountBytesMojo.java new file mode 100644 index 000000000..648f14863 --- /dev/null +++ b/integration-tests/src/test/projects/input-stream/src/main/java/org/apache/maven/its/CountBytesMojo.java @@ -0,0 +1,58 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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. + */ +package org.apache.maven.its; + +import org.apache.maven.plugin.AbstractMojo; +import org.apache.maven.plugin.MojoExecutionException; + +import org.apache.maven.plugins.annotations.Mojo; +import org.apache.maven.plugins.annotations.Parameter; + +import java.io.BufferedReader; +import java.io.File; +import java.io.FileWriter; +import java.io.IOException; +import java.util.Scanner; + +/** + * Goal which counts the number of bytes available on stdin. + */ +@Mojo( name = "countbytes", requiresProject = false ) +public class CountBytesMojo + extends AbstractMojo +{ + + public void execute() + throws MojoExecutionException + { + getLog().info("Reading from standard input. Type 'exit' to stop."); + + int available = 0; + byte[] bytes = new byte[] {}; + + try { + available = System.in.available(); + bytes = System.in.readAllBytes(); + } catch (Exception ex) { + getLog().error("Failed to read from stdin", ex); + } + + System.out.println("Saw " + available + " bytes available from stdin. Actually read " + bytes.length + " bytes from stdin."); + } +}