Dev mode captures console input destined to the application
### Describe the bug

Running an interactive CLI (Quarkus command mode) in Quarkus dev mode (`mvn quarkus:dev`), Quarkus dev mode partially captures the keystrokes.


### Expected behavior

I would expect the dev mode to capture the keystrokes only if the application didn't consume them before. Or if this is not possible, then I would expect a way to configure a modifier key to access the dev mode shortcuts, and the application to get everything when the modifier key is not pressed.

### Actual behavior

The pattern I found is that:
* for an input that is part of the dev mode shortcuts (eg `h`): the keystroke is captured by the dev mode and doesn't go to the application
* for an input that is not part of those shortcuts (eg `m`): the keystroke goes once to the application and is lost once (so if I type `m` twice, it appears only once for the application).

### How to Reproduce?

Please find a reproducer in https://github.com/PierreBtz/quarkus-termio-bug.

Steps to reproduce:

* Build the project (`mvn package`).
* Verify that it works properly with the uber-jar, you have two commands: 
`java -jar target/quarkus-textio-bug-1.0.0-SNAPSHOT-runner.jar textio` and
  `java -jar target/quarkus-textio-bug-1.0.0-SNAPSHOT-runner.jar reader` reading an input from Console and displaying it.
One command uses a basic `InputStreamReader`, the other [TextIO](https://text-io.beryx.org/releases/latest/) (which is
how I noticed the problem initially...).
* You can also verify it works with a native executable.
* Run `mvn quarkus:dev -Dquarkus.args="textio"` and observe how part of the inputs are captured by the Quarkus dev mode.

### Output of `uname -a` or `ver`

Darwin Pierres-MacBook-Pro-2.local 21.2.0 Darwin Kernel Version 21.2.0: Sun Nov 28 20:28:41 PST 2021; root:xnu-8019.61.5~1/RELEASE_ARM64_T6000 x86_64

### Output of `java -version`

openjdk version "11.0.13" 2021-10-19 LTS OpenJDK Runtime Environment Zulu11.52+13-CA (build 11.0.13+8-LTS) OpenJDK 64-Bit Server VM Zulu11.52+13-CA (build 11.0.13+8-LTS, mixed mode)

### GraalVM version (if different from Java)

gu --version
GraalVM Updater 21.3.0

### Quarkus version or git rev

2.6.3.Final

### Build tool (ie. output of `mvnw --version` or `gradlew --version`)

Apache Maven 3.8.4 (9b656c72d54e5bacbed989b64718c159fe39b537)

### Additional information

_No response_
diff --git a/core/deployment/src/main/java/io/quarkus/deployment/console/AeshConsole.java b/core/deployment/src/main/java/io/quarkus/deployment/console/AeshConsole.java
index 3eb446e39c0..b7e144bbbdd 100644
--- a/core/deployment/src/main/java/io/quarkus/deployment/console/AeshConsole.java
+++ b/core/deployment/src/main/java/io/quarkus/deployment/console/AeshConsole.java
@@ -86,6 +86,7 @@ public void run() {
             }
         }, "Console Shutdown Hook"));
         prompt = registerStatusLine(0);
