LinePipedInputStream extends PipedInputStream { public synchronized int read(byte[] b, int off, int len) throws IOException { int c = read(); if (c < 0) { return -1; } b[off] = (byte) c; int rlen = 1; if(c == NEW_LINE) { return rlen; } int enter = -1; while ((in >= 0) && (len > 1)) { int available; if (in > out) { available = Math.min((buffer.length - out), (in - out)); } else { available = buffer.length - out; } if (available > (len - 1)) { available = len - 1; } enter = findEnter(buffer, out, out+available); if(enter > -1) { if(enter+1 <= available) available = enter+1; else available = enter; } System.arraycopy(buffer, out, b, off + rlen, available); out += available; rlen += available; len -= available; if (out >= buffer.length) { out = 0; } if (in == out) { in = -1; } if(enter > -1) { return rlen; } } return rlen; } LinePipedInputStream(int pipeSize);  LinePipedInputStream(PipedOutputStream src); synchronized int read(byte[] b, int off, int len);  }
@Test public void testInputStream() throws IOException { PipedOutputStream outputStream = new PipedOutputStream(); LinePipedInputStream pipedInputStream = new LinePipedInputStream(outputStream); int numberOfLines = 0; String filename = "src"+Config.getPathSeparator()+"test"+Config.getPathSeparator()+ "resources"+Config.getPathSeparator()+"input_stream.txt"; for(String line : Files.readAllLines(Paths.get(filename), StandardCharsets.UTF_8)) { try { numberOfLines++; outputStream.write((line+ Config.getLineSeparator()).getBytes()); } catch (IOException e) { e.printStackTrace(); } } outputStream.flush(); int numberOfPipedLines = 0; byte[] bBuf = new byte[1024]; for(String line : Files.readAllLines(Paths.get(filename), StandardCharsets.UTF_8)) { try { int read = pipedInputStream.read(bBuf); String in = new String(bBuf, 0, read); assertEquals(line+Config.getLineSeparator(), in); numberOfPipedLines++; } catch (IOException e) { e.printStackTrace(); } } assertEquals(numberOfLines, numberOfPipedLines); }