Refactor JournalSegmentReader

[controller.git] / atomix-storage / src / main / java / io / atomix / storage / journal / MappedJournalSegmentReader.java

diff --git a/atomix-storage/src/main/java/io/atomix/storage/journal/MappedJournalSegmentReader.java b/atomix-storage/src/main/java/io/atomix/storage/journal/MappedJournalSegmentReader.java
index bbf100740a50d25af843dad80250c153f6013163..e9059723588893fe9a6a81b709aa2198fc567605 100644 (file)
--- a/atomix-storage/src/main/java/io/atomix/storage/journal/MappedJournalSegmentReader.java
+++ b/atomix-storage/src/main/java/io/atomix/storage/journal/MappedJournalSegmentReader.java
@@ -16,9 +16,7 @@
  */
 package io.atomix.storage.journal;
 
-import java.nio.BufferUnderflowException;
 import java.nio.ByteBuffer;
-import java.util.zip.CRC32;
 
 /**
  * Log segment reader.
@@ -26,59 +24,21 @@ import java.util.zip.CRC32;
  * @author <a href="http://github.com/kuujo">Jordan Halterman</a>
  */
 final class MappedJournalSegmentReader<E> extends JournalSegmentReader<E> {
-  private final ByteBuffer buffer;
+    private final ByteBuffer buffer;
 
-  MappedJournalSegmentReader(
-      ByteBuffer buffer,
-      JournalSegment<E> segment,
-      int maxEntrySize,
-      JournalSerdes namespace) {
-    super(segment, maxEntrySize, namespace);
-    this.buffer = buffer.slice();
-  }
-
-  @Override
-  void setPosition(int position) {
-    buffer.position(position);
-  }
-
-  @Override
-  Indexed<E> readEntry(final long index) {
-    // Mark the buffer so it can be reset if necessary.
-    buffer.mark();
-
-    try {
-      // Read the length of the entry.
-      final int length = buffer.getInt();
-
-      // If the buffer length is zero then return.
-      if (length <= 0 || length > maxEntrySize) {
-        buffer.reset();
-        return null;
-      }
-
-      // Read the checksum of the entry.
-      long checksum = buffer.getInt() & 0xFFFFFFFFL;
+    MappedJournalSegmentReader(final ByteBuffer buffer, final JournalSegment<E> segment, final int maxEntrySize,
+            final JournalSerdes namespace) {
+        super(segment, maxEntrySize, namespace);
+        this.buffer = buffer.slice().asReadOnlyBuffer();
+    }
 
-      // Compute the checksum for the entry bytes.
-      final CRC32 crc32 = new CRC32();
-      ByteBuffer slice = buffer.slice();
-      slice.limit(length);
-      crc32.update(slice);
+    @Override
+    void invalidateCache() {
+        // No-op: the mapping is guaranteed to be coherent
+    }
 
-      // If the stored checksum equals the computed checksum, return the entry.
-      if (checksum == crc32.getValue()) {
-        slice.rewind();
-        E entry = namespace.deserialize(slice);
-        buffer.position(buffer.position() + length);
-        return new Indexed<>(index, entry, length);
-      } else {
-        buffer.reset();
-        return null;
-      }
-    } catch (BufferUnderflowException e) {
-      buffer.reset();
-      return null;
+    @Override
+    ByteBuffer read(final int position, final int size) {
+        return buffer.slice(position, size);
     }
-  }
 }