/*
 * Copyright 2011, Bestofmedia, Inc.
 * 
 * Bestofmedia licenses this file to you under the Apache License, version
 * 2.0 (the "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *    http://www.apache.org/licenses/LICENSE-2.0
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
 * implied.  See the License for the specific language governing
 * permissions and limitations under the License.
 */
package com.tomslabs.grid.avro;

import java.io.IOException;

import org.apache.avro.file.DataFileReader;
import org.apache.avro.file.FileReader;
import org.apache.avro.generic.GenericDatumReader;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.mapred.FsInput;
import org.apache.hadoop.io.DataOutputBuffer;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.FileSplit;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.RecordReader;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.typedbytes.TypedBytesOutput;
import org.apache.hadoop.typedbytes.TypedBytesWritable;

public class AvroAsTextTypedBytesInputFormat extends FileInputFormat<TypedBytesWritable, TypedBytesWritable> {

    @Override
    public org.apache.hadoop.mapred.RecordReader<TypedBytesWritable, TypedBytesWritable> getRecordReader(org.apache.hadoop.mapred.InputSplit split,
            JobConf job, Reporter reporter) {
        try {
            return new AvroTypedBytesRecordReader<GenericRecord>(job, (FileSplit) split);
        } catch (IOException e) {
            e.printStackTrace();
            return null;
        }
    }

    static class AvroTypedBytesRecordReader<T> implements RecordReader<TypedBytesWritable, TypedBytesWritable> {

        private FileReader<T> reader;
        private long start;
        private long end;

        public AvroTypedBytesRecordReader(JobConf job, FileSplit split) throws IOException {
            this(new DataFileReader<T>(new FsInput(split.getPath(), job), new GenericDatumReader<T>()), split);
        }

        protected AvroTypedBytesRecordReader(FileReader<T> reader, FileSplit split) throws IOException {
            this.reader = reader;
            reader.sync(split.getStart()); // sync to start
            this.start = reader.tell();
            this.end = split.getStart() + split.getLength();
        }
        
        public TypedBytesWritable createKey() {
            return new TypedBytesWritable();
        }

        public TypedBytesWritable createValue() {
            return new TypedBytesWritable();
        }
        
        public boolean next(TypedBytesWritable key, TypedBytesWritable value) throws IOException {
            if (!reader.hasNext() || reader.pastSync(end)) {
                return false;
            }

            StringBuilder buf = new StringBuilder();
            JSONUtils.writeJSON(reader.next(), buf);

            key.setValue("");
            value.setValue(buf.toString());
            return true;
        }

        public float getProgress() throws IOException {
            if (end == start) {
                return 0.0f;
            } else {
                return Math.min(1.0f, (getPos() - start) / (float) (end - start));
            }
        }

        public long getPos() throws IOException {
            return reader.tell();
        }

        public void close() throws IOException {
            reader.close();
        }
    }

}

@@ -680,16 +680,67 @@
 ception;
+%0Aimport java.util.ArrayList;%0Aimport java.util.List;
 %0A%0Aimport
@@ -914,16 +914,64 @@
 Record;%0A
+import org.apache.avro.mapred.AvroOutputFormat;%0A
 import o
@@ -1030,27 +1030,21 @@
 oop.
-io.DataOutputBuffer
+fs.FileStatus
 ;%0Aim
@@ -1133,16 +1133,60 @@
 eSplit;%0A
+import org.apache.hadoop.mapred.InputSplit;%0A
 import o
@@ -1310,62 +1310,8 @@
 er;%0A
-import org.apache.hadoop.typedbytes.TypedBytesOutput;%0A
 impo
@@ -1498,39 +1498,365 @@
    p
-ublic org.apache.hadoop.mapred.
+rotected FileStatus%5B%5D listStatus(JobConf job) throws IOException %7B%0A      List%3CFileStatus%3E result = new ArrayList%3CFileStatus%3E();%0A      for (FileStatus file : super.listStatus(job))%0A        if (file.getPath().getName().endsWith(AvroOutputFormat.EXT))%0A          result.add(file);%0A      return result.toArray(new FileStatus%5B0%5D);%0A    %7D%0A%0A    @Override%0A    public 
 Reco
@@ -1924,33 +1924,8 @@
 der(
-org.apache.hadoop.mapred.
 Inpu
@@ -1937,28 +1937,16 @@
 t split,
-%0A           
  JobConf
@@ -1973,31 +1973,80 @@
 ter)
- %7B
 %0A      
-  try %7B%0A      
+throws IOException %7B%0A      reporter.setStatus(split.toString());%0A
     
@@ -2129,110 +2129,8 @@
 t);%0A
-        %7D catch (IOException e) %7B%0A            e.printStackTrace();%0A            return null;%0A        %7D%0A
     
@@ -3301,115 +3301,8 @@
  %7D%0A%0A
-            StringBuilder buf = new StringBuilder();%0A            JSONUtils.writeJSON(reader.next(), buf);%0A%0A
     
@@ -3358,11 +3358,21 @@
 lue(
-buf
+reader.next()
 .toS

refactor AvroAsTextTypedBytesInputFormat