package com.neverwinterdp.ssm.hdfs;

import java.io.EOFException;
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 {
      return dataNextRecordWithRetry();
    } catch(IOException ex) {
      System.err.println(
          "dataNextRecord() error, currentReadPos = " + currentReadPos +
          ", commit pos = " + segment.getDataSegmentLastCommitPos());
      ex.printStackTrace();
      throw ex;
    }
  }

  byte[] dataNextRecordWithRetry() throws IOException {
    try {
      int size    = dataIs.readInt();
      byte[] data = new byte[size];
      dataIs.readFully(data);
      currentReadPos += 4 + data.length;
      return data;
    } catch(EOFException ex) {
      dataIs.close();
      dataIs  = fs.open(new Path(segmentFullPath)) ;
      dataIs.seek(currentReadPos);
    }
    int size    = dataIs.readInt();
    byte[] data = new byte[size];
    dataIs.readFully(data);
    currentReadPos += 4 + data.length;
    return data;
  }
  
  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();
  }
}
@@ -1232,54 +1232,13 @@
     
-dataIs  = fs.open(new Path(segmentFullPath)) ;
+%0A    
 %0A   
@@ -1372,25 +1372,158 @@
 tion();%0A    
- 
+%7D%0A  %7D%0A%0A  void openDataInputStream() throws IOException %7B%0A    dataIs  = fs.open(new Path(segmentFullPath)) ;%0A    if(currentReadPos %3E 0)
  dataIs.seek
@@ -1540,26 +1540,22 @@
 adPos);%0A
-  
   %7D%0A  
-%7D%0A
 %0A  @Over
@@ -1622,24 +1622,72 @@
 %7B%0A    try %7B%0A
+      if(dataIs == null) openDataInputStream();%0A
       return
@@ -1769,27 +1769,16 @@
 println(
-%0A          
 %22dataNex
@@ -1790,15 +1790,8 @@
 rd()
- error,
  cur
@@ -1824,26 +1824,16 @@
 eadPos +
-%0A         
  %22, comm
@@ -2233,87 +2233,28 @@
     
-dataIs  = fs.open(new Path(segmentFullPath)) ;%0A      dataIs.seek(currentReadPos
+openDataInputStream(
 );%0A 

fix hdfs segment read