/*-
 * Copyright (C) 2002, 2018, Oracle and/or its affiliates. All rights reserved.
 *
 * This file was distributed by Oracle as part of a version of Oracle Berkeley
 * DB Java Edition made available at:
 *
 * http://www.oracle.com/technetwork/database/database-technologies/berkeleydb/downloads/index.html
 *
 * Please see the LICENSE file included in the top-level directory of the
 * appropriate version of Oracle Berkeley DB Java Edition for a copy of the
 * license and additional information.
 */

package com.sleepycat.je.log;

import java.io.FileNotFoundException;
import java.nio.ByteBuffer;
import java.util.logging.Logger;

import com.sleepycat.je.DatabaseException;
import com.sleepycat.je.EnvironmentFailureException;
import com.sleepycat.je.config.EnvironmentParams;
import com.sleepycat.je.dbi.DbConfigManager;
import com.sleepycat.je.dbi.EnvironmentFailureReason;
import com.sleepycat.je.dbi.EnvironmentImpl;
import com.sleepycat.je.utilint.DbLsn;
import com.sleepycat.je.utilint.LoggerUtils;
import com.sleepycat.je.utilint.VLSN;

/**
 * A FileReader is an abstract class that traverses the log files, reading-in
 * chunks of the file at a time. The class provides a iterator interface, via
 * its readNextEntry() method. Concrete subclasses must (a) provide the public
 * methods that allow their users to examine the contents of the logrec that
 * the iterator is currently positioned at, and (b) implement the (non-public)
 * isTarget() and processEntry() methods that may filter-out logrecs that do
 * not need to be seen by the caller of readNextEntry(), or perform specific
 * actions before each logrec is "returned" to the caller of readNextEntry().
 */
public abstract class FileReader {

    protected final EnvironmentImpl envImpl;

    protected final FileManager fileManager;

    /*
     * The ReadWindow is a data buffer that acts as a sliding window view
     * of the log. It is positioned against the log and filled up with data.
     */
    protected final ReadWindow window;

    /* 
     * For piecing together a log entry that is read from multiple read buffer
     * calls.
     */
    private ByteBuffer saveBuffer;   

    private final boolean singleFile;// if true, do not read across files

    /*
     * true if at end of the log.
     * TODO: assess whether this is redundant  with the EOFException, and
     * could be streamlined.
     */
    protected boolean eof;

    /* if true, we're reading forward; otherwise backwards */
    protected final boolean forward;   

    /* num entries we've seen */
    private int nRead;

    /* The log entry header for the entry that was just read. */
    protected LogEntryHeader currentEntryHeader;

    /*
     * The log entry before the current entry. In general,
     * currentEntryPrevOffset is the same as
     * currentEntryHeader.getPrevOffset(), but it's initialized and used before
     * a header is read. Only used for backward scanning.
     */
    protected long currentEntryPrevOffset;

    /*
     * nextEntryOffset is used to set the currentEntryOffset after we've read
     * an entry. Only used for forward scanning.
     */
    protected long currentEntryOffset;
    protected long nextEntryOffset;

    protected long startLsn;  // We start reading from this LSN.
    private final long finishLsn; // If going backwards, read up to this LSN.

    /* For checking checksum on the read. */
    protected ChecksumValidator cksumValidator;

    private boolean doChecksumOnRead;       // Validate checksums
    private boolean alwaysValidateChecksum; // Validate for all entry types

    protected final Logger logger;

    public FileReader(EnvironmentImpl envImpl,
                      int readBufferSize,
                      boolean forward,
                      long startLsn,
                      Long singleFileNumber,
                      long endOfFileLsn,
                      long finishLsn)
        throws DatabaseException {

        this(envImpl, readBufferSize, forward,
             startLsn, singleFileNumber, endOfFileLsn, finishLsn,
             envImpl.getLogManager().getChecksumOnRead());
    }

    public FileReader(EnvironmentImpl envImpl,
                      int readBufferSize,
                      boolean forward,
                      long startLsn,
                      Long singleFileNumber,
                      long endOfFileLsn,
                      long finishLsn,
                      boolean doChecksumOnRead)
        throws DatabaseException {

        this.envImpl = envImpl;
        this.fileManager = envImpl.getFileManager();
        this.singleFile = (singleFileNumber != null);
        this.forward = forward;

        this.doChecksumOnRead = doChecksumOnRead;
        if (this.doChecksumOnRead) {
            cksumValidator = new ChecksumValidator(envImpl);
        }

        window = makeWindow(readBufferSize);
        saveBuffer = ByteBuffer.allocate(readBufferSize);

        /* stats */
        nRead = 0;

        /* Determine the starting position. */
        this.startLsn = startLsn;
        this.finishLsn = finishLsn;

        logger = envImpl.getLogger();

        initStartingPosition(endOfFileLsn, singleFileNumber);
    }

    /**
     * May be overridden by other FileReaders.
     * @throws DatabaseException 
     */
    protected ReadWindow makeWindow(int readBufferSize) 
        throws DatabaseException {
        
        return new ReadWindow(readBufferSize, envImpl);
    }

