atomix-storage/src/main/java/io/atomix/storage/journal/JournalSegmentReader.java

   1 /*
   2  * Copyright (c) 2024 PANTHEON.tech, s.r.o. and others.  All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8 package io.atomix.storage.journal;
   9
  10 import static java.util.Objects.requireNonNull;
  11
  12 import io.atomix.storage.journal.index.JournalIndex;
  13 import io.atomix.storage.journal.index.Position;
  14 import java.util.NoSuchElementException;
  15 import org.eclipse.jdt.annotation.Nullable;
  16
  17 abstract sealed class JournalSegmentReader<E> implements JournalReader<E>
  18         permits FileChannelJournalSegmentReader, MappedJournalSegmentReader {
  19     final int maxEntrySize;
  20     private final JournalIndex index;
  21     final JournalSerdes namespace;
  22     private final long firstIndex;
  23
  24     private Indexed<E> currentEntry;
  25     private Indexed<E> nextEntry;
  26
  27     JournalSegmentReader(final JournalSegment<E> segment, final int maxEntrySize, final JournalIndex index,
  28             final JournalSerdes namespace) {
  29         this.maxEntrySize = maxEntrySize;
  30         this.index = requireNonNull(index);
  31         this.namespace = requireNonNull(namespace);
  32         firstIndex = segment.index();
  33     }
  34
  35     @Override
  36     public final long getFirstIndex() {
  37         return firstIndex;
  38     }
  39
  40     @Override
  41     public final long getCurrentIndex() {
  42         return currentEntry != null ? currentEntry.index() : 0;
  43     }
  44
  45     @Override
  46     public final Indexed<E> getCurrentEntry() {
  47         return currentEntry;
  48     }
  49
  50     @Override
  51     public final long getNextIndex() {
  52         return currentEntry != null ? currentEntry.index() + 1 : firstIndex;
  53     }
  54
  55     @Override
  56     public final boolean hasNext() {
  57         return nextEntry != null || (nextEntry = readNext()) != null;
  58     }
  59
  60     @Override
  61     public final Indexed<E> next() {
  62         if (!hasNext()) {
  63             throw new NoSuchElementException();
  64         }
  65
  66         // Set the current entry to the next entry.
  67         currentEntry = nextEntry;
  68
  69         // Reset the next entry to null.
  70         nextEntry = null;
  71
  72         // Read the next entry in the segment.
  73         nextEntry = readNext();
  74
  75         // Return the current entry.
  76         return currentEntry;
  77     }
  78
  79     @Override
  80     public final void reset() {
  81         currentEntry = null;
  82         nextEntry = null;
  83         setPosition(JournalSegmentDescriptor.BYTES);
  84         nextEntry = readNext();
  85     }
  86
  87     @Override
  88     public final void reset(final long index) {
  89         reset();
  90         Position position = this.index.lookup(index - 1);
  91         if (position != null) {
  92             currentEntry = new Indexed<>(position.index() - 1, null, 0);
  93             setPosition(position.position());
  94             nextEntry = readNext();
  95         }
  96         while (getNextIndex() < index && hasNext()) {
  97             next();
  98         }
  99     }
 100
 101     @Override
 102     public final void close() {
 103         // FIXME: CONTROLLER-2098: remove this method
 104     }
 105
 106     /**
 107      * Set the file position.
 108      *
 109      * @param position new position
 110      */
 111     abstract void setPosition(int position);
 112
 113     /**
 114      * Reads the next entry in the segment.
 115      *
 116      * @return Next entry, or {@code null}
 117      */
 118     abstract @Nullable Indexed<E> readNext();
 119 }