/*
 * 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.examples;

import java.io.IOException;
import java.io.OutputStream;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.util.concurrent.Callable;

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

import tachyon.Constants;
import tachyon.TachyonURI;
import tachyon.client.ReadType;
import tachyon.client.TachyonFS;
import tachyon.client.TachyonFile;
import tachyon.client.file.FileInStream;
import tachyon.client.file.TachyonFileSystem.TachyonFileSystemFactory;
import tachyon.client.file.options.OutStreamOptions;
import tachyon.conf.TachyonConf;
import tachyon.exception.TachyonException;

/**
 * An example to show to how use Tachyon's API
 */
public class BasicCheckpoint implements Callable<Boolean> {
  private static final Logger LOG = LoggerFactory.getLogger(Constants.LOGGER_TYPE);

  private final TachyonURI mLocation;
  private final String mFileFolder;
  private final int mNumFiles;

  public BasicCheckpoint(TachyonURI tachyonURI, String fileFolder, int numFiles) {
    mLocation = tachyonURI;
    mFileFolder = fileFolder;
    mNumFiles = numFiles;
  }

  @Override
  public Boolean call() throws Exception {
    TachyonFS tachyonClient = TachyonFS.get(mLocation, new TachyonConf());
    writeFile(tachyonClient);
    return readFile(tachyonClient);
  }

  private boolean readFile(TachyonFS tachyonClient) throws IOException {
    boolean pass = true;
    for (int i = 0; i < mNumFiles; i ++) {
      TachyonURI filePath = new TachyonURI(mFileFolder + "/part-" + i);
      LOG.debug("Reading data from {}", filePath);
      TachyonFile file = tachyonClient.getFile(filePath);
      FileInStream is = file.getInStream(ReadType.CACHE);
      ByteBuffer buf = ByteBuffer.allocate((int) file.getBlockSizeByte());
      is.read(buf.array());
      buf.order(ByteOrder.nativeOrder());
      for (int k = 0; k < mNumFiles; k ++) {
        pass = pass && (buf.getInt() == k);
      }
      is.close();
    }
    return pass;
  }

  private void writeFile(TachyonFS tachyonClient) throws IOException, TachyonException {
    for (int i = 0; i < mNumFiles; i ++) {
      ByteBuffer buf = ByteBuffer.allocate(80);
      buf.order(ByteOrder.nativeOrder());
      for (int k = 0; k < mNumFiles; k ++) {
        buf.putInt(k);
      }
      buf.flip();
      TachyonURI filePath = new TachyonURI(mFileFolder + "/part-" + i);
      LOG.debug("Writing data to {}", filePath);
      OutputStream os =
          TachyonFileSystemFactory.get().getOutStream(filePath, OutStreamOptions.defaults());
      os.write(buf.array());
      os.close();
    }
  }

  public static void main(String[] args) throws IOException {
    if (args.length != 3) {
      System.out.println("java -cp " + Constants.TACHYON_JAR
          + " tachyon.examples.BasicCheckpoint <TachyonMasterAddress> <FileFolder> <Files>");
      System.exit(-1);
    }

    Utils.runExample(new BasicCheckpoint(new TachyonURI(args[0]), args[1], Integer
        .parseInt(args[2])));
  }
}

@@ -1101,49 +1101,25 @@
 ent.
-ReadType;%0Aimport tachyon.client.TachyonFS
+file.FileInStream
 ;%0Aim
@@ -1138,16 +1138,21 @@
 .client.
+file.
 TachyonF
@@ -1179,35 +1179,40 @@
 client.file.
-FileInStrea
+TachyonFileSyste
 m;%0Aimport ta
@@ -1330,41 +1330,8 @@
 ns;%0A
-import tachyon.conf.TachyonConf;%0A
 impo
@@ -1368,16 +1368,48 @@
 ception;
+%0Aimport tachyon.thrift.FileInfo;
 %0A%0A/**%0A *
@@ -1939,33 +1939,41 @@
 n %7B%0A    TachyonF
-S
+ileSystem
  tachyonClient =
@@ -1985,42 +1985,29 @@
 yonF
-S.get(mLocation, new TachyonConf()
+ileSystemFactory.get(
 );%0A 
@@ -2103,33 +2103,41 @@
 eadFile(TachyonF
-S
+ileSystem
  tachyonClient) 
@@ -2150,24 +2150,42 @@
  IOException
+, TachyonException
  %7B%0A    boole
@@ -2409,15 +2409,12 @@
 ent.
-getFile
+open
 (fil
@@ -2449,39 +2449,89 @@
 s = 
-file.getInStream(ReadType.CACHE
+tachyonClient.getInStream(file);%0A      FileInfo info = tachyonClient.getInfo(file
 );%0A 
@@ -2578,20 +2578,20 @@
 e((int) 
-file
+info
 .getBloc
@@ -2599,16 +2599,17 @@
 SizeByte
+s
 ());%0A   
@@ -2851,17 +2851,25 @@
 TachyonF
-S
+ileSystem
  tachyon
@@ -3290,49 +3290,22 @@
 os =
-%0A          TachyonFileSystemFactory.get()
+ tachyonClient
 .get

Update BasicCheckpoint to use TachyonFileSystem.