StreamReader implements Closeable { @Override public void close() throws IOException { @LapsedWith(module = JDK8, value = "Optional#ifPresent") StreamScanner locScanner = this.scanner; if (locScanner != null) { locScanner.interrupt(); } @LapsedWith(module = JDK8, value = "Optional#ifPresent") Thread locThread = this.thread; if (locThread != null) { locThread.interrupt(); } } StreamReader(InputStream inputStream); void runReaderThread(final byte[] delimiter); void readUntilClosed(byte[] delimiter); @Override void close();  }
@Test public void canHandleDataAlreadyPresentSeparatedByNewline() throws Exception { List<String> expected = Arrays.asList("a", "b", "c"); StreamReader reader = process( new ByteArrayInputStream("a\nb\nc\n".getBytes()), "\n", expected); waitUntil(expected.size()); assertThat(received, is(expected)); reader.close(); }
@Test public void canHandleDataAlreadyPresentSeparatedByComma() throws Exception { List<String> expected = Arrays.asList("a", "b", "c"); StreamReader reader = process( new ByteArrayInputStream("a,b,c,".getBytes()), ",", expected); waitUntil(expected.size()); assertThat(received, is(expected)); reader.close(); }
@Test public void canHandleDataNotAlreadyPresentSeparatedByNewline() throws Exception { List<String> expected = Arrays.asList("a", "b", "c"); PipedOutputStream os = new PipedOutputStream(); PipedInputStream is = new PipedInputStream(os); StreamReader reader = process(is, "\n", expected); TimeUnit.SECONDS.sleep(2); os.write("a\nb\nc\n".getBytes()); waitUntil(expected.size()); assertThat(received, is(expected)); reader.close(); }
@Test public void canHandleDataNotAlreadyPresentSeparatedByComma() throws Exception { List<String> expected = Arrays.asList("a", "b", "c"); PipedOutputStream os = new PipedOutputStream(); PipedInputStream is = new PipedInputStream(os); StreamReader reader = process(is, ",", expected); TimeUnit.SECONDS.sleep(2); os.write("a,b,c,".getBytes()); waitUntil(expected.size()); assertThat(received, is(expected)); reader.close(); }