LocalProcessExecutor implements ProcessExecutor { @Override public <T extends Serializable> ProcessChannel<T> execute( ProcessConfig processConfig, ProcessCallable<T> processCallable) throws ProcessException { try { List<String> arguments = processConfig.getArguments(); List<String> commands = new ArrayList<>(arguments.size() + 4); commands.add(processConfig.getJavaExecutable()); commands.add("-cp"); commands.add(processConfig.getBootstrapClassPath()); commands.addAll(arguments); commands.add(LocalProcessLauncher.class.getName()); ProcessBuilder processBuilder = new ProcessBuilder(commands); Map<String, String> environment = processConfig.getEnvironment(); if (environment != null) { Map<String, String> currentEnvironment = processBuilder.environment(); currentEnvironment.clear(); currentEnvironment.putAll(environment); } final Process process = processBuilder.start(); ObjectOutputStream bootstrapObjectOutputStream = new ObjectOutputStream(process.getOutputStream()); bootstrapObjectOutputStream.writeObject(processCallable.toString()); bootstrapObjectOutputStream.writeObject( processConfig.getRuntimeClassPath()); ObjectOutputStream objectOutputStream = new ObjectOutputStream( bootstrapObjectOutputStream); objectOutputStream.writeObject(processCallable); objectOutputStream.flush(); AsyncBroker<Long, Serializable> asyncBroker = new AsyncBroker<>(); SubprocessReactor<T> subprocessReactor = new SubprocessReactor<>( process, processConfig.getProcessLogConsumer(), processConfig.getReactClassLoader(), asyncBroker); NoticeableFuture<T> noticeableFuture = _submit( _buildThreadName(processCallable, arguments), subprocessReactor); noticeableFuture.addFutureListener( future -> { if (future.isCancelled()) { process.destroy(); } }); return new LocalProcessChannel<>( noticeableFuture, objectOutputStream, asyncBroker); } catch (IOException ioException) { throw new ProcessException(ioException); } }  @Override ProcessChannel<T> execute(
			ProcessConfig processConfig, ProcessCallable<T> processCallable);  }