    /**
     * Helper for determining the starting position and opening up a file at
     * the desired location.
     */
    protected void initStartingPosition(long endOfFileLsn,
                                        Long ignoreSingleFileNumber) {
        eof = false;
        if (forward) {

            /*
             * Start off at the startLsn. If that's null, start at the
             * beginning of the log. If there are no log files, set eof.
             */
            if (startLsn != DbLsn.NULL_LSN) {
                window.initAtFileStart(startLsn);
            } else {
                Long firstNum = fileManager.getFirstFileNum();
                if (firstNum == null) {
                    eof = true;
                } else {
                    window.initAtFileStart(DbLsn.makeLsn(firstNum, 0));
                }
            }

            /*
             * After we read the first entry, the currentEntry will point here.
             */
            nextEntryOffset = window.getEndOffset();
        } else {

            /*
             * Make the read buffer look like it's positioned off the end of
             * the file. Initialize the first LSN we want to read. When
             * traversing the log backwards, we always start at the very end.
             */
            assert startLsn != DbLsn.NULL_LSN;
            window.initAtFileStart(endOfFileLsn);

            /*
             * currentEntryPrevOffset points to the entry we want to start out
             * reading when going backwards. If it's 0, the entry we want to
             * read is in a different file.
             */
            if (DbLsn.getFileNumber(startLsn) ==
                DbLsn.getFileNumber(endOfFileLsn)) {
                currentEntryPrevOffset = DbLsn.getFileOffset(startLsn);
            } else {
                currentEntryPrevOffset = 0;
            }
            currentEntryOffset = DbLsn.getFileOffset(endOfFileLsn);
        }
    }

    /**
     * Whether to always validate the checksum, even for non-target entries.
     */
    public void setAlwaysValidateChecksum(boolean validate) {
        alwaysValidateChecksum = validate;
    }

    /**
     * @return the number of entries processed by this reader.
     */
    public int getNumRead() {
        return nRead;
    }

    public long getNRepeatIteratorReads() {
        return window.getNRepeatIteratorReads();
    }

    /**
     * Get LSN of the last entry read.
     */
    public long getLastLsn() {
        return DbLsn.makeLsn(window.currentFileNum(), currentEntryOffset);
    }

    /**
     * Returns the VLSN of the last entry read, or null if the entry has no
     * VLSN (not replicated and has no migrated VLSN).
     */
    public VLSN getLastVlsn() {
        return currentEntryHeader.getVLSN();
    }

    /**
     * Returns the total size (including header) of the last entry read.
     */
    public int getLastEntrySize() {
        return currentEntryHeader.getEntrySize();
    }

    /**
     * Returns the log version of the last entry read.
     */
    public int getLogVersion() {
        return currentEntryHeader.getVersion();
    }

    /**
     * Scans the log files until either it has reached the end of the log or
     * has hit an invalid portion.
     *
     * @return true if an element has been read, false at end-of-log.
     *
     * @throws EnvironmentFailureException if a ChecksumException,
     * FileNotFoundException, or another internal problem occurs.
     */
    public boolean readNextEntry() {
        try {
            return readNextEntryAllowExceptions();
        } catch (FileNotFoundException e) {
            throw new EnvironmentFailureException(
                envImpl, EnvironmentFailureReason.LOG_FILE_NOT_FOUND, e);
        } catch (ChecksumException e) {
            throw new EnvironmentFailureException(
                envImpl, EnvironmentFailureReason.LOG_CHECKSUM, e);
        }
    }

