package edu.umn.cs.spatialHadoop.mapReduce;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.RecordWriter;
import org.apache.hadoop.spatial.CellInfo;
import org.apache.hadoop.spatial.Point;
import org.apache.hadoop.spatial.Shape;
import org.apache.hadoop.spatial.SpatialSite;
import org.apache.hadoop.util.Progressable;

public class RTreeGridOutputFormat<S extends Shape> extends FileOutputFormat<IntWritable, Text> {

  @Override
  public RecordWriter<IntWritable, Text> getRecordWriter(FileSystem ignored,
      JobConf job,
      String name,
      Progressable progress)
      throws IOException {
    // Output file name
    Path outFile = FileOutputFormat.getTaskOutputPath(job, name);

    // Get file system
    FileSystem fileSystem = outFile.getFileSystem(job);
    
    boolean overwrite = job.getBoolean(GridOutputFormat.OVERWRITE, false);

    // Get grid info
    CellInfo[] cellsInfo = GridOutputFormat.decodeCells(job.get(GridOutputFormat.OUTPUT_CELLS));
    RTreeGridRecordWriter writer = new RTreeGridRecordWriter
        (fileSystem, outFile, cellsInfo, overwrite);
    writer.setStockObject(createStockShape(job));
    return writer;
  }

  private S createStockShape(Configuration job) {
    S stockShape = null;
    String shapeClassName = job.get(SpatialSite.SHAPE_CLASS, Point.class.getName());
    try {
      Class<? extends Shape> shapeClass =
          Class.forName(shapeClassName).asSubclass(Shape.class);
      stockShape = (S) shapeClass.newInstance();
    } catch (ClassNotFoundException e) {
      e.printStackTrace();
    } catch (InstantiationException e) {
      e.printStackTrace();
    } catch (IllegalAccessException e) {
      e.printStackTrace();
    }
    return stockShape;
  }

}


@@ -71,53 +71,8 @@
 n;%0A%0A
-import org.apache.hadoop.conf.Configuration;%0A
 impo
@@ -107,16 +107,16 @@
 System;%0A
+
 import o
@@ -400,48 +400,8 @@
 fo;%0A
-import org.apache.hadoop.spatial.Point;%0A
 impo
@@ -1318,16 +1318,28 @@
 kObject(
+SpatialSite.
 createSt
@@ -1375,572 +1375,8 @@
 iter
-;%0A  %7D%0A%0A  private S createStockShape(Configuration job) %7B%0A    S stockShape = null;%0A    String shapeClassName = job.get(SpatialSite.SHAPE_CLASS, Point.class.getName());%0A    try %7B%0A      Class%3C? extends Shape%3E shapeClass =%0A          Class.forName(shapeClassName).asSubclass(Shape.class);%0A      stockShape = (S) shapeClass.newInstance();%0A    %7D catch (ClassNotFoundException e) %7B%0A      e.printStackTrace();%0A    %7D catch (InstantiationException e) %7B%0A      e.printStackTrace();%0A    %7D catch (IllegalAccessException e) %7B%0A      e.printStackTrace();%0A    %7D%0A    return stockShape
 ;%0A  

Use SpatialSite#createStockShape in RTreeGridRecordWriter