package azkaban.common.web;

import java.io.IOException;
import java.io.PrintWriter;

import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.SequenceFile;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.SequenceFile.Reader;

import voldemort.serialization.json.JsonTypeSerializer;
import azkaban.common.web.HdfsSequenceFileViewer;

public class JsonSequenceFileViewer extends HdfsSequenceFileViewer {

    public boolean canReadFile(Reader reader) {
        Text keySchema = reader.getMetadata().get(new Text("key.schema"));
        Text valueSchema = reader.getMetadata().get(new Text("value.schema"));

        return keySchema != null && valueSchema != null;
    }

    public void displaySequenceFile(SequenceFile.Reader reader,
                                    PrintWriter output,
                                    int startLine,
                                    int endLine) throws IOException {
        try {
            BytesWritable keyWritable = new BytesWritable();
            BytesWritable valueWritable = new BytesWritable();
            Text keySchema = reader.getMetadata().get(new Text("key.schema"));
            Text valueSchema = reader.getMetadata().get(new Text("value.schema"));

            JsonTypeSerializer keySerializer = new JsonTypeSerializer(keySchema.toString());
            JsonTypeSerializer valueSerializer = new JsonTypeSerializer(valueSchema.toString());

            // skip lines before the start line
            for(int i = 1; i < startLine; i++)
                reader.next(keyWritable, valueWritable);

            // now actually output lines
            for(int i = startLine; i <= endLine; i++) {
                boolean readSomething = reader.next(keyWritable, valueWritable);
                if(!readSomething)
                    break;
                output.write(keySerializer.toObject(keyWritable.get()).toString());
                output.write("\t=>\t");
                output.write(valueSerializer.toObject(valueWritable.get()).toString());
                output.write("\n");
                output.flush();
            }
        } finally {
            reader.close();
        }
    }
}
@@ -1847,16 +1847,29 @@
 t.write(
+safeToString(
 keySeria
@@ -1897,34 +1897,24 @@
 table.get())
-.toString(
 ));%0A        
@@ -1974,16 +1974,29 @@
 t.write(
+safeToString(
 valueSer
@@ -2032,26 +2032,16 @@
 e.get())
-.toString(
 ));%0A    
@@ -2182,9 +2182,153 @@
 %7D%0A    %7D%0A
+    %0A    private String safeToString(Object value) %7B%0A    %09if(value == null)%0A    %09%09return %22null%22;%0A    %09else%0A    %09%09return value.toString();%0A    %7D%0A
 %7D

Fix null pointer in file viewer.