@Test public void testInterlevedReadAndWrite() throws IOException { PipedInputStream pipedInputStream = new PipedInputStream(); PipedOutputStream pipedOutputStream = new PipedOutputStream(pipedInputStream); PollingInputStream pollingInputStream = new PollingInputStream(pipedInputStream, Executors.newCachedThreadPool()); byte[] expectedBuffer = new byte[64]; for (int i = 0; i < expectedBuffer.length; i++) { expectedBuffer[i] = (byte) i; } byte[] actualBuffer = new byte[64]; pipedOutputStream.write(expectedBuffer, 0, 16); assertEquals(8, pollingInputStream.read(actualBuffer, 0, 8)); pipedOutputStream.write(expectedBuffer, 16, 48); int bytesRead = 0; while (bytesRead < 56) { bytesRead += pollingInputStream.read(actualBuffer, 8 + bytesRead, 64 - bytesRead); } assertArrayEquals(expectedBuffer, actualBuffer); }
@Override public synchronized int read(byte[] buffer, int offset, int length) throws IOException { int bytesRead = 0; if (length > 0) { while (available() == 0) { Thread.yield(); } synchronized (readBuffer) { bytesRead = Math.min(length, available()); System.arraycopy(readBuffer, readPosition, buffer, offset, bytesRead); readPosition += bytesRead; } } return bytesRead; }
PollingInputStream extends InputStream { @Override public synchronized int read(byte[] buffer, int offset, int length) throws IOException { int bytesRead = 0; if (length > 0) { while (available() == 0) { Thread.yield(); } synchronized (readBuffer) { bytesRead = Math.min(length, available()); System.arraycopy(readBuffer, readPosition, buffer, offset, bytesRead); readPosition += bytesRead; } } return bytesRead; } }
PollingInputStream extends InputStream { @Override public synchronized int read(byte[] buffer, int offset, int length) throws IOException { int bytesRead = 0; if (length > 0) { while (available() == 0) { Thread.yield(); } synchronized (readBuffer) { bytesRead = Math.min(length, available()); System.arraycopy(readBuffer, readPosition, buffer, offset, bytesRead); readPosition += bytesRead; } } return bytesRead; } PollingInputStream(final InputStream inputStream, ExecutorService executorService); }
PollingInputStream extends InputStream { @Override public synchronized int read(byte[] buffer, int offset, int length) throws IOException { int bytesRead = 0; if (length > 0) { while (available() == 0) { Thread.yield(); } synchronized (readBuffer) { bytesRead = Math.min(length, available()); System.arraycopy(readBuffer, readPosition, buffer, offset, bytesRead); readPosition += bytesRead; } } return bytesRead; } PollingInputStream(final InputStream inputStream, ExecutorService executorService); @Override synchronized int read(byte[] buffer, int offset, int length); @Override int read(); @Override int available(); }
PollingInputStream extends InputStream { @Override public synchronized int read(byte[] buffer, int offset, int length) throws IOException { int bytesRead = 0; if (length > 0) { while (available() == 0) { Thread.yield(); } synchronized (readBuffer) { bytesRead = Math.min(length, available()); System.arraycopy(readBuffer, readPosition, buffer, offset, bytesRead); readPosition += bytesRead; } } return bytesRead; } PollingInputStream(final InputStream inputStream, ExecutorService executorService); @Override synchronized int read(byte[] buffer, int offset, int length); @Override int read(); @Override int available();  }