package com.neverwinterdp.ssm.hdfs;

import java.io.IOException;

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

import com.neverwinterdp.ssm.SSMReaderDescriptor;
import com.neverwinterdp.ssm.SSMRegistry;
import com.neverwinterdp.ssm.SegmentDescriptor;
import com.neverwinterdp.ssm.SegmentReadDescriptor;
import com.neverwinterdp.ssm.SegmentReader;

public class HdfsSegmentReader extends SegmentReader {
  private FileSystem        fs;
  private String            storageLocation;
  private String            segmentFullPath;
  private FSDataInputStream dataIs;
  private long              currentReadPos = 0;
  
  public HdfsSegmentReader(SSMRegistry registry, SSMReaderDescriptor readerDescriptor, 
                           SegmentDescriptor segment, SegmentReadDescriptor segmentReadDescriptor, 
                           FileSystem fs, String storageLoc) throws IllegalArgumentException, IOException {
    super(registry, readerDescriptor, segment, segmentReadDescriptor);
    
    this.fs = fs;
    this.storageLocation = storageLoc;
    
    segmentFullPath = storageLocation + "/" + segment.getSegmentId() + ".dat";
    dataIs  = fs.open(new Path(segmentFullPath)) ;
    if(segmentReadDescriptor.getCommitReadDataPosition() > 0) {
      currentReadPos = segmentReadDescriptor.getCommitReadDataPosition();
      dataIs.seek(currentReadPos);
    }
  }

  @Override
  protected byte[] dataNextRecord() throws IOException {
    try {
      int size    = dataIs.readInt();
      byte[] data = new byte[size];
      dataIs.readFully(data);
      currentReadPos += 4 + data.length;
      return data;
    } catch(IOException ex) {
      System.err.println("dataNextRecord() error, currentReadPos = " + currentReadPos + ", commit pos = " + segment.getDataSegmentLastCommitPos());
      ex.printStackTrace();
      throw ex;
    }
  }

  protected void rollback(long readRecordIndex, long pos) throws IOException {
    dataIs.seek(pos);
    currentReadPos = pos;
  }
  
  @Override
  protected long getCurrentReadPosition() { return currentReadPos; }
  
  protected void doClose() throws IOException {
    dataIs.close();
  }
}
@@ -1495,24 +1495,43 @@
 Exception %7B%0A
+    int size = 0 ;%0A
     try %7B%0A  
@@ -1530,28 +1530,24 @@
 try %7B%0A      
-int 
 size    = da
@@ -1743,16 +1743,27 @@
 println(
+%0A          
 %22dataNex
@@ -1816,16 +1816,91 @@
 eadPos +
+%0A          %22currentReadPos + size = %22 + (currentReadPos + size) +%0A         
  %22, comm

fix overflow buffer