package org.elasticsearch.hadoop.mr;

import java.util.List;
import java.util.Map;

import javax.xml.bind.DatatypeConverter;

import org.apache.hadoop.io.ArrayWritable;
import org.apache.hadoop.io.BooleanWritable;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.FloatWritable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.MapWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;
import org.elasticsearch.hadoop.serialization.FieldType;
import org.elasticsearch.hadoop.serialization.JdkValueReader;

public class WritableValueReader extends JdkValueReader {

    @Override
    public Map createMap() {
        return new MapWritable();
    }

    @Override
    public Object createArray(FieldType type) {
        Class<? extends Writable> arrayType = null;

        switch (type) {
        case NULL:
            arrayType = NullWritable.class;
            break;
        case STRING:
            arrayType = Text.class;
            break;
        case INTEGER:
            arrayType = IntWritable.class;
            break;
        case LONG:
            arrayType = LongWritable.class;
            break;
        case FLOAT:
            arrayType = FloatWritable.class;
            break;
        case DOUBLE:
            arrayType = DoubleWritable.class;
            break;
        case BOOLEAN:
            arrayType = BooleanWritable.class;
            break;
        case DATE:
            arrayType = LongWritable.class;
        case BINARY:
            arrayType = BytesWritable.class;
            break;
        case OBJECT:
            arrayType = MapWritable.class;
            break;
        }

        return new ArrayWritable(arrayType);
    }

    @Override
    public Object addToArray(Object array, List<Object> values) {
        ((ArrayWritable) array).set(values.toArray(new Writable[values.size()]));
        return array;
    }

    @Override
    protected Object binaryValue(byte[] value) {
        return new BytesWritable(value);
    }

    @Override
    protected Object booleanValue(String value) {
        return new BooleanWritable(Boolean.parseBoolean(value));
    }

    @Override
    protected Object doubleValue(String value) {
        return new DoubleWritable(Double.parseDouble(value));
    }

    @Override
    protected Object floatValue(String value) {
        return new FloatWritable(Float.parseFloat(value));
    }

    @Override
    protected Object longValue(String value) {
        return new LongWritable(Long.parseLong(value));
    }
    
	@Override
    protected Object intValue(String value) {
        return new IntWritable(Integer.parseInt(value));
    }

    @Override
    protected Object textValue(String value) {
        return new Text(value);
    }

    @Override
    protected Object nullValue(String value) {
        return NullWritable.get();
    }
    
    @Override
	protected Object date(String value) {
    	// convert String to Long
    	return new LongWritable(DatatypeConverter.parseDateTime(value).getTimeInMillis());
	}
}
@@ -81,50 +81,8 @@
 p;%0A%0A
-import javax.xml.bind.DatatypeConverter;%0A%0A
 impo
@@ -112,32 +112,32 @@
 .ArrayWritable;%0A
+
 import org.apach
@@ -1560,34 +1560,45 @@
 yType = 
-LongWritable.class
+dateType();%0A            break
 ;%0A      
@@ -2005,32 +2005,119 @@
 n array;%0A    %7D%0A%0A
+    protected Class%3C? extends Writable%3E dateType() %7B%0A        return Text.class;%0A    %7D%0A%0A
     @Override%0A  
@@ -2734,14 +2734,13 @@
   %7D%0A
+%0A
     
-%0A%09
 @Ove
@@ -3061,13 +3061,9 @@
   %7D%0A
-    
 %0A
+
     
@@ -3072,17 +3072,20 @@
 verride%0A
-%09
+    
 protecte
@@ -3122,123 +3122,39 @@
     
-%09// convert String to Long%0A    %09return new LongWritable(DatatypeConverter.parseDateTime(value).getTimeInMillis());%0A%09
+    return new Text(value);%0A    
 %7D%0A%7D

make WritableValueReader more pluggable