@Test public void messageReadSetIndexInputStreamTest() throws Exception { ArrayList<ByteBuffer> listbuf = new ArrayList<ByteBuffer>(); byte[] buf1 = new byte[1024]; byte[] buf2 = new byte[2048]; byte[] buf3 = new byte[4096]; new Random().nextBytes(buf1); new Random().nextBytes(buf2); new Random().nextBytes(buf3); listbuf.add(ByteBuffer.wrap(buf1)); listbuf.add(ByteBuffer.wrap(buf2)); listbuf.add(ByteBuffer.wrap(buf3)); ArrayList<StoreKey> storeKeys = new ArrayList<StoreKey>(); storeKeys.add(new MockId("012345678910123223233456789012")); storeKeys.add(new MockId("012345678910123223233456789013")); storeKeys.add(new MockId("012345678910123223233456789014")); MessageReadSet readSet = new MockMessageReadSet(listbuf, storeKeys); MessageReadSetIndexInputStream stream1 = new MessageReadSetIndexInputStream(readSet, 0, 0); byte[] buf1Output = new byte[1024]; Assert.assertEquals("Number of bytes read doesn't match", 1024, stream1.read(buf1Output, 0, 1024)); Assert.assertArrayEquals(buf1Output, buf1); MessageReadSetIndexInputStream stream2 = new MessageReadSetIndexInputStream(readSet, 1, 1024); byte[] buf2Output = new byte[1025]; Assert.assertEquals("Number of bytes read doesn't match", 512, stream2.read(buf2Output, 0, 512)); Assert.assertEquals("Number of bytes read doesn't match", 512, stream2.read(buf2Output, 512, 513)); for (int i = 0; i < 1024; i++) { Assert.assertEquals(buf2Output[i], buf2[i + 1024]); } MessageReadSetIndexInputStream stream3 = new MessageReadSetIndexInputStream(readSet, 2, 2048); byte[] buf3Output = new byte[2048]; for (int i = 0; i < 2048; i++) { Assert.assertEquals((byte) stream3.read(), buf3[i + 2048]); } Assert.assertEquals("Should return -1 if no more data available", -1, stream3.read(buf3Output, 0, 1)); Assert.assertEquals("Should return -1 if no more data available", -1, stream3.read()); }
public MessageReadSetIndexInputStream(MessageReadSet messageReadSet, int indexToRead, int startingOffset) { this.messageReadSet = messageReadSet; if (indexToRead >= messageReadSet.count()) { throw new IllegalArgumentException( "The index provided " + indexToRead + " " + "is outside the bounds of the number of messages in the read set " + messageReadSet.count()); } this.indexToRead = indexToRead; this.currentOffset = startingOffset; }
MessageReadSetIndexInputStream extends InputStream { public MessageReadSetIndexInputStream(MessageReadSet messageReadSet, int indexToRead, int startingOffset) { this.messageReadSet = messageReadSet; if (indexToRead >= messageReadSet.count()) { throw new IllegalArgumentException( "The index provided " + indexToRead + " " + "is outside the bounds of the number of messages in the read set " + messageReadSet.count()); } this.indexToRead = indexToRead; this.currentOffset = startingOffset; } }
MessageReadSetIndexInputStream extends InputStream { public MessageReadSetIndexInputStream(MessageReadSet messageReadSet, int indexToRead, int startingOffset) { this.messageReadSet = messageReadSet; if (indexToRead >= messageReadSet.count()) { throw new IllegalArgumentException( "The index provided " + indexToRead + " " + "is outside the bounds of the number of messages in the read set " + messageReadSet.count()); } this.indexToRead = indexToRead; this.currentOffset = startingOffset; } MessageReadSetIndexInputStream(MessageReadSet messageReadSet, int indexToRead, int startingOffset); }
MessageReadSetIndexInputStream extends InputStream { public MessageReadSetIndexInputStream(MessageReadSet messageReadSet, int indexToRead, int startingOffset) { this.messageReadSet = messageReadSet; if (indexToRead >= messageReadSet.count()) { throw new IllegalArgumentException( "The index provided " + indexToRead + " " + "is outside the bounds of the number of messages in the read set " + messageReadSet.count()); } this.indexToRead = indexToRead; this.currentOffset = startingOffset; } MessageReadSetIndexInputStream(MessageReadSet messageReadSet, int indexToRead, int startingOffset); @Override int read(); @Override int read(byte b[], int off, int len); }
MessageReadSetIndexInputStream extends InputStream { public MessageReadSetIndexInputStream(MessageReadSet messageReadSet, int indexToRead, int startingOffset) { this.messageReadSet = messageReadSet; if (indexToRead >= messageReadSet.count()) { throw new IllegalArgumentException( "The index provided " + indexToRead + " " + "is outside the bounds of the number of messages in the read set " + messageReadSet.count()); } this.indexToRead = indexToRead; this.currentOffset = startingOffset; } MessageReadSetIndexInputStream(MessageReadSet messageReadSet, int indexToRead, int startingOffset); @Override int read(); @Override int read(byte b[], int off, int len);  }