Add async process start

process/src/main/java/io/smallrye/common/process/PipelineBuilder.java

@@ -11,6 +11,7 @@
 import java.nio.file.StandardOpenOption;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.CompletableFuture;
 import java.util.function.Consumer;
 import java.util.function.IntPredicate;
 
@@ -92,6 +93,13 @@ public sealed interface PipelineBuilder<O> permits PipelineBuilder.Error, Pipeli
      */
     O run();
 
+    /**
+     * Run the process or pipeline asynchronously.
+     *
+     * @return the asynchronous result (not {@code null} but may yield {@code null})
+     */
+    CompletableFuture<O> runAsync();
+
     /**
      * Add a failure exit code checker.
      * The checker returns {@code true} if the process execution should fail for this exit code.

process/src/main/java/io/smallrye/common/process/PipelineRunner.java

@@ -59,6 +59,7 @@ class PipelineRunner<O> {
     private Thread waitForThread;
     private ProcessHandlerException exitCheckerProblem;
     private AbnormalExitException abnormalExit;
+    Thread asyncThread;
     Process process;
     private Gatherer errorGatherer;
     private Gatherer outputGatherer;
@@ -498,6 +499,7 @@ void startThreads() {
         errorExtraThreads.forEach(Thread::start);
         startThread(whileRunningThread);
         startThread(waitForThread);
+        startThread(asyncThread);
         if (prev != null) {
             prev.startThreads();
         }
@@ -542,6 +544,7 @@ void unpark() {
         errorExtraThreads.forEach(LockSupport::unpark);
         LockSupport.unpark(whileRunningThread);
         LockSupport.unpark(waitForThread);
+        LockSupport.unpark(asyncThread);
         if (prev != null) {
             prev.unpark();
         }

process/src/main/java/io/smallrye/common/process/ProcessBuilderImpl.java

@@ -18,6 +18,7 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.CompletableFuture;
 import java.util.function.Consumer;
 import java.util.function.IntPredicate;
 
@@ -189,6 +190,12 @@ public O run() {
         return makeRunner().run();
     }
 
+    public CompletableFuture<O> runAsync() {
+        check();
+        locked = true;
+        return makeRunner().runAsync();
+    }
+
     private ProcessRunner<O> makeRunner() {
         return new ProcessRunner<O>(this, prev == null ? null : prev.makePipelineRunner());
     }
@@ -242,6 +249,10 @@ public O run() throws AbnormalExitException {
             return ProcessBuilderImpl.this.run();
         }
 
+        public CompletableFuture<O> runAsync() {
+            return ProcessBuilderImpl.this.runAsync();
+        }
+
         public ProcessBuilder<O> exitCodeChecker(final IntPredicate checker) {
             return ProcessBuilderImpl.this.exitCodeChecker(checker);
         }

process/src/main/java/io/smallrye/common/process/ProcessRunner.java

@@ -10,6 +10,7 @@
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CopyOnWriteArraySet;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.locks.LockSupport;
@@ -59,17 +60,42 @@ void taskComplete() {
         }
     }
 
-    O run() {
-        ThreadFactory tf;
-        // todo: if (Thread.currentThread().isVirtual()) { ... }
-        tf = task -> new Thread(() -> {
-            log.trace("Starting process thread");
-            try {
-                task.run();
-            } finally {
-                log.trace("Ending process thread");
+    CompletableFuture<O> runAsync() {
+        CompletableFuture<O> cf = new CompletableFuture<>();
+        ThreadFactory tf = threadFactory();
+        asyncThread = tf.newThread(() -> {
+            if (awaitOk()) {
+                Thread shutdownHook = registerHook();
+                await();
+                try {
+                    cf.complete(complete());
+                } catch (Throwable t) {
+                    cf.completeExceptionally(t);
+                } finally {
+                    Runtime.getRuntime().removeShutdownHook(shutdownHook);
+                }
             }
         });
+        if (asyncThread == null) {
+            throw new PipelineExecutionException("Failed to start process thread(s)", noThread(tf));
+        }
+        asyncThread.setName("process-async-handler");
+        initialize(tf);
+        return cf;
+    }
+
+    O run() {
+        initialize(threadFactory());
+        Thread shutdownHook = registerHook();
+        try {
+            await();
+        } finally {
+            Runtime.getRuntime().removeShutdownHook(shutdownHook);
+        }
+        return complete();
+    }
+
+    private void initialize(final ThreadFactory tf) {
         int taskCnt;
         try {
             // start up input, output, error, and wait-for threads
@@ -102,8 +128,23 @@ O run() {
         }
         status = STATUS_STARTED;
         unpark();
+    }
+
+    private void await() {
+        int cnt = taskCount;
+        if (cnt != 0) {
+            waiters.add(Thread.currentThread());
+        }
+        do {
+            Thread.interrupted();
+            LockSupport.park(this);
+            cnt = taskCount;
+        } while (cnt != 0);
+    }
+
+    private Thread registerHook() {
         Thread shutdownHook = new Thread(() -> {
-            int cnt = taskCnt;
+            int cnt = taskCount;
             if (cnt != 0) {
                 waiters.add(Thread.currentThread());
                 do {
@@ -116,19 +157,10 @@ O run() {
             }
         }, "pipeline-shutdown");
         Runtime.getRuntime().addShutdownHook(shutdownHook);
-        try {
-            int cnt = taskCnt;
-            if (cnt != 0) {
-                waiters.add(Thread.currentThread());
-            }
-            do {
-                Thread.interrupted();
-                LockSupport.park(this);
-                cnt = taskCount;
-            } while (cnt != 0);
-        } finally {
-            Runtime.getRuntime().removeShutdownHook(shutdownHook);
-        }
+        return shutdownHook;
+    }
+
+    private O complete() {
         List<ProcessExecutionException> problems = new ArrayList<>(4);
         collectProblems(problems);
         return switch (problems.size()) {
@@ -142,6 +174,18 @@ O run() {
         };
     }
 
+    private static ThreadFactory threadFactory() {
+        // todo: if (Thread.currentThread().isVirtual()) { ... }
+        return task -> new Thread(() -> {
+            log.trace("Starting process thread");
+            try {
+                task.run();
+            } finally {
+                log.trace("Ending process thread");
+            }
+        });
+    }
+
     boolean awaitOk() {
         while (status == STATUS_WAITING) {
             LockSupport.park(this);

process/src/test/java/io/smallrye/common/process/TestBasicExecution.java

@@ -9,6 +9,8 @@
 import java.net.URL;
 import java.util.ArrayDeque;
 import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
 import java.util.stream.Stream;
 
 import org.junit.jupiter.api.Test;
@@ -31,6 +33,19 @@ public void testSimpleCat() throws Exception {
         assertEquals(strings, result);
     }
 
+    @Test
+    public void testSimpleCatAsync() throws Exception {
+        List<String> strings = List.of("Hello", "world", "foo", "bar");
+        CompletableFuture<List<String>> future = ProcessBuilder.newBuilder(ProcessUtil.pathOfJava(), findHelper(Cat.class))
+                .input()
+                .fromStrings(strings)
+                .output()
+                .toStringList(10, 1024)
+                .runAsync();
+        List<String> result = future.get();
+        assertEquals(strings, result);
+    }
+
     @Test
     public void testSimpleCatWithTee() throws Exception {
         List<String> strings = List.of("Hello", "world", "foo", "bar");
@@ -110,6 +125,23 @@ public void testFailure() throws Exception {
         }
     }
 
+    @Test
+    public void testFailureAsync() throws Throwable {
+        CompletableFuture<Void> cf = ProcessBuilder.newBuilder(ProcessUtil.pathOfJava(), findHelper(Errorifier.class, "7"))
+                .runAsync();
+        try {
+            try {
+                cf.get();
+                fail("Expected specific exception");
+            } catch (ExecutionException e) {
+                throw e.getCause();
+            }
+            fail("Expected specific exception");
+        } catch (AbnormalExitException e) {
+            assertEquals(7, e.exitCode());
+        }
+    }
+
     @Test
     public void testFailingPipeline() throws Exception {
         try {