@Test public void testStreamingIterator() { String file = TestData.getTemporaryTestFile(); List<Value> values = Lists.newArrayList(); int count = TestData.getScaleCount(); for (int i = 0; i < count; ++i) { values.add(TestData.getValue()); } ByteBuffer bytes = ByteableCollections.toByteBuffer(values); FileSystem.writeBytes(bytes, file); int bufferSize = TestData.getScaleCount(); Iterator<ByteBuffer> it = ByteableCollections.streamingIterator(file, bufferSize); List<Value> newValues = Lists.newArrayList(); while (it.hasNext()) { newValues.add(Value.fromByteBuffer(it.next())); } Assert.assertEquals(values, newValues); }
public static Iterator<ByteBuffer> streamingIterator(Path file, long position, long length, int bufferSize) { return new Iterator<ByteBuffer>() { private long bufSize = bufferSize; private boolean expandBuffer = false; private Iterator<ByteBuffer> it = null; private long index = position; private long limit = index + length; { adjustBuffer(); } @Override public boolean hasNext() { ByteBuffer backingBytes = ((ByteableCollectionIterator) it).bytes; if(index < limit && it.hasNext()) { return true; } else if(index < limit && limit - index >= 4) { if(backingBytes.remaining() >= 4) { if(backingBytes.getInt() == 0) { backingBytes.position(backingBytes.position() - 4); return false; } else { backingBytes.position(backingBytes.position() - 4); } } adjustBuffer(); expandBuffer = true; return hasNext(); } else { return false; } } @Override public ByteBuffer next() { try { ByteBuffer next = it.next(); index += next.capacity() + 4; expandBuffer = false; return next; } catch (BufferUnderflowException e) { adjustBuffer(); return next(); } } @Override public void remove() { throw new UnsupportedOperationException(); } private void adjustBuffer() { if(expandBuffer) { bufSize *= 2; } it = iterator(FileSystem.map(file, MapMode.READ_ONLY, index, Math.min(limit - index, bufSize))); } }; }
ByteableCollections { public static Iterator<ByteBuffer> streamingIterator(Path file, long position, long length, int bufferSize) { return new Iterator<ByteBuffer>() { private long bufSize = bufferSize; private boolean expandBuffer = false; private Iterator<ByteBuffer> it = null; private long index = position; private long limit = index + length; { adjustBuffer(); } @Override public boolean hasNext() { ByteBuffer backingBytes = ((ByteableCollectionIterator) it).bytes; if(index < limit && it.hasNext()) { return true; } else if(index < limit && limit - index >= 4) { if(backingBytes.remaining() >= 4) { if(backingBytes.getInt() == 0) { backingBytes.position(backingBytes.position() - 4); return false; } else { backingBytes.position(backingBytes.position() - 4); } } adjustBuffer(); expandBuffer = true; return hasNext(); } else { return false; } } @Override public ByteBuffer next() { try { ByteBuffer next = it.next(); index += next.capacity() + 4; expandBuffer = false; return next; } catch (BufferUnderflowException e) { adjustBuffer(); return next(); } } @Override public void remove() { throw new UnsupportedOperationException(); } private void adjustBuffer() { if(expandBuffer) { bufSize *= 2; } it = iterator(FileSystem.map(file, MapMode.READ_ONLY, index, Math.min(limit - index, bufSize))); } }; } }
ByteableCollections { public static Iterator<ByteBuffer> streamingIterator(Path file, long position, long length, int bufferSize) { return new Iterator<ByteBuffer>() { private long bufSize = bufferSize; private boolean expandBuffer = false; private Iterator<ByteBuffer> it = null; private long index = position; private long limit = index + length; { adjustBuffer(); } @Override public boolean hasNext() { ByteBuffer backingBytes = ((ByteableCollectionIterator) it).bytes; if(index < limit && it.hasNext()) { return true; } else if(index < limit && limit - index >= 4) { if(backingBytes.remaining() >= 4) { if(backingBytes.getInt() == 0) { backingBytes.position(backingBytes.position() - 4); return false; } else { backingBytes.position(backingBytes.position() - 4); } } adjustBuffer(); expandBuffer = true; return hasNext(); } else { return false; } } @Override public ByteBuffer next() { try { ByteBuffer next = it.next(); index += next.capacity() + 4; expandBuffer = false; return next; } catch (BufferUnderflowException e) { adjustBuffer(); return next(); } } @Override public void remove() { throw new UnsupportedOperationException(); } private void adjustBuffer() { if(expandBuffer) { bufSize *= 2; } it = iterator(FileSystem.map(file, MapMode.READ_ONLY, index, Math.min(limit - index, bufSize))); } }; }  }
ByteableCollections { public static Iterator<ByteBuffer> streamingIterator(Path file, long position, long length, int bufferSize) { return new Iterator<ByteBuffer>() { private long bufSize = bufferSize; private boolean expandBuffer = false; private Iterator<ByteBuffer> it = null; private long index = position; private long limit = index + length; { adjustBuffer(); } @Override public boolean hasNext() { ByteBuffer backingBytes = ((ByteableCollectionIterator) it).bytes; if(index < limit && it.hasNext()) { return true; } else if(index < limit && limit - index >= 4) { if(backingBytes.remaining() >= 4) { if(backingBytes.getInt() == 0) { backingBytes.position(backingBytes.position() - 4); return false; } else { backingBytes.position(backingBytes.position() - 4); } } adjustBuffer(); expandBuffer = true; return hasNext(); } else { return false; } } @Override public ByteBuffer next() { try { ByteBuffer next = it.next(); index += next.capacity() + 4; expandBuffer = false; return next; } catch (BufferUnderflowException e) { adjustBuffer(); return next(); } } @Override public void remove() { throw new UnsupportedOperationException(); } private void adjustBuffer() { if(expandBuffer) { bufSize *= 2; } it = iterator(FileSystem.map(file, MapMode.READ_ONLY, index, Math.min(limit - index, bufSize))); } }; }  static Iterator<ByteBuffer> iterator(ByteBuffer bytes); static Iterator<ByteBuffer> iterator(ByteBuffer bytes,
            int sizePerElement); static Iterator<ByteBuffer> streamingIterator(Path file,
            long position, long length, int bufferSize); static Iterator<ByteBuffer> streamingIterator(final String file,
            final int bufferSize); static ByteBuffer toByteBuffer(
            Collection<? extends Byteable> collection); static ByteBuffer toByteBuffer(
            Collection<? extends Byteable> collection, int sizePerElement); }
