/*
 * Licensed to the University of California, Berkeley under one or more contributor license
 * agreements. See the NOTICE file distributed with this work for additional information regarding
 * copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance with the License. You may obtain a
 * copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software distributed under the License
 * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
 * or implied. See the License for the specific language governing permissions and limitations under
 * the License.
 */

package tachyon.master.journal;

import java.io.IOException;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import tachyon.Constants;
import tachyon.master.Master;

public final class JournalTailer {
  private static final Logger LOG = LoggerFactory.getLogger(Constants.LOGGER_TYPE);

  private final Master mMaster;
  private final Journal mJournal;
  private final JournalReader mReader;

  public JournalTailer(Master master, Journal journal) {
    mMaster = master;
    mJournal = journal;
    mReader = mJournal.getNewReader();
  }

  /**
   * Returns whether or not the tailer is currently valid.
   *
   * @return true if this tailer is valid.
   */
  public boolean isValid() {
    return mReader.isValid();
  }

  /**
   * Returns the last modified time for the checkpoint file.
   *
   * @return the last modified time of the checkpoint file in ms.
   * @throws IOException
   */
  public long getCheckpointLastModifiedTimeMs() throws IOException {
    return mReader.getCheckpointLastModifiedTimeMs();
  }

  /**
   * Load and (optionally) process the journal checkpoint file.
   *
   * @param applyToMaster if true, apply all the checkpoint events to the master. Otherwise, simply
   *        open the checkpoint file.
   * @throws IOException
   */
  public void processJournalCheckpoint(boolean applyToMaster) throws IOException {
    // Load the checkpoint file.
    LOG.info("Loading checkpoint file: " + mJournal.getCheckpointFilePath());
    if (applyToMaster) {
      // TODO (Gene): `is` is not closed.
      JournalInputStream is = mReader.getCheckpointInputStream();
      mMaster.processJournalCheckpoint(is);
    }
  }

  /**
   * Processes all the next journal log files. {@link #processJournalCheckpoint(boolean)} must have
   * been called previously.
   *
   * @return the number of log files processed.
   * @throws IOException
   */
  public int processNextJournalLogFiles() throws IOException {
    int numProcessed = 0;
    while (true) {
      if (!mReader.isValid()) {
        LOG.info("The checkpoint is out of date. Must reload checkpoint file. "
            + mJournal.getCheckpointFilePath());
        return numProcessed;
      }

      // Process the new completed log file, if it exists.
      JournalInputStream inputStream = mReader.getNextInputStream();
      if (inputStream != null) {
        LOG.info("Processing a completed log file.");
        JournalEntry entry;
        while ((entry = inputStream.getNextEntry()) != null) {
          mMaster.processJournalEntry(entry);
        }
        inputStream.close();
        numProcessed ++;
        LOG.info("Finished processing the log file.");
      } else {
        return numProcessed;
      }
    }
  }
}

@@ -963,16 +963,169 @@
 aster;%0A%0A
+/**%0A * This class tails the journal for a master. It will process the journal checkpoint file, and then%0A * process all existing completed log files.%0A */%0A
 public f
@@ -2415,136 +2415,338 @@
     
-if (applyToMaster) %7B%0A      // TODO (Gene): %60is%60 is not closed.%0A      JournalInputStream is = mReader.getCheckpointInputStream();
+// The checkpoint stream must be retrieved before retrieving any log file streams, because the%0A    // journal reader verifies that the checkpoint was read before the log files.%0A    JournalInputStream is = mReader.getCheckpointInputStream();%0A%0A    if (applyToMaster) %7B%0A      // Only apply the checkpoint to the master, if specified.
 %0A   
@@ -2786,22 +2786,38 @@
 nt(is);%0A
-
     %7D%0A
+    is.close();%0A
   %7D%0A%0A  /

Improve comments in JournalTailer