### Question:
ExecNioSource extends EventSource.Base { public void close() throws IOException { readOut.shutdown(); readErr.shutdown(); boolean latched = false; try { latched = latch.await(5000, TimeUnit.MILLISECONDS); } catch (InterruptedException e) { LOG.debug("Waiting for exec thread exit was interrupted", e); } stdinISP.shutdown(); stderrISP.shutdown(); if (proc != null) { proc.destroy(); proc = null; } if (!latched) { throw new IOException("Timeout waiting for exec threads to exit"); } } ExecNioSource(String command, boolean aggregate, boolean restart, int period); void close(); Event next(); void open(); static SourceBuilder buildPeriodic(); static SourceBuilder buildStream(); static SourceBuilder builder(); static final String A_PROC_SOURCE; static final String A_EXEC_CMD; }### Answer:
@Test public void testClose() throws IOException, InterruptedException { File temp = File.createTempFile("flmtst", null); temp.deleteOnExit(); BufferedWriter out = new BufferedWriter(new FileWriter(temp)); out.write("#!/bin/bash\n while true; do sleep 1; done; >&2 \n"); out.close(); String cmd = temp.getAbsolutePath(); final EventSource source = new ExecNioSource.Builder().build("/bin/bash " + cmd); final CountDownLatch started = new CountDownLatch(1); final CountDownLatch latch = new CountDownLatch(1); Thread t = new Thread() { public void run() { try { source.open(); started.countDown(); source.next(); latch.countDown(); } catch (Exception e) { LOG.warn("Event consumption thread caught exception, test will fail", e); } } }; t.start(); started.await(1000, TimeUnit.MILLISECONDS); Clock.sleep(100); source.close(); assertTrue("source.next did not exit after close within 5 seconds", latch .await(5000, TimeUnit.MILLISECONDS)); }