// Read the checksum of the entry.
final long checksum = buffer.getInt() & 0xFFFFFFFFL;
+ // Slice off the entry's bytes
+ final ByteBuffer entryBytes = buffer.slice();
+ entryBytes.limit(length);
+
// Compute the checksum for the entry bytes.
final CRC32 crc32 = new CRC32();
- ByteBuffer slice = buffer.slice();
- slice.limit(length);
- crc32.update(slice);
-
- // If the stored checksum equals the computed checksum, return the entry.
- if (checksum == crc32.getValue()) {
- slice.rewind();
- final E entry = namespace.deserialize(slice);
- lastEntry = new Indexed<>(nextIndex, entry, length);
- this.index.index(nextIndex, position);
- nextIndex++;
- } else {
+ crc32.update(entryBytes);
+
+ // If the stored checksum does not equal the computed checksum, do not proceed further
+ if (checksum != crc32.getValue()) {
break;
}
+ entryBytes.rewind();
+ final E entry = namespace.deserialize(entryBytes);
+ lastEntry = new Indexed<>(nextIndex, entry, length);
+ this.index.index(nextIndex, position);
+ nextIndex++;
+
// Update the current position for indexing.
position = buffer.position() + length;
buffer.position(position);
}
}
- /**
- * Returns the size of the underlying buffer.
- *
- * @return The size of the underlying buffer.
- */
- public long size() {
- return buffer.position() + JournalSegmentDescriptor.BYTES;
- }
-
- /**
- * Returns a boolean indicating whether the segment is empty.
- *
- * @return Indicates whether the segment is empty.
- */
- public boolean isEmpty() {
- return lastEntry == null;
- }
-
@Override
public void append(Indexed<E> entry) {
final long nextIndex = getNextIndex();
this.index.truncate(index);
if (index < segment.index()) {
- buffer.position(JournalSegmentDescriptor.BYTES);
- buffer.putInt(0);
- buffer.putInt(0);
+ // Reset the writer to the first entry.
buffer.position(JournalSegmentDescriptor.BYTES);
} else {
// Reset the writer to the given index.
reset(index);
-
- // Zero entries after the given index.
- int position = buffer.position();
- buffer.putInt(0);
- buffer.putInt(0);
- buffer.position(position);
}
+
+ // Zero the entry header at current buffer position.
+ int position = buffer.position();
+ // Note: we issue a single putLong() instead of two putInt()s.
+ buffer.putLong(0);
+ buffer.position(position);
}
@Override