@Test public void testProcessCallableWithException() throws Exception { ProcessChannel<Serializable> processChannel = _localProcessExecutor.execute( _createJPDAProcessConfig(_JPDA_OPTIONS1), () -> { throw new ProcessException("ROOT ProcessException"); }); NoticeableFuture<Serializable> noticeableFuture = processChannel.getProcessNoticeableFuture(); try { noticeableFuture.get(); Assert.fail(); } catch (ExecutionException executionException) { Throwable throwable = executionException.getCause(); Assert.assertSame(ProcessException.class, throwable.getClass()); Assert.assertEquals( "ROOT ProcessException", throwable.getMessage()); } processChannel = _localProcessExecutor.execute( _createJPDAProcessConfig(_JPDA_OPTIONS1), () -> { throw new NullPointerException("ROOT NullPointerException"); }); noticeableFuture = processChannel.getProcessNoticeableFuture(); try { noticeableFuture.get(); Assert.fail(); } catch (ExecutionException executionException) { Throwable throwable = executionException.getCause(); Assert.assertSame(ProcessException.class, throwable.getClass()); throwable = throwable.getCause(); Assert.assertSame(NullPointerException.class, throwable.getClass()); Assert.assertEquals( "ROOT NullPointerException", throwable.getMessage()); } }
@Test public void testProcessConfigBuilderEnvironment() throws Exception { ProcessConfig.Builder builder = new ProcessConfig.Builder(); builder.setArguments(_createArguments(_JPDA_OPTIONS1)); builder.setBootstrapClassPath(System.getProperty("java.class.path")); builder.setReactClassLoader( LocalProcessExecutorTest.class.getClassLoader()); ProcessChannel<HashMap<String, String>> processChannel = _localProcessExecutor.execute( builder.build(), Operations.GET_ENVIRONMENT); Future<HashMap<String, String>> future = processChannel.getProcessNoticeableFuture(); Assert.assertEquals(System.getenv(), future.get()); Map<String, String> environmentMap = new HashMap<>(); environmentMap.put("key1", "value1"); environmentMap.put("key2", "value2"); builder.setEnvironment(environmentMap); processChannel = _localProcessExecutor.execute( builder.build(), Operations.GET_ENVIRONMENT); future = processChannel.getProcessNoticeableFuture(); Map<String, String> actualEnvironmentMap = future.get(); Assert.assertEquals("value1", actualEnvironmentMap.get("key1")); Assert.assertEquals("value2", actualEnvironmentMap.get("key2")); }
@Test public void testProcessConfigBuilderJavaExecutable() throws Exception { try { ProcessConfig.Builder builder = new ProcessConfig.Builder(); builder.setJavaExecutable("javax"); _localProcessExecutor.execute(builder.build(), Operations.SLEEP); Assert.fail(); } catch (ProcessException processException) { Throwable throwable = processException.getCause(); Assert.assertTrue(throwable instanceof IOException); } }
@Test public void testProcessConfigBuilderRuntimeClassPath() throws Exception { ProcessConfig.Builder builder = new ProcessConfig.Builder(); builder.setArguments(_createArguments(_JPDA_OPTIONS1)); char[] largeFileNameChars = new char[10 * 1024 * 1024]; largeFileNameChars[0] = CharPool.SLASH; for (int i = 1; i < largeFileNameChars.length; i++) { largeFileNameChars[i] = (char)('a' + (i % 26)); } String largeFileName = new String(largeFileNameChars); builder.setRuntimeClassPath(largeFileName); ProcessChannel<String> processChannel = _localProcessExecutor.execute( builder.build(), Operations.GET_RUNTIME_CLASS_PATH); Future<String> future = processChannel.getProcessNoticeableFuture(); Assert.assertEquals(largeFileName, future.get()); }
@Test public void testProcessContextAttach() throws Exception { ProcessChannel<String> processChannel = _localProcessExecutor.execute( _createJPDAProcessConfig(_JPDA_OPTIONS1), Operations.asControllable(Operations.SLEEP)); Future<Controller> future = processChannel.write( Operations.GET_CONTROLLER); Controller parentController = future.get(); Assert.assertTrue(parentController.isAlive()); Controller childController = parentController.invoke( Operations.asNewJVM(_JPDA_OPTIONS2, Operations.SLEEP)); Assert.assertTrue(childController.isAlive()); Assert.assertFalse(childController.invoke(Operations.IS_ATTACHED)); Assert.assertEquals("DONE", childController.invoke(Operations.DETACH)); Assert.assertFalse(childController.invoke(Operations.IS_ATTACHED)); Assert.assertTrue( childController.invoke( Operations.attach(ShutdownHooks.TERMINATE_SHUTDOWN_HOOK))); Assert.assertTrue(childController.invoke(Operations.IS_ATTACHED)); Assert.assertFalse( childController.invoke( Operations.attach(ShutdownHooks.TERMINATE_SHUTDOWN_HOOK))); Assert.assertEquals("DONE", childController.invoke(Operations.DETACH)); Assert.assertFalse(childController.invoke(Operations.IS_ATTACHED)); Assert.assertTrue( childController.invoke( Operations.attach(ShutdownHooks.TERMINATE_SHUTDOWN_HOOK))); Assert.assertTrue(childController.invoke(Operations.IS_ATTACHED)); parentController.invoke(Operations.TERMINATE); Assert.assertFalse(parentController.isAlive()); NoticeableFuture<String> noticeableFuture = processChannel.getProcessNoticeableFuture(); Assert.assertEquals("DONE", noticeableFuture.get()); _timeWaitAssertFalse( "The child process is still alive", childController::isAlive, 10, TimeUnit.MINUTES); }
@Test public void testProcessContextAttachWithNullShutdownHook() throws Exception { ProcessChannel<String> processChannel = _localProcessExecutor.execute( _createJPDAProcessConfig(_JPDA_OPTIONS1), Operations.asControllable(Operations.SLEEP)); Future<Controller> future = processChannel.write( Operations.GET_CONTROLLER); Controller parentController = future.get(); Assert.assertTrue(parentController.isAlive()); Controller childController = parentController.invoke( Operations.asNewJVM(_JPDA_OPTIONS2, Operations.SLEEP)); Assert.assertTrue(childController.isAlive()); Assert.assertEquals( "DONE", childController.invoke( () -> { try { LocalProcessLauncher.ProcessContext.attach( "NullShutdownHook", 1, null); return "NULL_SHUTDOWN_HOOK_ACCEPTED"; } catch (IllegalArgumentException illegalArgumentException) { if (!Objects.equals( illegalArgumentException.getMessage(), "Shutdown hook is null")) { return illegalArgumentException.getMessage(); } } return "DONE"; })); Assert.assertFalse(childController.invoke(Operations.IS_ATTACHED)); parentController.invoke(Operations.TERMINATE); Assert.assertFalse(parentController.isAlive()); childController.invoke(Operations.TERMINATE); Assert.assertFalse(childController.isAlive()); }
@Test public void testSpawnProcessWithoutAttach() throws Exception { ProcessChannel<String> processChannel = _localProcessExecutor.execute( _createJPDAProcessConfig(_JPDA_OPTIONS1), Operations.asControllable(Operations.SLEEP)); Future<Controller> future = processChannel.write( Operations.GET_CONTROLLER); Controller parentController = future.get(); Assert.assertTrue(parentController.isAlive()); Controller childController = parentController.invoke( Operations.asNewJVM(_JPDA_OPTIONS2, Operations.SLEEP)); Assert.assertTrue(childController.isAlive()); parentController.invoke(Operations.TERMINATE); Assert.assertFalse(parentController.isAlive()); NoticeableFuture<String> noticeableFuture = processChannel.getProcessNoticeableFuture(); Assert.assertEquals("DONE", noticeableFuture.get()); for (int i = 0; i < 10; i++) { Thread.sleep(100); Assert.assertTrue(childController.isAlive()); } childController.invoke(Operations.TERMINATE); Assert.assertFalse(childController.isAlive()); }
@Test public void testSubprocessReactorAbort() throws Exception { List<ProcessLog> processLogs = new ArrayList<>(); ProcessConfig.Builder builder = new ProcessConfig.Builder(); builder.setArguments(_createArguments(_JPDA_OPTIONS1)); builder.setBootstrapClassPath(System.getProperty("java.class.path")); builder.setProcessLogConsumer( processLog -> { if (processLog.getLevel() == ProcessLog.Level.ERROR) { processLogs.add(processLog); } }); builder.setReactClassLoader(new URLClassLoader(new URL[0], null)); ProcessChannel<Boolean> processChannel = _localProcessExecutor.execute( builder.build(), Operations.IS_ATTACHED); Future<Boolean> future = processChannel.getProcessNoticeableFuture(); try { future.get(); Assert.fail(); } catch (ExecutionException executionException) { Throwable throwable = executionException.getCause(); Assert.assertSame( ClassNotFoundException.class, throwable.getClass()); } Assert.assertEquals(processLogs.toString(), 1, processLogs.size()); ProcessLog processLog = processLogs.get(0); Assert.assertEquals("Abort subprocess piping", processLog.getMessage()); Throwable throwable = processLog.getThrowable(); Assert.assertSame(ClassNotFoundException.class, throwable.getClass()); }
@Test public void testSubprocessReactorCorruptedStream() throws Exception { List<ProcessLog> processLogs = new ArrayList<>(); ProcessChannel<Serializable> processChannel = _localProcessExecutor.execute( _createJPDAProcessConfig( _JPDA_OPTIONS1, processLog -> { if (processLog.getLevel() == ProcessLog.Level.ERROR) { processLogs.add(processLog); } }), Operations.CORRUPTED_STREAM); Future<Serializable> future = processChannel.getProcessNoticeableFuture(); try { future.get(); Assert.fail(); } catch (ExecutionException executionException) { Throwable throwable = executionException.getCause(); Assert.assertTrue(throwable instanceof ProcessException); Assert.assertEquals( "Corrupted object input stream", throwable.getMessage()); throwable = throwable.getCause(); Assert.assertSame( StreamCorruptedException.class, throwable.getClass()); } Assert.assertFalse(future.isCancelled()); Assert.assertTrue(future.isDone()); Assert.assertEquals(processLogs.toString(), 1, processLogs.size()); ProcessLog processLog = processLogs.get(0); String message = processLog.getMessage(); int index = message.lastIndexOf(' '); Assert.assertTrue(index != -1); Assert.assertEquals( "Dumping content of corrupted object input stream to", message.substring(0, index)); File file = new File(message.substring(index + 1)); Assert.assertTrue(file.exists()); file.delete(); Throwable throwable = processLog.getThrowable(); Assert.assertSame(StreamCorruptedException.class, throwable.getClass()); }
@Test public void testSubprocessReactorCrash() throws Exception { ProcessChannel<Serializable> processChannel = _localProcessExecutor.execute( _createJPDAProcessConfig(_JPDA_OPTIONS1), Operations.crashJVM(1)); Future<Serializable> future = processChannel.getProcessNoticeableFuture(); 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 1", throwable.getMessage()); TerminationProcessException terminationProcessException = (TerminationProcessException)throwable; Assert.assertEquals(1, terminationProcessException.getExitCode()); } processChannel = _localProcessExecutor.execute( _createJPDAProcessConfig(_JPDA_OPTIONS1), Operations.crashJVM(0)); future = processChannel.getProcessNoticeableFuture(); try { future.get(); Assert.fail(); } catch (ExecutionException executionException) { Throwable throwable = executionException.getCause(); Assert.assertSame(ProcessException.class, throwable.getClass()); Assert.assertEquals( "Subprocess piping back ended prematurely", throwable.getMessage()); throwable = throwable.getCause(); Assert.assertSame(EOFException.class, throwable.getClass()); } }
@Test public void testSubprocessReactorKillByCancel() throws Exception { List<ProcessLog> processLogs = new ArrayList<>(); ProcessChannel<String> processChannel = _localProcessExecutor.execute( _createJPDAProcessConfig( _JPDA_OPTIONS1, processLog -> { if (processLog.getLevel() == ProcessLog.Level.ERROR) { processLogs.add(processLog); } }), Operations.asControllable(Operations.SLEEP)); Future<Controller> future = processChannel.write( Operations.GET_CONTROLLER); Controller controller = future.get(); Assert.assertTrue(controller.isAlive()); Map<String, Object> attributes = LocalProcessLauncher.ProcessContext.getAttributes(); BlockingQueue<Thread> reactorThreadBlockingQueue = new SynchronousQueue<>(); attributes.put( "reactorThreadBlockingQueue", reactorThreadBlockingQueue); controller.invoke( () -> { try { LocalProcessLauncher.ProcessContext.writeProcessCallable( () -> { Map<String, Object> localAttributes = LocalProcessLauncher.ProcessContext. getAttributes(); BlockingQueue<Thread> localReactorThreadBlockingQueue = (BlockingQueue<Thread>) localAttributes.remove( "reactorThreadBlockingQueue"); try { localReactorThreadBlockingQueue.put( Thread.currentThread()); } catch (InterruptedException interruptedException) { throw new ProcessException( interruptedException); } return null; }); } catch (IOException ioException) { throw new ProcessException(ioException); } return null; }); Thread reactorThread = reactorThreadBlockingQueue.take(); NoticeableFuture<String> noticeableFuture = processChannel.getProcessNoticeableFuture(); noticeableFuture.cancel(false); reactorThread.join(); Assert.assertEquals(processLogs.toString(), 1, processLogs.size()); ProcessLog processLog = processLogs.get(0); Assert.assertEquals("Abort subprocess piping", processLog.getMessage()); Throwable throwable = processLog.getThrowable(); Assert.assertSame(IOException.class, throwable.getClass()); }
@Test public void testSubprocessReactorKillByInterruption() throws Exception { ProcessChannel<String> processChannel = _localProcessExecutor.execute( _createJPDAProcessConfig(_JPDA_OPTIONS1), Operations.asControllable(Operations.SLEEP)); Future<Controller> future = processChannel.write( Operations.GET_CONTROLLER); Controller controller = future.get(); Assert.assertTrue(controller.isAlive()); Map<String, Object> attributes = LocalProcessLauncher.ProcessContext.getAttributes(); BlockingQueue<Thread> reactorThreadBlockingQueue = new SynchronousQueue<>(); attributes.put( "reactorThreadBlockingQueue", reactorThreadBlockingQueue); controller.invoke( () -> { try { LocalProcessLauncher.ProcessContext.writeProcessCallable( () -> { Map<String, Object> localAttributes = LocalProcessLauncher.ProcessContext. getAttributes(); BlockingQueue<Thread> localReactorThreadBlockingQueue = (BlockingQueue<Thread>) localAttributes.remove( "reactorThreadBlockingQueue"); try { localReactorThreadBlockingQueue.put( Thread.currentThread()); } catch (InterruptedException interruptedException) { throw new ProcessException( interruptedException); } return null; }); Object processOutputStream = ReflectionTestUtil.getFieldValue( LocalProcessLauncher.ProcessContext.class, "_processOutputStream"); ReflectionTestUtil.invoke( processOutputStream, "close", new Class<?>[0]); } catch (IOException ioException) { throw new ProcessException(ioException); } return null; }); Thread reactorThread = reactorThreadBlockingQueue.take(); reactorThread.interrupt(); NoticeableFuture<String> noticeableFuture = processChannel.getProcessNoticeableFuture(); try { noticeableFuture.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()); throwable = throwable.getCause(); Assert.assertSame(InterruptedException.class, throwable.getClass()); } }
@Test public void testSubprocessReactorLeadingLog() throws Exception { List<ProcessLog> processLogs = new ArrayList<>(); AtomicReference<ProcessLog.Level> levelReference = new AtomicReference<>(ProcessLog.Level.WARN); Consumer<ProcessLog> processLogConsumer = processLog -> { ProcessLog.Level level = processLog.getLevel(); if (level.compareTo(levelReference.get()) >= 0) { processLogs.add(processLog); } }; ProcessChannel<String> processChannel = _localProcessExecutor.execute( _createJPDAProcessConfig(_JPDA_OPTIONS1, processLogConsumer), Operations.LEADING_LOG); Future<String> future = processChannel.getProcessNoticeableFuture(); Assert.assertEquals("DONE", future.get()); Assert.assertEquals(processLogs.toString(), 1, processLogs.size()); ProcessLog processLog = processLogs.remove(0); Assert.assertEquals( "Found corrupt leading log Leading log", processLog.getMessage()); levelReference.set(ProcessLog.Level.DEBUG); processChannel = _localProcessExecutor.execute( _createJPDAProcessConfig(_JPDA_OPTIONS1, processLogConsumer), Operations.LEADING_LOG); future = processChannel.getProcessNoticeableFuture(); Assert.assertEquals("DONE", future.get()); Assert.assertEquals(processLogs.toString(), 3, processLogs.size()); processLog = processLogs.remove(0); Assert.assertEquals( "Found corrupt leading log Leading log", processLog.getMessage()); processLog = processLogs.remove(0); String message = processLog.getMessage(); Assert.assertTrue( message, message.contains("Invoked generic process callable")); processLog = processLogs.remove(0); message = processLog.getMessage(); Assert.assertTrue( message, message.contains("Invoked generic process callable")); levelReference.set(ProcessLog.Level.ERROR); processChannel = _localProcessExecutor.execute( _createJPDAProcessConfig(_JPDA_OPTIONS1, processLogConsumer), Operations.LEADING_LOG); future = processChannel.getProcessNoticeableFuture(); Assert.assertEquals("DONE", future.get()); Assert.assertTrue(processLogs.toString(), processLogs.isEmpty()); }
@Test public void testSubprocessReactorPipingBackExceptionProcessCallable() throws Exception { List<ProcessLog> processLogs = new ArrayList<>(); ProcessChannel<Serializable> processChannel = _localProcessExecutor.execute( _createJPDAProcessConfig( _JPDA_OPTIONS1, processLog -> { if (processLog.getLevel() == ProcessLog.Level.ERROR) { processLogs.add(processLog); } }), Operations.PIPING_BACK_EXCEPTION_PROCESS_CALLABLE); NoticeableFuture<Serializable> noticeableFuture = processChannel.getProcessNoticeableFuture(); Assert.assertNull(noticeableFuture.get()); Assert.assertEquals(processLogs.toString(), 1, processLogs.size()); ProcessLog processLog = processLogs.get(0); Assert.assertEquals( "Unable to invoke generic process callable", processLog.getMessage()); Throwable throwable = processLog.getThrowable(); Assert.assertSame(ProcessException.class, throwable.getClass()); Assert.assertEquals( "Exception ProcessCallable", throwable.getMessage()); }
@Test public void testSubprocessReactorPipingBackNonprocessCallable() throws Exception { List<ProcessLog> processLogs = new ArrayList<>(); ProcessChannel<Serializable> processChannel = _localProcessExecutor.execute( _createJPDAProcessConfig( _JPDA_OPTIONS1, processLog -> { if (processLog.getLevel() == ProcessLog.Level.INFO) { processLogs.add(processLog); } }), Operations.PIPING_BACK_NONPROCESS_CALLABLE); NoticeableFuture<Serializable> noticeableFuture = processChannel.getProcessNoticeableFuture(); noticeableFuture.get(); Assert.assertEquals(processLogs.toString(), 1, processLogs.size()); ProcessLog processLog = processLogs.get(0); Assert.assertEquals( "Received a nonprocess callable piping back string piping back " + "object", processLog.getMessage()); }
@Test public void testSubprocessReactorPipingBackWriteAborted() throws Exception { List<ProcessLog> processLogs = new ArrayList<>(); ProcessChannel<Serializable> processChannel = _localProcessExecutor.execute( _createJPDAProcessConfig( _JPDA_OPTIONS1, processLog -> { if (processLog.getLevel() == ProcessLog.Level.WARN) { processLogs.add(processLog); } }), Operations.PIPING_BACK_WRITE_ABORTED); NoticeableFuture<Serializable> noticeableFuture = processChannel.getProcessNoticeableFuture(); try { noticeableFuture.get(); Assert.fail(); } catch (ExecutionException executionException) { Throwable throwable = executionException.getCause(); Assert.assertSame(ProcessException.class, throwable.getClass()); throwable = throwable.getCause(); Assert.assertSame( NotSerializableException.class, throwable.getClass()); Assert.assertEquals(processLogs.toString(), 1, processLogs.size()); ProcessLog processLog = processLogs.get(0); Assert.assertEquals( "Caught a write aborted exception", processLog.getMessage()); throwable = processLog.getThrowable(); Assert.assertSame( WriteAbortedException.class, throwable.getClass()); throwable = throwable.getCause(); Assert.assertSame( NotSerializableException.class, throwable.getClass()); } }