    /**
     * Variant of readNextEntry that throws FileNotFoundException and
     * ChecksumException, rather than wrapping them in an
     * EnvironmentFailureException and invalidating the enviornment.  This
     * allows users of this class (see cleaner.FileProcessor), and subclasses
     * that override readNextEntry (see ScavengerFileReader and
     * LastFileReader), to handle these exceptions specially.
     */
    public final boolean readNextEntryAllowExceptions()
        throws FileNotFoundException, ChecksumException {

        boolean foundEntry = false;
        long savedCurrentEntryOffset = currentEntryOffset;
        long savedNextEntryOffset = nextEntryOffset;

        try {
            while ((!eof) && (!foundEntry)) {

                /* Read the invariant portion of the next header. */
                getLogEntryInReadBuffer();
                ByteBuffer dataBuffer =
                    readData(LogEntryHeader.MIN_HEADER_SIZE,
                             true); // collectData

                readBasicHeader(dataBuffer);

                boolean isTarget;
                boolean isChecksumTarget;
                boolean collectData;

                if (currentEntryHeader.isVariableLength()) {

                    /*
                     * For all variable length entries, init the checksum w/the
                     * invariant portion of the header, before we know whether
                     * the entry is a target for this reader.  This has
                     * to be done before we read the variable portion of the
                     * header, because readData() only guarantees that it
                     * returns a dataBuffer that contains the next bytes that
                     * are needed, and has no guarantee that it holds any bytes
                     * that were previously read. The act of calling
                     * readData() to obtain the optional portion may reset the
                     * dataBuffer, and nudge the invariant part of the header
                     * out of the buffer returned by readData()
                     */
                    if (currentEntryHeader.hasChecksum()) {
                        startChecksum(dataBuffer);
                    }

                    int optionalPortionLen =
                        currentEntryHeader.getVariablePortionSize();

                    /* Load the optional part of the header into a buffer. */
                    dataBuffer = readData(optionalPortionLen, true);

                    /*
                     * Add to checksum while the buffer is positioned at
                     * the start of the new bytes.
                     */
                    if (currentEntryHeader.hasChecksum()) {
                        addToChecksum(dataBuffer, optionalPortionLen);
                    }

                    /* Now read the optional bytes. */
                    currentEntryHeader.readVariablePortion(dataBuffer);
                }

                /*
                 * We've read the header of the next logrec. Move up our
                 * offsets if we're moving forward. If we're moving
                 * backwards, we set our offset before we read the header,
                 * because we knew where the entry started.
                 */
                if (forward) {
                    currentEntryOffset = nextEntryOffset;
                    nextEntryOffset +=
                        currentEntryHeader.getSize() +    // header size
                        currentEntryHeader.getItemSize(); // item size
                }
                
                try {
                    isTarget = isTargetEntry();

                    isChecksumTarget = currentEntryHeader.hasChecksum() &&
                        (isTarget || alwaysValidateChecksum);

                    if (!currentEntryHeader.isVariableLength()) {
                        startChecksum(dataBuffer, isChecksumTarget);
                    }

                    collectData =
                        (isChecksumTarget && doChecksumOnRead) || isTarget;

                    /*
                     * Read in the body of the next entry. Note that even if
                     * this isn't a targeted entry, we have to move the buffer
                     * position along.
                     */
                    dataBuffer = readData(currentEntryHeader.getItemSize(),
                                          collectData);
                } catch (Throwable e) {
                    if (forward) {
                        currentEntryOffset = savedCurrentEntryOffset;
                        nextEntryOffset = savedNextEntryOffset;
                    }
                    throw e;
                }

                /* Validate the log entry checksum. */
                validateChecksum(dataBuffer, isChecksumTarget);
                
                if (isTarget) {

                    /*
                     * For a target entry, call the subclass reader's
                     * processEntry method to do whatever we need with the
                     * entry.  It returns true if this entry is one that should
                     * be returned.  Note that some entries, although targeted
                     * and read, are not returned.
                     */
                    if (processEntry(dataBuffer)) {
                        foundEntry = true;
                        nRead++;
                    }
                } else if (collectData) {

                    /*
                     * For a non-target entry that was validated, the buffer is
                     * positioned at the start of the entry; skip over it.
                     */
                    skipEntry(dataBuffer);
                }
            }
        } catch (EOFException e) {
            eof = true;
        } catch (DatabaseException e) {
            eof = true;
            /* Report on error. */
            reportProblem(e);
            throw e;
        }

        return foundEntry;
    }

    /**
     * May be called by processEntry when it determines that the entry does not
     * need to be read/de-serialized.
     */
    protected void skipEntry(ByteBuffer entryBuffer) {
        entryBuffer.position(
            entryBuffer.position() +
            currentEntryHeader.getItemSize());
    }

    private void reportProblem(Exception e) {

        StringBuilder sb = new StringBuilder();
        sb.append("Halted log file reading at file 0x").
            append(Long.toHexString(window.currentFileNum())).
            append(" offset 0x").
            append(Long.toHexString(nextEntryOffset)).
            append(" offset(decimal)=").
            append(nextEntryOffset).
            append(" prev=0x").
            append(Long.toHexString(currentEntryPrevOffset));

            if (currentEntryHeader != null) {
                LogEntryType problemType =
                    LogEntryType.findType(currentEntryHeader.getType());
            sb.append(":\nentry=").
                append(problemType).
                append("type=").
                append(currentEntryHeader.getType()).
                append(",version=").
                append(currentEntryHeader.getVersion()).
                append(")\nprev=0x").
                append(Long.toHexString(currentEntryPrevOffset)).
                append("\nsize=").
                append(currentEntryHeader.getItemSize()).
                append("\nNext entry should be at 0x").
                append(Long.toHexString(nextEntryOffset +
                                               currentEntryHeader.getSize() +
                                        currentEntryHeader.getItemSize()));
        }

        LoggerUtils.traceAndLogException
            (envImpl, "FileReader", "readNextEntry", sb.toString(), e);
    }

    /**
     * Make sure that the start of the target log entry is in the header. 
     */
    private void getLogEntryInReadBuffer()
        throws ChecksumException,
               EOFException,
               FileNotFoundException,
               DatabaseException {

        /*
         * If we're going forward, because we read every byte sequentially,
         * we're always sure the read buffer is positioned at the right spot.
         * If we go backwards, we need to jump the buffer position. These
         * methods may be overridden by subclasses.
         */
        if (forward) {
            setForwardPosition();
        } else {
            setBackwardPosition();
        }
    }

