package net.pixelcop.sewer.sink;

import java.io.DataOutputStream;
import java.io.IOException;

import net.pixelcop.sewer.Event;
import net.pixelcop.sewer.Sink;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.io.compress.Compressor;
import org.apache.hadoop.io.compress.GzipCodec;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Simple wrapper around {@link FSDataOutputStream} with compression
 *
 * @author chetan
 *
 */
public class DfsSink implements Sink {

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

  /**
   * Configured DFS path to write to
   */
  private String configPath;

  /**
   * Reference to DFS Path object
   */
  private Path dstPath;

  private DataOutputStream writer;

  public DfsSink(String path) {
    this.configPath = path;
  }

  @Override
  public void close() throws IOException {
    writer.close();
  }

  @Override
  public void open() throws IOException {

    String fullPath = BucketPath.escapeString(configPath, null);
    createWriter(fullPath);

  }

  private void createWriter(String path) throws IOException {

    Configuration conf = new Configuration();
    FileSystem hdfs;

    // String codecName = conf.getCollectorDfsCompressCodec();
    // CompressionCodec codec = getCodec(conf, codecName);
    CompressionCodec codec = new GzipCodec(); // TODO (handle pluggable
                                              // compression with HDFS)

    // if (codec == null) {
    // dstPath = new Path(path);
    // hdfs = dstPath.getFileSystem(conf);
    // writer = hdfs.create(dstPath);
    // LOG.info("Creating HDFS file: " + dstPath.toString());
    // return;
    // }

    Compressor cmp = codec.createCompressor();
    dstPath = new Path(path + codec.getDefaultExtension());
    hdfs = dstPath.getFileSystem(conf);
    writer = new DataOutputStream(codec.createOutputStream(hdfs.create(dstPath), cmp));

    if (LOG.isInfoEnabled()) {
      LOG.info("Creating " + codec + " compressed HDFS file: " + dstPath.toString());
    }
  }

  @Override
  public void append(Event event) throws IOException {
    event.write(writer);
  }

}

@@ -46,54 +46,63 @@
 .io.
-DataOutputStream;%0Aimport java.io.IOException;%0A
+IOException;%0A%0Aimport net.pixelcop.sewer.ByteArrayEvent;
 %0Aimp
@@ -331,16 +331,207 @@
 s.Path;%0A
+import org.apache.hadoop.io.NullWritable;%0Aimport org.apache.hadoop.io.SequenceFile;%0Aimport org.apache.hadoop.io.SequenceFile.CompressionType;%0Aimport org.apache.hadoop.io.SequenceFile.Writer;%0A
 import o
@@ -950,16 +950,80 @@
 lass);%0A%0A
+  private static final NullWritable NULL = NullWritable.get();%0A%0A
   /**%0A  
@@ -1174,32 +1174,22 @@
 private 
-DataOutputStream
+Writer
  writer;
@@ -1306,24 +1306,52 @@
 Exception %7B%0A
+    if (writer != null) %7B%0A  
     writer.c
@@ -1358,16 +1358,22 @@
 lose();%0A
+    %7D%0A
   %7D%0A%0A  @
@@ -1419,25 +1419,24 @@
 Exception %7B%0A
-%0A
     String f
@@ -1516,17 +1516,16 @@
 lPath);%0A
-%0A
   %7D%0A%0A  p
@@ -2295,19 +2295,23 @@
 (conf);%0A
+%0A
    
+ //
  writer 
@@ -2386,16 +2386,150 @@
 , cmp));
+%0A    writer = SequenceFile.createWriter(%0A        hdfs, conf, dstPath, NullWritable.class, ByteArrayEvent.class, CompressionType.NONE);
 %0A%0A    if
@@ -2721,16 +2721,50 @@
 n %7B%0A    
+writer.append(NULL, event);%0A    //
 event.wr

switch to writing SequenceFiles