@Test public void testRead() throws Exception { DistributedLogManager dlm = mock(DistributedLogManager.class); LogReader reader = mock(LogReader.class); when(dlm.getInputStream(any(DLSN.class))).thenReturn(reader); byte[] data = "test-read".getBytes(UTF_8); LogRecordWithDLSN record = mock(LogRecordWithDLSN.class); when(record.getPayLoadInputStream()) .thenReturn(new ByteArrayInputStream(data)); when(reader.readNext(anyBoolean())) .thenReturn(record) .thenThrow(new EndOfStreamException("eos")); DLInputStream in = new DLInputStream(dlm); int numReads = 0; int readByte; while ((readByte = in.read()) != -1) { assertEquals(data[numReads], readByte); ++numReads; } assertEquals(data.length, numReads); }
@Override public int read() throws IOException { byte[] b = new byte[1]; if (read(b, 0, 1) != 1) { return -1; } else { return b[0]; } }
DLInputStream extends InputStream { @Override public int read() throws IOException { byte[] b = new byte[1]; if (read(b, 0, 1) != 1) { return -1; } else { return b[0]; } } }
DLInputStream extends InputStream { @Override public int read() throws IOException { byte[] b = new byte[1]; if (read(b, 0, 1) != 1) { return -1; } else { return b[0]; } } DLInputStream(DistributedLogManager dlm); }
DLInputStream extends InputStream { @Override public int read() throws IOException { byte[] b = new byte[1]; if (read(b, 0, 1) != 1) { return -1; } else { return b[0]; } } DLInputStream(DistributedLogManager dlm); @Override int read(); @Override int read(byte[] b, int off, int len); @Override void close(); }
DLInputStream extends InputStream { @Override public int read() throws IOException { byte[] b = new byte[1]; if (read(b, 0, 1) != 1) { return -1; } else { return b[0]; } } DLInputStream(DistributedLogManager dlm); @Override int read(); @Override int read(byte[] b, int off, int len); @Override void close();  }