Optimize SegmentedJournalReader.getCurrentEntry()

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

diff --git a/atomix-storage/src/main/java/io/atomix/storage/journal/SegmentedJournal.java b/atomix-storage/src/main/java/io/atomix/storage/journal/SegmentedJournal.java
index f2d99eec533a094f36ef82476d8094f7f2ddd5cb..7da4655a48591d1192be4be4a6f6419f0c2e9371 100644 (file)
--- a/atomix-storage/src/main/java/io/atomix/storage/journal/SegmentedJournal.java
+++ b/atomix-storage/src/main/java/io/atomix/storage/journal/SegmentedJournal.java
@@ -40,7 +40,6 @@ import static java.util.Objects.requireNonNull;
  * Segmented journal.
  */
 public final class SegmentedJournal<E> implements Journal<E> {
-
   /**
    * Returns a new Raft log builder.
    *
@@ -91,7 +90,7 @@ public final class SegmentedJournal<E> implements Journal<E> {
     this.indexDensity = indexDensity;
     this.flushOnCommit = flushOnCommit;
     open();
-    this.writer = openWriter();
+    this.writer = new SegmentedJournalWriter<>(this);
   }
 
   /**
@@ -194,13 +193,13 @@ public final class SegmentedJournal<E> implements Journal<E> {
   }
 
   @Override
-  public SegmentedJournalWriter<E> writer() {
+  public JournalWriter<E> writer() {
     return writer;
   }
 
   @Override
-  public SegmentedJournalReader<E> openReader(long index) {
-    return openReader(index, SegmentedJournalReader.Mode.ALL);
+  public JournalReader<E> openReader(long index) {
+    return openReader(index, JournalReader.Mode.ALL);
   }
 
   /**
@@ -210,21 +209,22 @@ public final class SegmentedJournal<E> implements Journal<E> {
    * @param mode The mode in which to read entries.
    * @return The Raft log reader.
    */
-  public SegmentedJournalReader<E> openReader(long index, SegmentedJournalReader.Mode mode) {
-    SegmentedJournalReader<E> reader = new SegmentedJournalReader<>(this, index, mode);
+  public JournalReader<E> openReader(long index, JournalReader.Mode mode) {
+    final var segment = getSegment(index);
+    final var reader = switch (mode) {
+      case ALL -> new SegmentedJournalReader<>(this, segment);
+      case COMMITS -> new CommitsSegmentJournalReader<>(this, segment);
+    };
+
+    // Forward reader to specified index
+    for (long next = reader.getNextIndex(); index > next && reader.hasNext(); next = reader.getNextIndex()) {
+      reader.next();
+    }
+
     readers.add(reader);
     return reader;
   }
 
-  /**
-   * Opens a new journal writer.
-   *
-   * @return A new journal writer.
-   */
-  protected SegmentedJournalWriter<E> openWriter() {
-    return new SegmentedJournalWriter<>(this);
-  }
-
   /**
    * Opens the segments.
    */