package net.pixelcop.sewer.source;

import java.io.IOException;

import net.pixelcop.sewer.ByteArrayEvent;
import net.pixelcop.sewer.Event;
import net.pixelcop.sewer.Sink;
import net.pixelcop.sewer.Source;
import net.pixelcop.sewer.node.Node;
import net.pixelcop.sewer.sink.BucketedSink;
import net.pixelcop.sewer.sink.durable.Transaction;
import net.pixelcop.sewer.util.HdfsUtil;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.SequenceFile;
import org.apache.hadoop.io.SequenceFile.Reader;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class TransactionSource extends Source {

  private static final Logger LOG = LoggerFactory.getLogger(TransactionSource.class);

  private Transaction tx;

  private Path path;
  private String bucket;
  private String ext;
  private Sink sink;

  public TransactionSource(Transaction tx) {

    this.tx = tx;

    this.path = tx.createTxPath();
    this.bucket = tx.getBucket();
    this.ext = tx.getFileExt();
  }

  @Override
  public void close() throws IOException {
    if (sink != null) {
      sink.close();
    }
  }

  /**
   * Opens the SequenceFile and drains its contents to the configured Sink
   */
  @Override
  public void open() throws IOException {

    LOG.debug("Going to drain " + path.toString() + " to bucket " + bucket);
    setStatus(OPENING);

    // before we do anything, let's delete the existing destination file so
    // we don't have any problems
    if (bucket != null) {
      // only if bucketed sink being used
      Path dst = new Path(bucket + ext);
      try {
        HdfsUtil.deletePath(dst);
      } catch (InterruptedException e) {
        setStatus(ERROR);
        throw new IOException("Interrupted trying to delete " + dst, e);
      }
    }

    sink = getSinkFactory().build();
    if (sink instanceof BucketedSink && bucket != null) {
      ((BucketedSink) sink).setNextBucket(bucket);
    }
    sink.open();

    Reader reader = null;
    try {
      try {
        reader = createReader();
      } catch (IOException e) {
        // May occur if the file wasn't found or is zero bytes (never received any data)
        // This generally happens if the server was stopped improperly (kill -9, crash, reboot)
        LOG.warn("Failed to read tx " + tx + " at " + tx.createTxPath().toString(), e);
        return;
      }
      setStatus(FLOWING);

      NullWritable nil = NullWritable.get();
      Event event = null;
      try {
        event = tx.newEvent();
      } catch (Exception e) {
        // Should really never happen, since the Event class should always be available
        throw new IOException("Failed to create Event class", e);
      }

      while (reader.next(nil, event)) {
        sink.append(event);
      }

    } finally {
      if (reader != null) {
        try {
          reader.close();
        } catch (IOException e) {
          // safe to ignore (hopefully :)
        }
      }
    }

    setStatus(CLOSED);
  }

  public Reader createReader() throws IOException {

    Configuration conf = Node.getInstance().getConf();
    FileSystem hdfs = path.getFileSystem(conf);

    return new SequenceFile.Reader(hdfs, path, conf);
  }

  @Override
  public Class<?> getEventClass() {
    return ByteArrayEvent.class;
  }

}

@@ -29,16 +29,45 @@
 ource;%0A%0A
+import java.io.EOFException;%0A
 import j
@@ -86,16 +86,16 @@
 eption;%0A
-
 %0Aimport 
@@ -714,16 +714,17 @@
 ctory;%0A%0A
+%0A
 public c
@@ -2409,20 +2409,20 @@
 iled to 
-read
+open
  tx %22 + 
@@ -2463,16 +2463,111 @@
 String()
+%0A            + %22; this usually means the file is 0 bytes or the header is corrupted/incomplete%22
 , e);%0A  
@@ -2918,24 +2918,25 @@
 );%0A      %7D%0A%0A
+%0A
       while 
@@ -2940,34 +2940,316 @@
 le (
-reader.next(nil, event)) %7B
+true) %7B%0A%0A        try %7B%0A          if (!reader.next(nil, event)) %7B%0A            break;%0A          %7D%0A        %7D catch (IOException e) %7B%0A          if (e instanceof EOFException) %7B%0A            LOG.warn(%22Caught EOF reading from buffer; skipping to close%22);%0A            break;%0A          %7D%0A          throw e;%0A        %7D%0A
 %0A   
@@ -3278,24 +3278,25 @@
 );%0A      %7D%0A%0A
+%0A
     %7D finall
@@ -3393,32 +3393,32 @@
 OException e) %7B%0A
-
           // saf
@@ -3443,16 +3443,100 @@
 ully :)%0A
+          // because at this point, we only care about errors when closing the sink%0A
         

handle EOF while reading from buffer (after file is already open)