third-party/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 io.atomix.utils.serializer.Namespace;
  20
  21 import java.io.IOException;
  22 import java.nio.ByteBuffer;
  23 import java.nio.channels.FileChannel;
  24
  25 /**
  26  * Mappable log segment reader.
  27  */
  28 class MappableJournalSegmentReader<E> implements JournalReader<E> {
  29   private final JournalSegment<E> segment;
  30   private final FileChannel channel;
  31   private final int maxEntrySize;
  32   private final JournalIndex index;
  33   private final Namespace namespace;
  34   private JournalReader<E> reader;
  35
  36   MappableJournalSegmentReader(
  37       FileChannel channel,
  38       JournalSegment<E> segment,
  39       int maxEntrySize,
  40       JournalIndex index,
  41       Namespace namespace) {
  42     this.channel = channel;
  43     this.segment = segment;
  44     this.maxEntrySize = maxEntrySize;
  45     this.index = index;
  46     this.namespace = namespace;
  47     this.reader = new FileChannelJournalSegmentReader<>(channel, segment, maxEntrySize, index, namespace);
  48   }
  49
  50   /**
  51    * Converts the reader to a mapped reader using the given buffer.
  52    *
  53    * @param buffer the mapped buffer
  54    */
  55   void map(ByteBuffer buffer) {
  56     if (!(reader instanceof MappedJournalSegmentReader)) {
  57       JournalReader<E> reader = this.reader;
  58       this.reader = new MappedJournalSegmentReader<>(buffer, segment, maxEntrySize, index, namespace);
  59       this.reader.reset(reader.getNextIndex());
  60       reader.close();
  61     }
  62   }
  63
  64   /**
  65    * Converts the reader to an unmapped reader.
  66    */
  67   void unmap() {
  68     if (reader instanceof MappedJournalSegmentReader) {
  69       JournalReader<E> reader = this.reader;
  70       this.reader = new FileChannelJournalSegmentReader<>(channel, segment, maxEntrySize, index, namespace);
  71       this.reader.reset(reader.getNextIndex());
  72       reader.close();
  73     }
  74   }
  75
  76   @Override
  77   public long getFirstIndex() {
  78     return reader.getFirstIndex();
  79   }
  80
  81   @Override
  82   public long getCurrentIndex() {
  83     return reader.getCurrentIndex();
  84   }
  85
  86   @Override
  87   public Indexed<E> getCurrentEntry() {
  88     return reader.getCurrentEntry();
  89   }
  90
  91   @Override
  92   public long getNextIndex() {
  93     return reader.getNextIndex();
  94   }
  95
  96   @Override
  97   public boolean hasNext() {
  98     return reader.hasNext();
  99   }
 100
 101   @Override
 102   public Indexed<E> next() {
 103     return reader.next();
 104   }
 105
 106   @Override
 107   public void reset() {
 108     reader.reset();
 109   }
 110
 111   @Override
 112   public void reset(long index) {
 113     reader.reset(index);
 114   }
 115
 116   @Override
 117   public void close() {
 118     reader.close();
 119     try {
 120       channel.close();
 121     } catch (IOException e) {
 122       throw new StorageException(e);
 123     } finally {
 124       segment.closeReader(this);
 125     }
 126   }
 127 }