package com.neverwinterdp.vm.environment.yarn;

import java.io.BufferedInputStream;
import java.io.IOException;
import java.io.OutputStream;

import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

public class HDFSUtil {
  static public int getStreamId(Path path) {
    String name = path.getName();
    int dashIdx = name.lastIndexOf('-');
    return Integer.parseInt(name.substring(dashIdx + 1)) ;
  }
  
  static public void concat(FileSystem fs, Path dest, Path[] src) throws IOException {
    FSDataOutputStream output = fs.create(dest) ;
    for(int i = 0; i < src.length; i++) {
      FSDataInputStream is = fs.open(src[i]);
      //BufferedInputStream buffer = new BufferedInputStream(is);
      byte[] data = new byte[4912];
      int available = -1;
      while ((available = is.read(data)) > -1) {
        output.write(data, 0, available);
      }
      is.close();
      //buffer.close();
    }
    output.hflush();
    output.close();
  }
  
  static public void concat(FileSystem fs, Path dest, Path[] src, boolean deleteSrc) throws IOException {
    concat(fs, dest, src);
    if(deleteSrc) {
      for(int i = 0; i < src.length; i++) {
        fs.delete(src[i], true);
      }
    }
  }
  
  static public void dump(FileSystem fs, String dir) throws IOException {
    System.out.println("----------------------------------------------------");
    System.out.println(dir);
    System.out.println("----------------------------------------------------");
    Path path = new Path(dir);
    if(!fs.exists(path)) {
      System.out.println(dir + " does not exist!") ;
      return;
    }
    dump(fs, path, "");
  }
  
  static void dump(FileSystem fs, Path path, String indentation) throws IOException {
    System.out.println(indentation + " " + path.getName());
    if(fs.isFile(path)) return;
    FileStatus[] status = fs.listStatus(path) ;
    String nextIndentation = indentation + "  " ;
    for(int i = 0; i < status.length; i++) {
      dump(fs, status[i].getPath(), nextIndentation);
    }
  }
}
@@ -60,84 +60,19 @@
 .io.
-BufferedInputStream;%0Aimport java.io.IOException;%0Aimport java.io.OutputStream
+IOException
 ;%0A%0Ai
@@ -245,16 +245,61 @@
 System;%0A
+import org.apache.hadoop.fs.LocalFileSystem;%0A
 import o
@@ -527,32 +527,38 @@
  + 1)) ;%0A  %7D%0A  %0A
+  %0A  %0A
   static public 
@@ -624,24 +624,66 @@
 Exception %7B%0A
+    if(fs instanceof LocalFileSystem) %7B%0A  
     FSDataOu
@@ -716,24 +716,26 @@
 ate(dest) ;%0A
+  
     for(int 
@@ -766,24 +766,26 @@
 ++) %7B%0A      
+  
 FSDataInputS
@@ -812,16 +812,18 @@
 rc%5Bi%5D);%0A
+  
       //
@@ -886,16 +886,18 @@
 ;%0A      
+  
 byte%5B%5D d
@@ -920,24 +920,26 @@
 912%5D;%0A      
+ 
+ 
 int availabl
@@ -946,16 +946,18 @@
 e = -1;%0A
+  
       wh
@@ -1001,24 +1001,26 @@
 ) %7B%0A        
+ 
+ 
 output.write
@@ -1037,24 +1037,26 @@
 available);%0A
+  
       %7D%0A    
@@ -1057,16 +1057,18 @@
 %7D%0A      
+  
 is.close
@@ -1067,24 +1067,26 @@
 is.close();%0A
+  
       //buff
@@ -1097,26 +1097,30 @@
 lose();%0A    
-%7D%0A
+  %7D%0A  
     output.h
@@ -1128,24 +1128,26 @@
 lush();%0A    
+  
 output.close
@@ -1150,16 +1150,63 @@
 lose();%0A
+    %7D else %7B%0A      fs.concat(dest, src);%0A    %7D%0A
   %7D%0A  %0A 

add native hdfs file concat for real hdfs