    /**
     * Ensure that the next target is in the window. The default behavior is
     * that the next target is the next, following entry, so we can assume that
     * it's in the window.  All we have to do is to check if we've gone past
     * the specified end point.
     * @throws DatabaseException 
     * @throws FileNotFoundException 
     * @throws ChecksumException 
     */
    protected void setForwardPosition() 
        throws EOFException,
               DatabaseException, 
               ChecksumException, 
               FileNotFoundException {

        if (finishLsn != DbLsn.NULL_LSN) {
            /* The next log entry has passed the end LSN. */
            long nextLsn = DbLsn.makeLsn(window.currentFileNum(),
                                         nextEntryOffset);
            if (DbLsn.compareTo(nextLsn, finishLsn) >= 0) {
                throw new EOFException();
            }
        }
    }

    /**
     * Ensure that the next target is in the window. The default behavior is
     * that the next target is the next previous entry.
     * @throws DatabaseException 
     */
    protected void setBackwardPosition() 
        throws ChecksumException,
               FileNotFoundException,
               EOFException,
               DatabaseException {

        /*
         * currentEntryPrevOffset is the entry before the current entry.
         * currentEntryOffset is the entry we just read (or the end of the
         * file if we're starting out.
         */
        if ((currentEntryPrevOffset != 0) &&
            window.containsOffset(currentEntryPrevOffset)) {

            /* The next log entry has passed the start LSN. */
            long nextLsn = DbLsn.makeLsn(window.currentFileNum(),
                                         currentEntryPrevOffset);
            if (finishLsn != DbLsn.NULL_LSN) {
                if (DbLsn.compareTo(nextLsn, finishLsn) == -1) {
                    throw new EOFException("finish=" + 
                                           DbLsn.getNoFormatString(finishLsn) +
                                           "next=" + 
                                           DbLsn.getNoFormatString(nextLsn));
                }
            }

            /* This log entry starts in this buffer, just reposition. */
            window.positionBuffer(currentEntryPrevOffset);
        } else {

            /*
             * The start of the log entry is not in this read buffer so
             * we must fill the buffer again. 
             *
             * 1) The target log entry is in a different file from the
             * current window's file. Move the window to the previous
             * file and start the read from the target LSN.
             *
             * 2) The target log entry is the same file but the log entry
             * is larger than the read chunk size. Start the next read
             * buffer from the target LSN. It's going to take multiple
             * reads to get the log entry, and we might as well get as
             * much as possible.
             *
             * 3) In the same file, and the log entry fits within one
             * read buffer. Try to position the next buffer chunk so the
             * target entry is held within the buffer, all the way at the
             * end. That way, since we're reading backwards, there will be
             * more buffered data available for following reads.
             */
            long nextFile;
            long nextWindowStart;
            long nextTarget;

            if (currentEntryPrevOffset == 0) {
                /* Case 1: Go to another file. */
                currentEntryPrevOffset = fileManager.getFileHeaderPrevOffset
                    (window.currentFileNum());

                Long prevFileNum =
                    fileManager.getFollowingFileNum(window.currentFileNum(),
                                                    false);
                if (prevFileNum == null) {
                    throw new EOFException("No file following " +
                                           window.currentFileNum());
                }

                /*
                 *  Check finishLSN  before proceeding, in case we should stop
                 *  the search before attempting to set the file reader to a
                 *  position in the previous file. In  [#22407] we threw a
                 *  spurious EFE complaining that we cannot read backwards over
                 *  a cleaned file because the previous file had  been cleaned
                 *  away.
                 */
                if (finishLsn != DbLsn.NULL_LSN &&
                    prevFileNum < DbLsn.getFileNumber(finishLsn)) {
                    throw new EOFException(
                        "finish=" + DbLsn.getNoFormatString(finishLsn) +
                        " nextFile=0x" + Long.toHexString(prevFileNum));
                }

                if (window.currentFileNum() - prevFileNum.longValue() != 1) {
                    handleGapInBackwardsScan(prevFileNum);
                }

                nextFile = prevFileNum;
                nextWindowStart = currentEntryPrevOffset;
                nextTarget = currentEntryPrevOffset;
            } else if ((currentEntryOffset - currentEntryPrevOffset) >
                       window.capacity()) {

                /*
                 * Case 2: The entry is in the same file, but is bigger
                 * than one buffer. Position it at the front of the buffer.
                 */
                nextFile = window.currentFileNum();
                nextWindowStart = currentEntryPrevOffset;
                nextTarget = currentEntryPrevOffset;
            } else {

                /* 
                 * Case 3: In same file, but not in this buffer. The target
                 * entry will fit in one buffer.
                 */
                nextFile = window.currentFileNum();
                long newPosition = currentEntryOffset -
                    window.capacity();
                nextWindowStart = (newPosition < 0) ? 0 : newPosition;
                nextTarget = currentEntryPrevOffset;
            }

            /* The next log entry has passed the start LSN. */
            long nextLsn = DbLsn.makeLsn(nextFile,
                                         currentEntryPrevOffset);
            if (finishLsn != DbLsn.NULL_LSN) {
                if (DbLsn.compareTo(nextLsn, finishLsn) == -1) {
                    throw new EOFException("finish=" + 
                                           DbLsn.getNoFormatString(finishLsn) +
                                           " next=" +
                                           DbLsn.getNoFormatString(nextLsn));
                }
            }

            window.slideAndFill
                (nextFile, nextWindowStart, nextTarget, forward);
        }

        /* The current entry will start at this offset. */
        currentEntryOffset = currentEntryPrevOffset;
    }