+
     }
 
     private void updatePromptOnChange(StringBuilder buffer, int newLines) {
@@ -222,6 +223,26 @@ public void run() {
             });
             // Keyboard handling
             conn.setStdinHandler(keys -> {
+
+                QuarkusConsole.StateChangeInputStream redirectIn = QuarkusConsole.REDIRECT_IN;
+                //see if the users application wants to read the keystrokes:
+                int pos = 0;
+                while (pos < keys.length) {
+                    if (!redirectIn.acceptInput(keys[pos])) {
+                        break;
+                    }
+                    ++pos;
+                }
+                if (pos > 0) {
+                    if (pos == keys.length) {
+                        return;
+                    }
+                    //the app only consumed some keys
+                    //stick the rest in a new array
+                    int[] newKeys = new int[keys.length - pos];
+                    System.arraycopy(keys, pos, newKeys, 0, newKeys.length);
+                    keys = newKeys;
+                }
                 try {
                     if (delegateConnection != null) {
                         //console mode
diff --git a/core/deployment/src/main/java/io/quarkus/deployment/console/ConsoleHelper.java b/core/deployment/src/main/java/io/quarkus/deployment/console/ConsoleHelper.java
index b9710dc97a4..06c2e574642 100644
--- a/core/deployment/src/main/java/io/quarkus/deployment/console/ConsoleHelper.java
+++ b/core/deployment/src/main/java/io/quarkus/deployment/console/ConsoleHelper.java
@@ -47,8 +47,11 @@ public void accept(Connection connection) {
                         connection.setStdinHandler(new Consumer<int[]>() {
                             @Override
                             public void accept(int[] ints) {
+                                QuarkusConsole.StateChangeInputStream redirectIn = QuarkusConsole.REDIRECT_IN;
                                 for (int i : ints) {
-                                    queue.add(i);
+                                    if (redirectIn != null && !redirectIn.acceptInput(i)) {
+                                        queue.add(i);
+                                    }
                                 }
                             }
                         });
diff --git a/core/devmode-spi/src/main/java/io/quarkus/dev/console/QuarkusConsole.java b/core/devmode-spi/src/main/java/io/quarkus/dev/console/QuarkusConsole.java
index bc6b3766e88..faca9f2dc0e 100644
--- a/core/devmode-spi/src/main/java/io/quarkus/dev/console/QuarkusConsole.java
+++ b/core/devmode-spi/src/main/java/io/quarkus/dev/console/QuarkusConsole.java
@@ -1,9 +1,13 @@
 package io.quarkus.dev.console;
 
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InterruptedIOException;
 import java.io.PrintStream;
 import java.util.List;
 import java.util.Locale;
 import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.LinkedBlockingDeque;
 import java.util.function.BiPredicate;
 import java.util.function.Consumer;
 
@@ -52,9 +56,12 @@ public abstract class QuarkusConsole {
 
     public final static PrintStream ORIGINAL_OUT = System.out;
     public final static PrintStream ORIGINAL_ERR = System.err;
+    public final static InputStream ORIGINAL_IN = System.in;
 
     public static PrintStream REDIRECT_OUT = null;
     public static PrintStream REDIRECT_ERR = null;
+    public static StateChangeInputStream REDIRECT_IN;
+    protected volatile boolean userReadInProgress;
 
     public synchronized static void installRedirects() {
         if (redirectsInstalled) {
@@ -67,8 +74,10 @@ public synchronized static void installRedirects() {
         QuarkusConsole.INSTANCE.isInputSupported();
         REDIRECT_OUT = new RedirectPrintStream(false);
         REDIRECT_ERR = new RedirectPrintStream(true);
+        REDIRECT_IN = new StateChangeInputStream();
         System.setOut(REDIRECT_OUT);
         System.setErr(REDIRECT_ERR);
+        System.setIn(REDIRECT_IN);
     }
 
     public synchronized static void uninstallRedirects() {
@@ -86,8 +95,10 @@ public synchronized static void uninstallRedirects() {
             REDIRECT_ERR.close();
             REDIRECT_ERR = null;
         }
+        REDIRECT_IN = null;
         System.setOut(ORIGINAL_OUT);
         System.setErr(ORIGINAL_ERR);
+        System.setIn(ORIGINAL_IN);
 
         redirectsInstalled = false;
     }
@@ -176,4 +187,69 @@ public boolean isAnsiSupported() {
         return false;
     }
 
+    protected void userReadStart() {
+
+    }
+
+    protected void userReadStop() {
+
+    }
+
+    public static class StateChangeInputStream extends InputStream {
+
+        private final LinkedBlockingDeque<Integer> queue = new LinkedBlockingDeque<>();
+
+        private volatile boolean reading;
+
+        public synchronized boolean acceptInput(int input) {
+            if (reading) {
+                queue.add(input);
+                notifyAll();
+                return true;
+            }
+            return false;
+        }
+
+        @Override
+        public synchronized int read() throws IOException {
+            reading = true;
+            try {
+                while (queue.isEmpty()) {
+                    try {
+                        wait();
+                    } catch (InterruptedException e) {
+                        throw new InterruptedIOException();
+                    }
+                }
+                return queue.pollFirst();
+            } finally {
+                reading = false;
+            }
+        }
+
+        @Override
+        public synchronized int read(byte[] b, int off, int len) throws IOException {
+            reading = true;
+            int read = 0;
+            try {
+                while (read < len) {
+                    while (queue.isEmpty()) {
+                        try {
+                            wait();
+                        } catch (InterruptedException e) {
+                            throw new InterruptedIOException();
+                        }
+                    }
+                    byte byteValue = queue.poll().byteValue();
+                    b[read++] = byteValue;
+                    if (byteValue == '\\n' || byteValue == '\\r') {
+                        return read;
+                    }
+                }
+                return read;
+            } finally {
+                reading = false;
+            }
+        }
+    }
 }
['core/devmode-spi/src/main/java/io/quarkus/dev/console/QuarkusConsole.java', 'core/deployment/src/main/java/io/quarkus/deployment/console/AeshConsole.java', 'core/deployment/src/main/java/io/quarkus/deployment/console/ConsoleHelper.java']