SegmentBufferWriter implements WriteOperationHandler { @Override public void flush(@NotNull SegmentStore store) throws IOException { if (dirty) { int referencedSegmentIdCount = segmentReferences.size(); BinaryUtils.writeInt(buffer, Segment.REFERENCED_SEGMENT_ID_COUNT_OFFSET, referencedSegmentIdCount); statistics.segmentIdCount = referencedSegmentIdCount; int recordNumberCount = recordNumbers.size(); BinaryUtils.writeInt(buffer, Segment.RECORD_NUMBER_COUNT_OFFSET, recordNumberCount); int totalLength = align(HEADER_SIZE + referencedSegmentIdCount * SEGMENT_REFERENCE_SIZE + recordNumberCount * RECORD_SIZE + length, 16); if (totalLength > buffer.length) { LOG.warn("Segment buffer corruption detected\n{}", dumpSegmentBuffer()); throw new IllegalStateException(String.format( "Too much data for a segment %s (referencedSegmentIdCount=%d, recordNumberCount=%d, length=%d, totalLength=%d)", segment.getSegmentId(), referencedSegmentIdCount, recordNumberCount, length, totalLength)); } statistics.size = length = totalLength; int pos = HEADER_SIZE; if (pos + length <= buffer.length) { arraycopy(buffer, 0, buffer, buffer.length - length, pos); pos += buffer.length - length; } else { length = buffer.length; } for (SegmentId segmentId : segmentReferences) { pos = BinaryUtils.writeLong(buffer, pos, segmentId.getMostSignificantBits()); pos = BinaryUtils.writeLong(buffer, pos, segmentId.getLeastSignificantBits()); } for (Entry entry : recordNumbers) { pos = BinaryUtils.writeInt(buffer, pos, entry.getRecordNumber()); pos = BinaryUtils.writeByte(buffer, pos, (byte) entry.getType().ordinal()); pos = BinaryUtils.writeInt(buffer, pos, entry.getOffset()); } SegmentId segmentId = segment.getSegmentId(); LOG.debug("Writing data segment: {} ", statistics); store.writeSegment(segmentId, buffer, buffer.length - length, length); newSegment(store); } } SegmentBufferWriter(@NotNull SegmentIdProvider idProvider,
                               @NotNull SegmentReader reader,
                               @Nullable String wid,
                               @NotNull GCGeneration gcGeneration); @NotNull @Override RecordId execute(@NotNull GCGeneration gcGeneration,
                            @NotNull WriteOperation writeOperation); @Override @NotNull GCGeneration getGCGeneration(); void writeByte(byte value); void writeShort(short value); void writeInt(int value); void writeLong(long value); void writeRecordId(RecordId recordId); void writeBytes(byte[] data, int offset, int length); @Override void flush(@NotNull SegmentStore store); RecordId prepare(RecordType type, int size, Collection<RecordId> ids, SegmentStore store);  }
@Test public void nonDirtyBuffersShouldNotBeFlushed() throws Exception { List<SegmentId> before; try (FileStore store = openFileStore()) { } try (ReadOnlyFileStore store = openReadOnlyFileStore()) { before = newArrayList(store.getSegmentIds()); } try (FileStore store = openFileStore()) { defaultSegmentWriterBuilder("t").build(store).flush(); } List<SegmentId> after; try (ReadOnlyFileStore store = openReadOnlyFileStore()) { after = newArrayList(store.getSegmentIds()); } assertEquals(before, after); }
@Test public void dirtyBuffersShouldBeFlushed() throws Exception { List<SegmentId> before; try (FileStore store = openFileStore()) { } try (ReadOnlyFileStore store = openReadOnlyFileStore()) { before = newArrayList(store.getSegmentIds()); } try (FileStore store = openFileStore()) { SegmentWriter writer = defaultSegmentWriterBuilder("t").build(store); writer.writeNode(EmptyNodeState.EMPTY_NODE); writer.flush(); } List<SegmentId> after; try (ReadOnlyFileStore store = openReadOnlyFileStore()) { after = newArrayList(store.getSegmentIds()); } assertNotEquals(before, after); }