    /**
     * Read the basic log entry header, leaving the buffer mark at the
     * beginning of the checksummed header data.
     */
    private void readBasicHeader(ByteBuffer dataBuffer)
        throws ChecksumException, DatabaseException  {

        /* Read the header for this entry. */
        currentEntryHeader = new LogEntryHeader(
            dataBuffer, window.logVersion, window.getCurrentLsn());

        /*
         * currentEntryPrevOffset is a separate field, and is not obtained
         * directly from the currentEntryHeader, because it is initialized and
         * used before any log entry was read.
         */
        currentEntryPrevOffset = currentEntryHeader.getPrevOffset();
    }

    /**
     * Reset the checksum validator and add the new header bytes. Assumes that
     * the data buffer is positioned just past the end of the invariant
     * portion of the log entry header.
     * @throws DatabaseException
     */
    private void startChecksum(ByteBuffer dataBuffer) 
        throws ChecksumException {

        startChecksum(dataBuffer, true  /* isChecksumTarget */);
    }

    private void startChecksum(ByteBuffer dataBuffer,
                               boolean isChecksumTarget)
        throws ChecksumException {

        if (!doChecksumOnRead) {
            return;
        }

        if (!isChecksumTarget) {
            return;
        }

        /* Clear out any previous data. */
        cksumValidator.reset();

        int originalPosition = dataBuffer.position();
        if (currentEntryHeader.isInvisible()) {

            /* 
             * Turn off invisibility so that the checksum will succeed. When
             * entries are made invisible, the checksum is not adjusted. Note
             * that the dataBuffer can leave the invisible bit transformed,
             * because the header has already been initialized, and this data
             * will never be read again.
             */
            LogEntryHeader.turnOffInvisible(dataBuffer, originalPosition -
                                            LogEntryHeader.MIN_HEADER_SIZE);
        }

        /* Position the buffer at the start of the data, after the checksum. */
        int headerSizeMinusChecksum =
            currentEntryHeader.getInvariantSizeMinusChecksum();
        int entryTypeStart = originalPosition - headerSizeMinusChecksum;
        dataBuffer.position(entryTypeStart);

        /* Load the validate with the header bytes. */
        cksumValidator.update(dataBuffer, headerSizeMinusChecksum);

        /* Move the data buffer back to the original position. */
        dataBuffer.position(originalPosition);
    }

    private void addToChecksum(ByteBuffer dataBuffer, int length) 
        throws ChecksumException {

        if (!doChecksumOnRead) {
            return;
        }

        cksumValidator.update(dataBuffer, length);
    }

    /**
     * Add the entry bytes to the checksum and check the value.  This method
     * must be called with the buffer positioned at the start of the entry.
     */
    private void validateChecksum(ByteBuffer dataBuffer, 
                                  boolean isChecksumTarget)
        throws ChecksumException {

        if (!doChecksumOnRead) {
            return;
        }

        if (!isChecksumTarget) {
            return;
        }

        cksumValidator.update(dataBuffer, currentEntryHeader.getItemSize());
        cksumValidator.validate(currentEntryHeader.getChecksum(),
                                window.currentFileNum(),
                                currentEntryOffset);
    }

    /**
     * Try to read a specified number of bytes.
     * @param amountToRead is the number of bytes we need
     * @param collectData is true if we need to actually look at the data.
     *  If false, we know we're skipping this entry, and all we need to
     *  do is to count until we get to the right spot.
     * @return a byte buffer positioned at the head of the desired portion,
     * or null if we reached eof.
     */
    private ByteBuffer readData(int amountToRead, boolean collectData)
        throws ChecksumException,
               EOFException,
               FileNotFoundException,
               DatabaseException {

        int alreadyRead = 0;
        ByteBuffer completeBuffer = null;
        saveBuffer.clear();

        while ((alreadyRead < amountToRead) && !eof) {

            int bytesNeeded = amountToRead - alreadyRead;
            if (window.hasRemaining()) {

                /* There's data in the window, process it. */
                if (collectData) {

                    /*
                     * Save data in a buffer for processing.
                     */
                    if ((alreadyRead > 0) ||
                        (window.remaining() < bytesNeeded)) {

                        /* We need to piece an entry together. */
                        copyToSaveBuffer(bytesNeeded);
                        alreadyRead = saveBuffer.position();
                        completeBuffer = saveBuffer;
                    } else {

                        /* A complete entry is available in this buffer. */
                        completeBuffer = window.getBuffer();
                        alreadyRead = amountToRead;
                    }
                } else {

                    /*
                     * We're not processing the data, so need to save it. just
                     * move buffer positions.
                     */
                    int positionIncrement =
                        (window.remaining() > bytesNeeded) ?
                        bytesNeeded : window.remaining();

                    alreadyRead += positionIncrement;
                    window.incrementBufferPosition(positionIncrement);
                    completeBuffer = window.getBuffer();
                }
            } else {

                /*
                 * Look for more data.
                 */
                if (window.fillNext(singleFile, bytesNeeded)) {
                    /* This call to fillNext slid the window to a new file. */
                    nextEntryOffset = 0;
                }
            }
        }

        /* Flip the save buffer just in case we've been accumulating in it. */
        saveBuffer.flip();

        return completeBuffer;
    }
    
