BufferedPullDataSource extends SimplePullDataSource { @Override public int read(long position, byte[] buffer, int offset, int len) throws IOException { int total = 0; int read = 0; int sanity = 0; while (true) { read = bufferedRead(position + total, buffer, offset + total, Math.min(len - total, MAX_BUFFER)); if (read == -1 && total == 0) return -1; if (read == -1) break; if (read == 0) break; total += read; if (total >= len) break; if (sanity++ > 500) { throw new IOException("Invalid buffer fill request: len: " + len + "; total: " + total); } } return total; } BufferedPullDataSource();  BufferedPullDataSource(String host); @Override void close(); @Override int read(long position, byte[] buffer, int offset, int len); @Override long open(String uri); void flush();  }
@Test public void testRead() throws Exception { BufferedPullDataSource pds = spy(new BufferedPullDataSource()); pds.remoteReader = new DataInputStream(new MockInputStream(BufferedPullDataSource.MAX_BUFFER * 5)); pds.remoteWriter = new ByteArrayOutputStream(); pds._buffer = new byte[BufferedPullDataSource.MAX_BUFFER]; pds.opened = true; pds.size = BufferedPullDataSource.MAX_BUFFER * 5; byte buffer[] = new byte[BufferedPullDataSource.MAX_BUFFER]; int size = 1; int read = pds.read(0, buffer, 0, 1); assertEquals(size, read); assertEquals(new String(data, 0, size), new String(buffer, 0, size)); verify(pds, times(1)).fillBuffer(eq(0l)); read = pds.read(0, buffer, 1, 1); assertEquals(size, read); assertEquals(new String(data, 1, size), new String(buffer, 1, size)); verify(pds, times(1)).fillBuffer(eq(0l)); read = pds.read(BufferedPullDataSource.MAX_BUFFER * 2, buffer, 0, 1); assertEquals(size, read); verify(pds, times(1)).fillBuffer(eq(BufferedPullDataSource.MAX_BUFFER * 2l)); }