View Javadoc

    1   /***
    2    *
    3    * Copyright 2004 Hiram Chirino
    4    *
    5    * Licensed under the Apache License, Version 2.0 (the "License");
    6    * you may not use this file except in compliance with the License.
    7    * You may obtain a copy of the License at
    8    *
    9    * http://www.apache.org/licenses/LICENSE-2.0
    10   *
    11   * Unless required by applicable law or agreed to in writing, software
    12   * distributed under the License is distributed on an "AS IS" BASIS,
    13   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    14   * See the License for the specific language governing permissions and
    15   * limitations under the License.
    16   *
    17   **/
    18  package org.activeio.journal.active;
    19
    20  import java.io.File;
    21  import java.io.IOException;
    22  import java.io.RandomAccessFile;
    23  import java.nio.ByteBuffer;
    24  import java.nio.channels.FileChannel;
    25
    26  import org.activeio.Disposable;
    27
    28  /***
    29   * Allows read/append access to a LogFile.
    30   *
    31   * @version $Revision: 1.1 $
    32   */
    33  final public class LogFile implements Disposable {
    34
    35      private final RandomAccessFile file;
    36      private final FileChannel channel;
    37
    38      /*** Prefered size. The size that the log file is set to when initilaized. */
    39      private final int initialSize;
    40
    41      /*** Where the we are in the file right now */
    42      private int currentOffset;
    43  	private boolean disposed;
    44
    45      public LogFile(File file, int initialSize) throws IOException {
    46          this.initialSize = initialSize;
    47          boolean initializationNeeeded = !file.exists();
    48          this.file = new RandomAccessFile(file, "rw");
    49          channel = this.file.getChannel();
    50          if( initializationNeeeded )
    51              resize();
    52          channel.position(0);
    53          reloadCurrentOffset();
    54      }
    55
    56      /***
    57       * To avoid doing un-needed seeks.
    58       */
    59      private void seek(int offset) throws IOException {
    60          if( offset == currentOffset ) {
    61              if( currentOffset != channel.position() )
    62                  throw new RuntimeException(" "+currentOffset+", "+channel.position() );
    63              return;
    64          }
    65          channel.position(offset);
    66          currentOffset = offset;
    67      }
    68      private void reloadCurrentOffset() throws IOException {
    69          currentOffset= (int) channel.position();
    70      }
    71      private void addToCurrentOffset(int rc) {
    72          currentOffset+=rc;
    73      }
    74
    75      public boolean loadAndCheckRecord(int offset, Record record) throws IOException {
    76
    77          try {
    78              // Read the next header
    79              seek(offset);
    80              record.readHeader(file);
    81
    82              if (Record.isChecksumingEnabled()) {
    83                  record.checksum(file);
    84              }
    85              // Load the footer.
    86              seek(offset+record.getPayloadLength()+Record.RECORD_HEADER_SIZE);
    87              record.readFooter(file);
    88
    89              addToCurrentOffset(record.getRecordLength());
    90              return true;
    91
    92          } catch (IOException e) {
    93              reloadCurrentOffset();
    94              return false;
    95          }
    96      }
    97
    98      public void resize() throws IOException {
    99          file.setLength(initialSize);
    100     }
    101
    102     public void force() throws IOException {
    103         channel.force(false);
    104     }
    105
    106     public void dispose() {
    107     	if( disposed )
    108     		return;
    109     	disposed=true;
    110         try {
    111 			this.file.close();
    112 		} catch (IOException e) {
    113 		}
    114     }
    115
    116     public void write(int offset, ByteBuffer buffer) throws IOException {
    117
    118         try {
    119
    120             int size = buffer.remaining();
    121             seek(offset);
    122             while (buffer.hasRemaining()) {
    123                 channel.write(buffer);
    124             }
    125             addToCurrentOffset(size);
    126
    127         } catch (IOException e) {
    128             reloadCurrentOffset();
    129         }
    130     }
    131
    132     public void readRecordHeader(int offset, Record record) throws IOException {
    133         seek(offset);
    134         try {
    135             record.readHeader(file);
    136         } catch ( IOException e ) {
    137             reloadCurrentOffset();
    138             throw e;
    139         }
    140         addToCurrentOffset(Record.RECORD_HEADER_SIZE);
    141     }
    142
    143     public void read(int offset, byte[] answer) throws IOException {
    144         seek(offset);
    145         file.readFully(answer);
    146         addToCurrentOffset(answer.length);
    147     }
    148 }