ByteableCollections { public static Iterator<ByteBuffer> streamingIterator(Path file, long position, long length, int bufferSize) { return new Iterator<ByteBuffer>() { private long bufSize = bufferSize; private boolean expandBuffer = false; private Iterator<ByteBuffer> it = null; private long index = position; private long limit = index + length; { adjustBuffer(); } @Override public boolean hasNext() { ByteBuffer backingBytes = ((ByteableCollectionIterator) it).bytes; if(index < limit && it.hasNext()) { return true; } else if(index < limit && limit - index >= 4) { if(backingBytes.remaining() >= 4) { if(backingBytes.getInt() == 0) { backingBytes.position(backingBytes.position() - 4); return false; } else { backingBytes.position(backingBytes.position() - 4); } } adjustBuffer(); expandBuffer = true; return hasNext(); } else { return false; } } @Override public ByteBuffer next() { try { ByteBuffer next = it.next(); index += next.capacity() + 4; expandBuffer = false; return next; } catch (BufferUnderflowException e) { adjustBuffer(); return next(); } } @Override public void remove() { throw new UnsupportedOperationException(); } private void adjustBuffer() { if(expandBuffer) { bufSize *= 2; } it = iterator(FileSystem.map(file, MapMode.READ_ONLY, index, Math.min(limit - index, bufSize))); } }; }  static Iterator<ByteBuffer> iterator(ByteBuffer bytes); static Iterator<ByteBuffer> iterator(ByteBuffer bytes,
            int sizePerElement); static Iterator<ByteBuffer> streamingIterator(Path file,
            long position, long length, int bufferSize); static Iterator<ByteBuffer> streamingIterator(final String file,
            final int bufferSize); static ByteBuffer toByteBuffer(
            Collection<? extends Byteable> collection); static ByteBuffer toByteBuffer(
            Collection<? extends Byteable> collection, int sizePerElement);  }