/*
 * Licensed to the Apache Software Foundation (ASF) 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.client;

import java.io.File;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.channels.FileChannel.MapMode;

import org.apache.log4j.Logger;

import com.google.common.base.Preconditions;

import tachyon.Constants;
import tachyon.util.CommonUtils;

/**
 * BlockHandler for files on LocalFS, such as RamDisk, SSD and HDD.
 */
public final class BlockHandlerLocal extends BlockHandler {

  private final Logger mLog = Logger.getLogger(Constants.LOGGER_TYPE);
  private final RandomAccessFile mLocalFile;
  private final FileChannel mLocalFileChannel;
  private boolean mPermission = false;
  protected final String mFilePath;

  BlockHandlerLocal(String filePath) throws IOException {
    mFilePath = Preconditions.checkNotNull(filePath);
    mLog.debug(mFilePath + " is created");
    mLocalFile = new RandomAccessFile(mFilePath, "rw");
    mLocalFileChannel = mLocalFile.getChannel();
  }

  @Override
  public int append(long blockOffset, ByteBuffer srcBuf) throws IOException {
    checkPermission();
    ByteBuffer out = mLocalFileChannel.map(MapMode.READ_WRITE, blockOffset, srcBuf.limit());
    out.put(srcBuf);

    return srcBuf.limit();
  }

  private void checkPermission() throws IOException {
    if (!mPermission) {
      // change the permission of the file and use the sticky bit
      CommonUtils.changeLocalFileToFullPermission(mFilePath);
      CommonUtils.setLocalFileStickyBit(mFilePath);
      mPermission = true;
    }
  }

  @Override
  public void close() throws IOException {
    IOException exception = null;
    if (mLocalFileChannel != null) {
      try {
        mLocalFileChannel.close();
      } catch (IOException e) {
        exception = e;
      }
    }
    if (mLocalFile != null) {
      try {
        mLocalFile.close();
      } catch (IOException e) {
        if (exception == null) {
          exception = e;
        } else {
          mLog.error("Error during close file:" + mFilePath, e);
        }
      }
    }
    if (exception != null) {
      throw exception;
    }
  }

  @Override
  public boolean delete() throws IOException {
    checkPermission();
    return new File(mFilePath).delete();
  }

  @Override
  public ByteBuffer read(long blockOffset, int length) throws IOException {
    long fileLength = mLocalFile.length();
    String error = null;
    if (blockOffset > fileLength) {
      error =
          String.format("blockOffset(%d) is larger than file length(%d)", blockOffset, fileLength);
    }
    if (error == null && length != -1 && blockOffset + length > fileLength) {
      error =
          String.format("blockOffset(%d) plus length(%d) is larger than file length(%d)",
              blockOffset, length, fileLength);
    }
    if (error != null) {
      throw new IllegalArgumentException(error);
    }
    if (length == -1) {
      length = (int) (fileLength - blockOffset);
    }
    ByteBuffer buf = mLocalFileChannel.map(FileChannel.MapMode.READ_ONLY, blockOffset, length);
    return buf;
  }
}

@@ -808,16 +808,42 @@
 lient;%0A%0A
+import java.io.Closeable;%0A
 import j
@@ -1027,16 +1027,69 @@
 MapMode;
+%0Aimport java.util.ArrayDeque;%0Aimport java.util.Deque;
 %0A%0Aimport
@@ -1568,15 +1568,13 @@
   pr
-otec
+iva
 te
-d
  fin
@@ -1593,16 +1593,88 @@
 ilePath;
+%0A  private final Deque%3CCloseable%3E mStack = new ArrayDeque%3CCloseable%3E(2);
 %0A%0A  Bloc
@@ -1923,24 +1923,89 @@
 tChannel();%0A
+    mStack.push(mLocalFile);%0A    mStack.push(mLocalFileChannel);%0A
   %7D%0A%0A  @Over
@@ -2649,183 +2649,32 @@
     
-if (mLocalFileChannel != null) %7B%0A      try %7B%0A        mLocalFileChannel.close();%0A      %7D catch (IOException e) %7B%0A        exception = e;%0A      %7D%0A    %7D%0A    if (mLocalFile != null
+while (!mStack.isEmpty()
 ) %7B%0A
@@ -2694,25 +2694,27 @@
        m
-LocalFile
+Stack.pop()
 .close()
@@ -2847,46 +2847,22 @@
 ror(
-%22Error during close file:%22 + mFilePath
+e.getMessage()
 , e)

make close() simpler