    /* Try to skip over a specified number of bytes. */
    public void skipData(int amountToSkip) 
        throws ChecksumException,
               EOFException,
               FileNotFoundException,
               DatabaseException {
               
        try {
            readData(amountToSkip, false);
        } catch (DatabaseException e) {
            reportProblem(e);
            throw e;
        }    
    }

    /**
     * Copy the required number of bytes into the save buffer.
     */
    private void copyToSaveBuffer(int bytesNeeded) {
        /* How much can we get from this current read buffer? */
        int bytesFromThisBuffer;

        if (bytesNeeded <= window.remaining()) {
            bytesFromThisBuffer = bytesNeeded;
        } else {
            bytesFromThisBuffer = window.remaining();
        }

        /* Gather it all into this save buffer. */
        ByteBuffer temp;

        /* Make sure the save buffer is big enough. */
        if (saveBuffer.capacity() - saveBuffer.position() <
            bytesFromThisBuffer) {
            /* Grow the save buffer. */
            temp = ByteBuffer.allocate(saveBuffer.capacity() +
                                       bytesFromThisBuffer);
            saveBuffer.flip();
            temp.put(saveBuffer);
            saveBuffer = temp;
        }

        /*
         * Bulk copy only the required section from the read buffer into the
         * save buffer. We need from readBuffer.position() to
         * readBuffer.position() + bytesFromThisBuffer
         */
        temp = window.getBuffer().slice();
        temp.limit(bytesFromThisBuffer);
        saveBuffer.put(temp);
        window.incrementBufferPosition(bytesFromThisBuffer);
    }

    /**
     * Returns the number of reads since the last time this method was called.
     */
    public int getAndResetNReads() {
        return window.getAndResetNReads();
    }

    /**
     * This method is called by readNextEntry() after the header of the current
     * logrec has been de-serialized, but not the body. Based on header info
     * only, it may perform some actions and then decide whether the rest of
     * the logrec should be de-serialized or just skipped.
     * 
     * @return true if this reader should process the current logrec further,
     * via the processEntry() method. A logrec must be passed to processEntry
     * if the full logrec (not just the header) must be de-serialized for
     * further processing. Return false if no further processing is needed,
     * in which case the current logrec will be skipped (i.e, not returned
     * to the caller of readNextEntry().
     *
     * @throws DatabaseException from subclasses.
     */
    protected boolean isTargetEntry()
        throws DatabaseException {

        return true;
    }

    /**
     * Each file reader implements this method to process the entry data.
     *
     * @param entryBuffer A ByteBuffer that the logrec data and is positioned
     * at the start of the logrec body (i.e., just after the logrec header).
     *
     * @return true if this entry should be returned to the caller of
     * readNextEntry().
     */
    protected abstract boolean processEntry(ByteBuffer entryBuffer)
        throws DatabaseException;

    /**
     * Never seen by user, used to indicate that the file reader should stop.
     */
    @SuppressWarnings("serial")
    public static class EOFException extends Exception {
        public EOFException() {
            super();
        }

        /* 
         * @param message The message is used to hold debugging 
         * information.
         */
        public EOFException(String message) {
            super(message);
        }
    }

    /**
     * @return true if the current entry is part of replication stream.
     */
    public boolean entryIsReplicated() {

        if (currentEntryHeader == null) {
            throw EnvironmentFailureException.unexpectedState
                ("entryIsReplicated should not be used before reader is " +
                 "initialized");
        } 
        return currentEntryHeader.getReplicated();
    }

    /**
     * TBW
     */
    protected void handleGapInBackwardsScan(long prevFileNum) {
        throw new EnvironmentFailureException
            (envImpl,
             EnvironmentFailureReason.LOG_INTEGRITY,
             "Cannot read backward over cleaned file" +
             " from 0x" + Long.toHexString(window.currentFileNum()) +
             " to 0x" + Long.toHexString(prevFileNum));
    }

    /**
     * A ReadWindow provides a swath of data read from the JE log. 
     */
    protected static class ReadWindow {

        /*
         * fileNum, startOffset and endOffset indicate how the read buffer maps
         * to the JE log. For example, if the read buffer size is 200 and the
         * read buffer was filled from file 9, starting at byte 100, then:
         *          fileNum = 9
         *          startOffset = 100
         *          endOffset = 300
         * Note that the end point is not inclusive; endOffset is > the
         * readBuffer's end.
         */
        private long fileNum;      // file number we're pointing to
        private int logVersion;    // log version for fileNum/readBuffer
        protected long startOffset;// file offset that maps to buf start
        protected long endOffset;  // file offset that maps to buf end
        protected ByteBuffer readBuffer;   // buffer for reading from the file

