package com.neverwinterdp.scribengin.storage.hdfs.sink;

import java.io.IOException;
import java.util.UUID;

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

import com.neverwinterdp.scribengin.dataflow.DataflowMessage;
import com.neverwinterdp.scribengin.storage.sink.SinkStreamWriter;
import com.neverwinterdp.util.JSONSerializer;
import com.neverwinterdp.vm.environment.yarn.HDFSUtil;

public class HDFSSinkStreamWriter implements SinkStreamWriter {
  private FileSystem fs;
  private String location ;
  private SinkBuffer currentBuffer ;
  
  public HDFSSinkStreamWriter(FileSystem fs, String location) throws IOException {
    this.fs = fs;
    this.location = location;
    if(!fs.exists(new Path(location))) fs.mkdirs(new Path(location));
  }
  
  @Override
  synchronized public void append(DataflowMessage dataflowMessage) throws Exception {
    if(currentBuffer == null) {
      currentBuffer = nextSinkBuffer();
    }
    currentBuffer.append(dataflowMessage);
  }

  @Override
  public void rollback() throws Exception {
    currentBuffer.rollback();
  }

  @Override
  public void prepareCommit() throws Exception {
    //TODO: reimplement correctly 2 phases commit
  }

  @Override
  public void completeCommit() throws Exception {
  //TODO: reimplement correctly 2 phases commit
    currentBuffer.commit();
    currentBuffer = null ;
  }
  
  @Override
  synchronized public void commit() throws Exception {
    prepareCommit();
    completeCommit();
  }
  
  @Override
  synchronized public void close() throws Exception {
    if(currentBuffer != null) {
      currentBuffer.rollback();
      currentBuffer = null;
    }
  }
  
  private SinkBuffer nextSinkBuffer() throws IOException {
    SinkBuffer buffer = new SinkBuffer() ;
    return buffer;
  }
  
  public String toString() {
    StringBuilder b = new StringBuilder() ;
    b.append("location=").append(location);
    return b.toString() ;
  }
  
  class SinkBuffer {
    private Path writingPath;
    private Path completePath;
    private FSDataOutputStream output;
    private int count = 0 ;
    
    public SinkBuffer() throws IOException {
      String name = "data-" + UUID.randomUUID().toString() ;
      writingPath = new Path(location + "/" + name + ".writing") ;
      completePath = new Path(location + "/"  + name + ".dat") ;
      output = fs.create(writingPath) ;
    }
    
    public void append(DataflowMessage dataflowMessage) throws IOException {
      byte[] bytes = JSONSerializer.INSTANCE.toBytes(dataflowMessage) ;
      output.writeInt(bytes.length);
      output.write(bytes);
      count++;
    }
    
    public void delete() throws IOException {
      output.close();
      fs.delete(writingPath, true) ;
    }
    
    public void rollback() throws IOException {
      delete();
      count = 0;
      output = fs.create(writingPath, true) ;
    }
    
    public void commit() throws IOException {
      if(count <= 0) {
        delete() ;
      } else {
        output.close();
        fs.rename(writingPath, completePath);
      }
    }
  }
}
@@ -154,48 +154,8 @@
 am;%0A
-import org.apache.hadoop.fs.FileStatus;%0A
 impo
@@ -190,16 +190,16 @@
 System;%0A
+
 import o
@@ -403,63 +403,8 @@
 zer;
-%0Aimport com.neverwinterdp.vm.environment.yarn.HDFSUtil;
 %0A%0Apu
@@ -776,37 +776,24 @@
 @Override%0A  
-synchronized 
 public void 
@@ -981,99 +981,8 @@
  %7D%0A%0A
-  @Override%0A  public void rollback() throws Exception %7B%0A    currentBuffer.rollback();%0A  %7D%0A%0A
   @O
@@ -1155,16 +1155,18 @@
 ption %7B%0A
+  
   //TODO
@@ -1205,16 +1205,16 @@
  commit%0A
-
     curr
@@ -1376,24 +1376,142 @@
 ommit();%0A  %7D
+%0A%0A  @Override%0A  public void rollback() throws Exception %7B%0A    currentBuffer.rollback();%0A    currentBuffer = null ;%0A  %7D
 %0A  %0A  @Overr
@@ -2168,16 +2168,17 @@
 ng name 
+ 
 = %22data-
@@ -2226,24 +2226,25 @@
 writingPath 
+ 
 = new Path(l
@@ -2350,32 +2350,38 @@
  ;%0A      output 
+      
 = fs.create(writ
@@ -2657,22 +2657,24 @@
 ic void 
-delete
+rollback
 () throw
@@ -2752,83 +2752,8 @@
 ) ;%0A
-    %7D%0A    %0A    public void rollback() throws IOException %7B%0A      delete();%0A
     
@@ -2864,24 +2864,46 @@
 Exception %7B%0A
+      output.close();%0A
       if(cou
@@ -2921,23 +2921,43 @@
         
+fs.
 delete(
+writingPath, true
 ) ;%0A    
@@ -2971,32 +2971,8 @@
 e %7B%0A
-        output.close();%0A
     

fix hdfs sink writer