package org.elasticsearch.hadoop.pig;

import java.util.Calendar;

import javax.xml.bind.DatatypeConverter;

import org.apache.pig.data.DataType;
import org.joda.time.DateTime;
import org.joda.time.format.ISODateTimeFormat;


class DateConverter {

	private static boolean pig11Available;
	
	static {
		// initialize
		pig11Available = "datetime".equals(DataType.findTypeName((byte) 30));
	}
	
	static String convertToES(Object pigDate) {
		return (pig11Available ? Pig11OrHigherConverter.convertToES(pigDate) : PigUpTo10Converter.convertToES(pigDate));
	}
	
	static Object convertFromES(String esDate) {
		return (pig11Available ? Pig11OrHigherConverter.convertFromES(esDate) : PigUpTo10Converter.convertFromES(esDate));
	}
	
	private static abstract class PigUpTo10Converter {
		static String convertToES(Object pigDate) {
			if (pigDate instanceof Number) {
				Calendar cal = Calendar.getInstance();
				cal.setTimeInMillis(((Number) pigDate).longValue());
				return DatatypeConverter.printDateTime(cal);
			}
			if (pigDate instanceof String) {
				return ((String) pigDate);
			}
			throw new IllegalArgumentException(String.format("Cannot convert [%s] to date", pigDate));
		}
		
		static Object convertFromES(String esDate) {
			return DatatypeConverter.parseDateTime(esDate).getTimeInMillis();
		}
	}
	
	private static abstract class Pig11OrHigherConverter {
		static String convertToES(Object pigDate) {
			DateTime dt = (DateTime) pigDate;
			return dt.toString(ISODateTimeFormat.dateOptionalTimeParser());
		}
		
		static Object convertFromES(String esDate) {
			return ISODateTimeFormat.dateOptionalTimeParser().parseDateTime(esDate);
		}
	}
}
@@ -242,17 +242,20 @@
 rter %7B%0A%0A
-%09
+    
 private 
@@ -285,19 +285,21 @@
 ilable;%0A
-%09%0A%09
+%0A    
 static %7B
@@ -299,18 +299,24 @@
 tatic %7B%0A
-%09%09
+        
 // initi
@@ -321,18 +321,24 @@
 tialize%0A
-%09%09
+        
 pig11Ava
@@ -395,30 +395,35 @@
 byte) 30));%0A
-%09%7D%0A%09%0A%09
+    %7D%0A%0A    
 static Strin
@@ -446,34 +446,40 @@
 ject pigDate) %7B%0A
-%09%09
+        
 return (pig11Ava
@@ -575,22 +575,27 @@
 Date));%0A
-%09%7D%0A%09%0A%09
+    %7D%0A%0A    
 static O
@@ -631,18 +631,24 @@
 Date) %7B%0A
-%09%09
+        
 return (
@@ -754,22 +754,27 @@
 Date));%0A
-%09%7D%0A%09%0A%09
+    %7D%0A%0A    
 private 
@@ -808,34 +808,40 @@
 To10Converter %7B%0A
-%09%09
+        
 static String co
@@ -868,19 +868,28 @@
 Date) %7B%0A
-%09%09%09
+            
 if (pigD
@@ -913,20 +913,32 @@
 mber) %7B%0A
-%09%09%09%09
+                
 Calendar
@@ -968,20 +968,32 @@
 ance();%0A
-%09%09%09%09
+                
 cal.setT
@@ -1037,20 +1037,32 @@
 lue());%0A
-%09%09%09%09
+                
 return D
@@ -1098,24 +1098,42 @@
 e(cal);%0A
-%09%09%09%7D%0A%09%09%09
+            %7D%0A            
 if (pigD
@@ -1161,12 +1161,24 @@
 ) %7B%0A
-%09%09%09%09
+                
 retu
@@ -1204,16 +1204,34 @@
 e);%0A
-%09%09%09%7D%0A%09%09%09
+            %7D%0A            
 thro
@@ -1313,33 +1313,43 @@
  pigDate));%0A
-%09%09%7D%0A%09%09%0A%09%09
+        %7D%0A%0A        
 static Objec
@@ -1377,27 +1377,36 @@
 g esDate) %7B%0A
-%09%09%09
+            
 return Datat
@@ -1463,18 +1463,29 @@
 ();%0A
-%09%09%7D%0A%09%7D%0A%09%0A%09
+        %7D%0A    %7D%0A%0A    
 priv
@@ -1535,18 +1535,24 @@
 erter %7B%0A
-%09%09
+        
 static S
@@ -1587,19 +1587,28 @@
 Date) %7B%0A
-%09%09%09
+            
 DateTime
@@ -1637,30 +1637,23 @@
 te;%0A
-%09%09%09return dt.toString(
+            // 
 ISOD
@@ -1694,20 +1694,94 @@
 er()
-);%0A%09%09%7D%0A%09%09%0A%09%09
+ throws %22printing not supported%22%0A            return dt.toString();%0A        %7D%0A%0A        
 stat
@@ -1825,11 +1825,20 @@
 ) %7B%0A
-%09%09%09
+            
 retu
@@ -1910,12 +1910,21 @@
 e);%0A
-%09%09%7D%0A%09
+        %7D%0A    
 %7D%0A%7D

use joda optional dateTimeparser instead of dateOptionalTime one