        /* read buffer can't grow larger than this */
        private final int maxReadBufferSize;   

        protected final EnvironmentImpl envImpl;
        protected final FileManager fileManager;

        /*
         * The number of times we've tried to read in a log entry that was too
         * large for the read buffer.
         */
        private long nRepeatIteratorReads;

        /* Number of reads since the last time getAndResetNReads was called. */
        private int nReadOperations;

        protected ReadWindow(int readBufferSize, EnvironmentImpl envImpl) {
            DbConfigManager configManager = envImpl.getConfigManager();
            maxReadBufferSize =
                configManager.getInt(EnvironmentParams.LOG_ITERATOR_MAX_SIZE);
            this.envImpl = envImpl;
            fileManager = envImpl.getFileManager();

            readBuffer = ByteBuffer.allocate(readBufferSize);
            readBuffer.flip();
        }

        /* 
         * Position this window at this LSN, but leave it empty, it has no data
         * yet.
         */
        public void initAtFileStart(long startLsn) {
            setFileNum(DbLsn.getFileNumber(startLsn),
                       LogEntryType.UNKNOWN_FILE_HEADER_VERSION);
            startOffset = DbLsn.getFileOffset(startLsn);
            endOffset = startOffset;
        }

        public long getEndOffset() {
            return endOffset;
        }        

        /**
         * Ensure that whenever we change the fileNum, the logVersion is also
         * updated.  The fileNum and logVersion fields should be kept private.
         */
        protected void setFileNum(final long fileNum, final int logVersion) {
            this.fileNum = fileNum;
            this.logVersion = logVersion;
        }

        public long currentFileNum() {
            return fileNum;
        }

        /* Return true if this offset is contained with the readBuffer. */
        boolean containsOffset(long targetOffset) {
            return (targetOffset >= startOffset) &&
                (targetOffset < endOffset);
        }

        /* Return true if this lsn  is contained with the readBuffer. */
        public boolean containsLsn(long targetFileNumber, long targetOffset) {
            return ((fileNum == targetFileNumber) &&
                    containsOffset(targetOffset));
        }

        /* Position the readBuffer to the targetOffset. */
        public void positionBuffer(long targetOffset) {

            assert containsOffset(targetOffset) : this + " doesn't contain " +
                DbLsn.getNoFormatString(targetOffset);

            readBuffer.position((int) (targetOffset - startOffset));
        }

        /* Move the readBuffer position up by the given increment. */
        void incrementBufferPosition(int increment) {
            int currentPosition = readBuffer.position();
            readBuffer.position(currentPosition + increment);
        }

        /* 
         * Reposition to the specified file, and fill starting at
         * startOffset. Position the window's buffer to point at the log entry
         * indicated by targetOffset
         */
        public void slideAndFill(long windowfileNum, 
                                 long windowStartOffset, 
                                 long targetOffset,
                                 boolean forward)
            throws ChecksumException,
                   FileNotFoundException,
                   DatabaseException {

            FileHandle fileHandle = fileManager.getFileHandle(windowfileNum);
            try {
                startOffset = windowStartOffset;
                setFileNum(windowfileNum, fileHandle.getLogVersion());
                boolean foundData = fillFromFile(fileHandle, targetOffset);

                /*
                 * When reading backwards, we need to guarantee there is no log
                 * gap, throws out an EnvironmentFailreException if it exists.
                 */
                if (!foundData && !forward) {
                    throw EnvironmentFailureException.unexpectedState
                        ("Detected a log file gap when reading backwards. " + 
                         "Target position = " + DbLsn.getNoFormatString
                         (DbLsn.makeLsn(windowfileNum, targetOffset)) +
                         " starting position = " + DbLsn.getNoFormatString
                         (DbLsn.makeLsn(windowfileNum, windowStartOffset)) +
                         " end position = " + DbLsn.getNoFormatString
                         (DbLsn.makeLsn(windowfileNum, endOffset)));
                }
            } finally {
                fileHandle.release();
            }
        }

