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