ProcessUtil { public static <O, E> NoticeableFuture<Map.Entry<O, E>> execute( OutputProcessor<O, E> outputProcessor, List<String> arguments) throws ProcessException { if (outputProcessor == null) { throw new NullPointerException("Output processor is null"); } if (arguments == null) { throw new NullPointerException("Arguments is null"); } ProcessBuilder processBuilder = new ProcessBuilder(arguments); String threadNamePrefix = _buildThreadNamePrefix(arguments); try { Process process = processBuilder.start(); NoticeableFuture<O> stdOutNoticeableFuture = _submit( threadNamePrefix.concat("StdOut"), new ProcessStdOutCallable<>(outputProcessor, process)); NoticeableFuture<E> stdErrNoticeableFuture = _submit( threadNamePrefix.concat("StdErr"), new ProcessStdErrCallable<>(outputProcessor, process)); return _wrapNoticeableFuture( stdOutNoticeableFuture, stdErrNoticeableFuture, process); } catch (IOException ioException) { throw new ProcessException(ioException); } }  static NoticeableFuture<Map.Entry<O, E>> execute(
			OutputProcessor<O, E> outputProcessor, List<String> arguments); static NoticeableFuture<Map.Entry<O, E>> execute(
			OutputProcessor<O, E> outputProcessor, String... arguments);  }
@Test public void testEcho() throws Exception { List<Map.Entry<Boolean, String>> logRecords = new CopyOnWriteArrayList<>(); Future<Map.Entry<Void, Void>> loggingFuture = ProcessUtil.execute( new LoggingOutputProcessor( (stdErr, line) -> logRecords.add( new AbstractMap.SimpleImmutableEntry<>(stdErr, line))), _buildArguments(Echo.class, "2")); loggingFuture.get(); loggingFuture.cancel(true); List<String> messageRecords = new ArrayList<>(); for (Map.Entry<Boolean, String> logRecord : logRecords) { messageRecords.add(logRecord.getValue()); } Assert.assertTrue( messageRecords.toString(), messageRecords.contains(Echo.buildMessage(false, 0))); Assert.assertTrue( messageRecords.toString(), messageRecords.contains(Echo.buildMessage(false, 1))); Assert.assertTrue( messageRecords.toString(), messageRecords.contains(Echo.buildMessage(true, 0))); Assert.assertTrue( messageRecords.toString(), messageRecords.contains(Echo.buildMessage(true, 1))); Future<Map.Entry<byte[], byte[]>> collectorFuture = ProcessUtil.execute( CollectorOutputProcessor.INSTANCE, _buildArguments(Echo.class, "2")); Map.Entry<byte[], byte[]> objectValuePair = collectorFuture.get(); collectorFuture.cancel(true); Assert.assertEquals( StringBundler.concat( Echo.buildMessage(true, 0), "\n", Echo.buildMessage(true, 1), "\n"), new String(objectValuePair.getKey())); Assert.assertEquals( StringBundler.concat( Echo.buildMessage(false, 0), "\n", Echo.buildMessage(false, 1), "\n"), new String(objectValuePair.getValue())); }
@Test public void testErrorExit() throws Exception { Future<?> future = ProcessUtil.execute( ConsumerOutputProcessor.INSTANCE, _buildArguments(ErrorExit.class)); try { future.get(); Assert.fail(); } catch (ExecutionException executionException) { Throwable throwable = executionException.getCause(); Assert.assertSame( TerminationProcessException.class, throwable.getClass()); Assert.assertEquals( "Subprocess terminated with exit code " + ErrorExit.EXIT_CODE, throwable.getMessage()); TerminationProcessException terminationProcessException = (TerminationProcessException)throwable; Assert.assertEquals( ErrorExit.EXIT_CODE, terminationProcessException.getExitCode()); } }
@Test public void testErrorOutputProcessor() throws Exception { String[] arguments = _buildArguments(Echo.class, "1"); Future<?> future = ProcessUtil.execute( new ErrorStderrOutputProcessor(), arguments); try { future.get(); Assert.fail(); } catch (ExecutionException executionException) { Throwable throwable = executionException.getCause(); Assert.assertEquals(ProcessException.class, throwable.getClass()); Assert.assertEquals( ErrorStderrOutputProcessor.class.getName(), throwable.getMessage()); } future = ProcessUtil.execute( new ErrorStdoutOutputProcessor(), arguments); try { future.get(); Assert.fail(); } catch (ExecutionException executionException) { Throwable throwable = executionException.getCause(); Assert.assertEquals(ProcessException.class, throwable.getClass()); Assert.assertEquals( ErrorStdoutOutputProcessor.class.getName(), throwable.getMessage()); } }
@Test public void testFuture() throws Exception { String[] arguments = _buildArguments(Pause.class); Future<?> future = ProcessUtil.execute( ConsumerOutputProcessor.INSTANCE, arguments); Assert.assertFalse(future.isCancelled()); Assert.assertFalse(future.isDone()); try { future.get(1, TimeUnit.SECONDS); Assert.fail(); } catch (TimeoutException timeoutException) { } future.cancel(true); future.cancel(true); future = ProcessUtil.execute( new ConsumerOutputProcessor() { @Override public Void processStdErr(InputStream stdOutInputStream) { return null; } }, arguments); Assert.assertFalse(future.isCancelled()); Assert.assertFalse(future.isDone()); try { future.get(1, TimeUnit.SECONDS); Assert.fail(); } catch (TimeoutException timeoutException) { } future.cancel(true); future = ProcessUtil.execute( ConsumerOutputProcessor.INSTANCE, _buildArguments(Echo.class, "0")); future.get(1, TimeUnit.MINUTES); }
@Test public void testInterruptPause() throws Exception { String threadNamePrefix = ReflectionTestUtil.invoke( ProcessUtil.class, "_buildThreadNamePrefix", new Class<?>[] {List.class}, Arrays.asList(_buildArguments(Pause.class))); String threadName = threadNamePrefix + "StdOut"; SyncThrowableThread<Void> syncThrowableThread = new SyncThrowableThread<>( () -> { while (true) { for (Thread thread : ThreadTestUtil.getThreads()) { if ((thread != null) && threadName.equals(thread.getName())) { thread.interrupt(); return null; } } } }); syncThrowableThread.start(); Future<?> future = ProcessUtil.execute( new OutputProcessor<Void, Void>() { @Override public Void processStdErr(InputStream stdErrInputStream) { return null; } @Override public Void processStdOut(InputStream stdOutInputStream) { return null; } }, _buildArguments(Pause.class)); try { future.get(); Assert.fail(); } catch (ExecutionException executionException) { Throwable throwable = executionException.getCause(); Assert.assertSame(ProcessException.class, throwable.getClass()); Assert.assertEquals( "Forcibly killed subprocess on interruption", throwable.getMessage()); } finally { syncThrowableThread.sync(); } }
@Test public void testWrongArguments() throws ProcessException { try { ProcessUtil.execute(null, (List<String>)null); Assert.fail(); } catch (NullPointerException nullPointerException) { Assert.assertEquals( "Output processor is null", nullPointerException.getMessage()); } try { ProcessUtil.execute( ConsumerOutputProcessor.INSTANCE, (List<String>)null); Assert.fail(); } catch (NullPointerException nullPointerException) { Assert.assertEquals( "Arguments is null", nullPointerException.getMessage()); } try { ProcessUtil.execute( ConsumerOutputProcessor.INSTANCE, "commandNotExist"); Assert.fail(); } catch (ProcessException processException) { Throwable throwable = processException.getCause(); Assert.assertSame(IOException.class, throwable.getClass()); } }