        /**
         * Fill up the read buffer with more data, moving along to the
         * following file (next largest number) if needed.
         * @return true if the fill moved us to a new file.
         */
        protected boolean fillNext(boolean singleFile, int bytesNeeded)
            throws ChecksumException,
                   FileNotFoundException,
                   EOFException,
                   DatabaseException {

            adjustReadBufferSize(bytesNeeded);

            FileHandle fileHandle = null;
            try {
                /* Get a file handle to read in more log. */
                fileHandle = fileManager.getFileHandle(fileNum);

                /*
                 * Check to see if we've come to the end of the file.  If so,
                 * get the next file.
                 */
                startOffset = endOffset;
                if (fillFromFile(fileHandle, startOffset)) {
                    /* 
                     * Successfully filled the read buffer, but didn't move to 
                     * a new file. 
                     */
                    return false; 
                }

                /* This file is done -- can we read in the next file? */
                if (singleFile) {
                    throw new EOFException("Single file only");
                }

                Long nextFile = 
                    fileManager.getFollowingFileNum(fileNum, 
                                                    true /* forward */);

                if (nextFile == null) {
                    throw new EOFException();
                }

                fileHandle.release();
                fileHandle = null;
                /*
                 * TODO: This doesn't work if file is deleted after calling
                 * getFollowingFileNum above and calling getFileHandle below.
                 * That can happen to due to deletion of reserved files by the
                 * cleaner or eraser. We need a retry loop here that catches
                 * FileNotFoundException and calls getFollowingFileNum again.
                 * The retry method probably belongs in FileManager.
                 */
                fileHandle = fileManager.getFileHandle(nextFile);
                setFileNum(nextFile, fileHandle.getLogVersion());
                startOffset = 0;
                fillFromFile(fileHandle, 0);
                return true;
            } finally {
                if (fileHandle != null) {
                    fileHandle.release();
                }
            }
        }

        /* 
         * Assume that the window is properly positioned. Try to fill the read
         * buffer with data from this file handle, starting at the location
         * indicated by the starting offset field. If this file contains more
         * data, return true. If this file doesn't contain more data, return
         * false.
         *
         * In all cases, leave the the read buffer pointing at the target
         * offset and in a state that's ready to support reads, even if there
         * is nothing in the buffer. Note that the target offset, which may not
         * be the same as starting offset.
         * @return true if more data was read, false if not.         
         */
        protected boolean fillFromFile(FileHandle fileHandle, 
                                       long targetOffset) 
            throws DatabaseException {

            boolean foundData = false;
            readBuffer.clear();
            if (fileManager.readFromFile(fileHandle.getFile(), 
                                         readBuffer,
                                         startOffset,
                                         fileHandle.getFileNum(),
                                         false /* dataKnownToBeInFile */)) {
                foundData = true;
                nReadOperations += 1;
                /*
                 * Ensure that fileNum and logVersion are in sync.  setFileNum
                 * handles changes in the file number.  But we must also update
                 * the logVersion here to handle the first read after we
                 * initialize fileNum and logVersion is unknown.
                 */
                logVersion = fileHandle.getLogVersion();
            }

            /* 
             * In all cases, setup read buffer for valid reading. If the buffer
             * has no data, it will be positioned at the beginning, and will be
             * able to correctly return the fact that there is no data present.
             */

            endOffset = startOffset + readBuffer.position();
            readBuffer.flip();
            readBuffer.position((int) (targetOffset - startOffset));
            return foundData;
        }

        /**
         * Change the read buffer size if we start hitting large log entries so
         * we don't get into an expensive cycle of multiple reads and piecing
         * together of log entries.  
         */
        protected void adjustReadBufferSize(int amountToRead) {

            int readBufferSize = readBuffer.capacity();

            /* 
             * We need to read something larger than the current buffer
             * size. 
             */
            if (amountToRead > readBufferSize) {

                /* We're not at the max yet. */
                if (readBufferSize < maxReadBufferSize) {

                    /*
                     * Make the buffer the minimum of amountToRead or a
                     * maxReadBufferSize.
                     */
                    if (amountToRead < maxReadBufferSize) {
                        readBufferSize = amountToRead;
                        /* Make it a multiple of 1K. */
                        int remainder = readBufferSize % 1024;
                        readBufferSize += 1024 - remainder;
                        readBufferSize = Math.min(readBufferSize,
                                                  maxReadBufferSize);
                    } else {
                        readBufferSize = maxReadBufferSize;
                    }
                    readBuffer = ByteBuffer.allocate(readBufferSize);
                }

                if (amountToRead > readBuffer.capacity()) {
                    nRepeatIteratorReads++;
                    envImpl.getLogManager().incRepeatIteratorReads();
                }
            }
        }

        int capacity() {
            return readBuffer.capacity();
        }

        int remaining() {
            return readBuffer.remaining();
        }

        boolean hasRemaining() {
            return readBuffer.hasRemaining();
        }

        ByteBuffer getBuffer() {
            return readBuffer;
        }

        /**
         * Returns the number of reads since the last time this method was
         * called.
         */
        int getAndResetNReads() {
            int tmp = nReadOperations;
            nReadOperations = 0;
            return tmp;
        }

        long getNRepeatIteratorReads() {
            return nRepeatIteratorReads;
        }

        @Override
        public String toString() {
            StringBuilder sb = new StringBuilder();
            long start = DbLsn.makeLsn(fileNum, startOffset);
            long end = DbLsn.makeLsn(fileNum, endOffset);
            sb.append("window covers ");
            sb.append(DbLsn.getNoFormatString(start)).append(" to ");
            sb.append(DbLsn.getNoFormatString(end));
            sb.append(" positioned at ");
            long target = getCurrentLsn();
            sb.append(DbLsn.getNoFormatString(target));
            return sb.toString();
        }

        long getCurrentLsn() {
            return DbLsn.makeLsn(fileNum, startOffset + readBuffer.position());
        }
    }
}