ExecNioSource extends EventSource.Base { public void open() throws IOException { if (proc != null) { throw new IllegalStateException("Tried to open exec process twice"); } latch = new CountDownLatch(2); outFinished.set(false); errFinished.set(false); proc = Runtime.getRuntime().exec(command); stdinISP = new InputStreamPipe(proc.getInputStream()); stderrISP = new InputStreamPipe(proc.getErrorStream()); stdout = (ReadableByteChannel) stdinISP.getChannel(); stderr = (ReadableByteChannel) stderrISP.getChannel(); readOut = new ReaderThread(stdout, "STDOUT", outFinished); readErr = new ReaderThread(stderr, "STDERR", errFinished); stdinISP.start(); stderrISP.start(); readOut.start(); readErr.start(); } 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; }
@Test public void testNewPipeBlockingSemantics() throws IOException { Pipe pipe = Pipe.open(); ByteBuffer buf = ByteBuffer.allocate(10); assertTrue(pipe.source().isBlocking()); pipe.source().configureBlocking(false); assertTrue(!pipe.source().isBlocking()); int rdSz = pipe.source().read(buf); assertTrue(rdSz <= 0); }
@Test public void testNewPipePartialReadSemantics() throws IOException { Pipe pipe = Pipe.open(); int sz = 10; ByteBuffer buf = ByteBuffer.allocate(sz); pipe.sink().write(ByteBuffer.wrap("small".getBytes())); int rdSz = pipe.source().read(buf); assertTrue(rdSz <= sz); }