@Test public void testReadWriteMultiFilterRecord() throws Exception { final List<FilterRecord<BloomFilter>> records = generateFilterRecords(100); for (FilterRecord<BloomFilter> record : records) { record.writeFullyTo(writeRecordChannel); } try (FilterRecordInputStream<GuavaBloomFilter> stream = new FilterRecordInputStream<>(tempFile.toPath(), factory)) { for (FilterRecord<BloomFilter> expectRecord : records) { assertThat(stream.nextFilterRecord()).isEqualTo(expectRecord); } assertThat(stream.nextFilterRecord()).isNull(); } }
@Nullable public FilterRecord<? extends F> nextFilterRecord() throws IOException { if (end - position <= HEADER_OVERHEAD) { if (end == position) { return null; } throw shortReadFilterHeader(recordFilePath.toString(), (int) (HEADER_OVERHEAD - (end - position))); } headerBuffer.rewind(); readFullyOrFail(channel, headerBuffer, position); headerBuffer.rewind(); final int bodyLen = headerBuffer.getInt(BODY_LENGTH_OFFSET); if (end - position < HEADER_OVERHEAD + bodyLen) { throw shortReadFilterBody(recordFilePath.toString(), (int) ((bodyLen + HEADER_OVERHEAD) - (end - position))); } checkMagic(headerBuffer); final ByteBuffer bodyBuffer = ByteBuffer.allocate(bodyLen); readFullyOrFail(channel, bodyBuffer, position + HEADER_OVERHEAD); bodyBuffer.flip(); checkCrc(headerBuffer, bodyBuffer); final ByteBuffer filterNameBuffer = readFilterNameBuffer(bodyBuffer); final String name = StandardCharsets.UTF_8.decode(filterNameBuffer).toString(); final F filter = factory.readFrom(new ByteArrayInputStream(bodyBuffer.array(), bodyBuffer.position() + filterNameBuffer.limit(), bodyBuffer.remaining())); position += bodyLen + HEADER_OVERHEAD; return new FilterRecord<>(name, filter); }
FilterRecordInputStream implements Closeable { @Nullable public FilterRecord<? extends F> nextFilterRecord() throws IOException { if (end - position <= HEADER_OVERHEAD) { if (end == position) { return null; } throw shortReadFilterHeader(recordFilePath.toString(), (int) (HEADER_OVERHEAD - (end - position))); } headerBuffer.rewind(); readFullyOrFail(channel, headerBuffer, position); headerBuffer.rewind(); final int bodyLen = headerBuffer.getInt(BODY_LENGTH_OFFSET); if (end - position < HEADER_OVERHEAD + bodyLen) { throw shortReadFilterBody(recordFilePath.toString(), (int) ((bodyLen + HEADER_OVERHEAD) - (end - position))); } checkMagic(headerBuffer); final ByteBuffer bodyBuffer = ByteBuffer.allocate(bodyLen); readFullyOrFail(channel, bodyBuffer, position + HEADER_OVERHEAD); bodyBuffer.flip(); checkCrc(headerBuffer, bodyBuffer); final ByteBuffer filterNameBuffer = readFilterNameBuffer(bodyBuffer); final String name = StandardCharsets.UTF_8.decode(filterNameBuffer).toString(); final F filter = factory.readFrom(new ByteArrayInputStream(bodyBuffer.array(), bodyBuffer.position() + filterNameBuffer.limit(), bodyBuffer.remaining())); position += bodyLen + HEADER_OVERHEAD; return new FilterRecord<>(name, filter); } }
FilterRecordInputStream implements Closeable { @Nullable public FilterRecord<? extends F> nextFilterRecord() throws IOException { if (end - position <= HEADER_OVERHEAD) { if (end == position) { return null; } throw shortReadFilterHeader(recordFilePath.toString(), (int) (HEADER_OVERHEAD - (end - position))); } headerBuffer.rewind(); readFullyOrFail(channel, headerBuffer, position); headerBuffer.rewind(); final int bodyLen = headerBuffer.getInt(BODY_LENGTH_OFFSET); if (end - position < HEADER_OVERHEAD + bodyLen) { throw shortReadFilterBody(recordFilePath.toString(), (int) ((bodyLen + HEADER_OVERHEAD) - (end - position))); } checkMagic(headerBuffer); final ByteBuffer bodyBuffer = ByteBuffer.allocate(bodyLen); readFullyOrFail(channel, bodyBuffer, position + HEADER_OVERHEAD); bodyBuffer.flip(); checkCrc(headerBuffer, bodyBuffer); final ByteBuffer filterNameBuffer = readFilterNameBuffer(bodyBuffer); final String name = StandardCharsets.UTF_8.decode(filterNameBuffer).toString(); final F filter = factory.readFrom(new ByteArrayInputStream(bodyBuffer.array(), bodyBuffer.position() + filterNameBuffer.limit(), bodyBuffer.remaining())); position += bodyLen + HEADER_OVERHEAD; return new FilterRecord<>(name, filter); } FilterRecordInputStream(Path recordFilePath, BloomFilterFactory<F, ?> factory); }
FilterRecordInputStream implements Closeable { @Nullable public FilterRecord<? extends F> nextFilterRecord() throws IOException { if (end - position <= HEADER_OVERHEAD) { if (end == position) { return null; } throw shortReadFilterHeader(recordFilePath.toString(), (int) (HEADER_OVERHEAD - (end - position))); } headerBuffer.rewind(); readFullyOrFail(channel, headerBuffer, position); headerBuffer.rewind(); final int bodyLen = headerBuffer.getInt(BODY_LENGTH_OFFSET); if (end - position < HEADER_OVERHEAD + bodyLen) { throw shortReadFilterBody(recordFilePath.toString(), (int) ((bodyLen + HEADER_OVERHEAD) - (end - position))); } checkMagic(headerBuffer); final ByteBuffer bodyBuffer = ByteBuffer.allocate(bodyLen); readFullyOrFail(channel, bodyBuffer, position + HEADER_OVERHEAD); bodyBuffer.flip(); checkCrc(headerBuffer, bodyBuffer); final ByteBuffer filterNameBuffer = readFilterNameBuffer(bodyBuffer); final String name = StandardCharsets.UTF_8.decode(filterNameBuffer).toString(); final F filter = factory.readFrom(new ByteArrayInputStream(bodyBuffer.array(), bodyBuffer.position() + filterNameBuffer.limit(), bodyBuffer.remaining())); position += bodyLen + HEADER_OVERHEAD; return new FilterRecord<>(name, filter); } FilterRecordInputStream(Path recordFilePath, BloomFilterFactory<F, ?> factory); @Nullable FilterRecord<? extends F> nextFilterRecord(); @Override void close(); }
FilterRecordInputStream implements Closeable { @Nullable public FilterRecord<? extends F> nextFilterRecord() throws IOException { if (end - position <= HEADER_OVERHEAD) { if (end == position) { return null; } throw shortReadFilterHeader(recordFilePath.toString(), (int) (HEADER_OVERHEAD - (end - position))); } headerBuffer.rewind(); readFullyOrFail(channel, headerBuffer, position); headerBuffer.rewind(); final int bodyLen = headerBuffer.getInt(BODY_LENGTH_OFFSET); if (end - position < HEADER_OVERHEAD + bodyLen) { throw shortReadFilterBody(recordFilePath.toString(), (int) ((bodyLen + HEADER_OVERHEAD) - (end - position))); } checkMagic(headerBuffer); final ByteBuffer bodyBuffer = ByteBuffer.allocate(bodyLen); readFullyOrFail(channel, bodyBuffer, position + HEADER_OVERHEAD); bodyBuffer.flip(); checkCrc(headerBuffer, bodyBuffer); final ByteBuffer filterNameBuffer = readFilterNameBuffer(bodyBuffer); final String name = StandardCharsets.UTF_8.decode(filterNameBuffer).toString(); final F filter = factory.readFrom(new ByteArrayInputStream(bodyBuffer.array(), bodyBuffer.position() + filterNameBuffer.limit(), bodyBuffer.remaining())); position += bodyLen + HEADER_OVERHEAD; return new FilterRecord<>(name, filter); } FilterRecordInputStream(Path recordFilePath, BloomFilterFactory<F, ?> factory); @Nullable FilterRecord<? extends F> nextFilterRecord(); @Override void close();  }