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

   1 /*
   2  * Copyright 2018-present Open Networking Foundation
   3  *
   4  * Licensed under the Apache License, Version 2.0 (the "License");
   5  * you may not use this file except in compliance with the License.
   6  * You may obtain a copy of the License at
   7  *
   8  * http://www.apache.org/licenses/LICENSE-2.0
   9  *
  10  * Unless required by applicable law or agreed to in writing, software
  11  * distributed under the License is distributed on an "AS IS" BASIS,
  12  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13  * See the License for the specific language governing permissions and
  14  * limitations under the License.
  15  */
  16 package io.atomix.storage.journal;
  17
  18 import io.atomix.storage.journal.index.JournalIndex;
  19 import java.nio.ByteBuffer;
  20 import java.nio.channels.FileChannel;
  21
  22 /**
  23  * Mappable log segment reader.
  24  */
  25 final class MappableJournalSegmentReader<E> implements JournalReader<E> {
  26   private final JournalSegment<E> segment;
  27   private final FileChannel channel;
  28   private final int maxEntrySize;
  29   private final JournalIndex index;
  30   private final JournalSerdes namespace;
  31   private JournalSegmentReader<E> reader;
  32
  33   MappableJournalSegmentReader(
  34       FileChannel channel,
  35       JournalSegment<E> segment,
  36       int maxEntrySize,
  37       JournalIndex index,
  38       JournalSerdes namespace) {
  39     this.channel = channel;
  40     this.segment = segment;
  41     this.maxEntrySize = maxEntrySize;
  42     this.index = index;
  43     this.namespace = namespace;
  44     this.reader = new FileChannelJournalSegmentReader<>(channel, segment, maxEntrySize, index, namespace);
  45   }
  46
  47   /**
  48    * Converts the reader to a mapped reader using the given buffer.
  49    *
  50    * @param buffer the mapped buffer
  51    */
  52   void map(ByteBuffer buffer) {
  53     if (!(reader instanceof MappedJournalSegmentReader)) {
  54       JournalReader<E> reader = this.reader;
  55       this.reader = new MappedJournalSegmentReader<>(buffer, segment, maxEntrySize, index, namespace);
  56       this.reader.reset(reader.getNextIndex());
  57       reader.close();
  58     }
  59   }
  60
  61   /**
  62    * Converts the reader to an unmapped reader.
  63    */
  64   void unmap() {
  65     if (reader instanceof MappedJournalSegmentReader) {
  66       JournalReader<E> reader = this.reader;
  67       this.reader = new FileChannelJournalSegmentReader<>(channel, segment, maxEntrySize, index, namespace);
  68       this.reader.reset(reader.getNextIndex());
  69       reader.close();
  70     }
  71   }
  72
  73   @Override
  74   public long getFirstIndex() {
  75     return reader.getFirstIndex();
  76   }
  77
  78   @Override
  79   public long getCurrentIndex() {
  80     return reader.getCurrentIndex();
  81   }
  82
  83   @Override
  84   public Indexed<E> getCurrentEntry() {
  85     return reader.getCurrentEntry();
  86   }
  87
  88   @Override
  89   public long getNextIndex() {
  90     return reader.getNextIndex();
  91   }
  92
  93   @Override
  94   public boolean hasNext() {
  95     return reader.hasNext();
  96   }
  97
  98   @Override
  99   public Indexed<E> next() {
 100     return reader.next();
 101   }
 102
 103   @Override
 104   public void reset() {
 105     reader.reset();
 106   }
 107
 108   @Override
 109   public void reset(long index) {
 110     reader.reset(index);
 111   }
 112
 113   @Override
 114   public void close() {
 115     reader.close();
 116     segment.closeReader(this);
 117   }
 118 }