package org.jerrymouse.database.hadoop.jpa;

import java.io.IOException;
import java.lang.reflect.Constructor;
import java.lang.reflect.Field;
import java.lang.reflect.InvocationTargetException;
import java.util.Date;
import java.util.UUID;

import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.util.Bytes;

public class HadoopEntity extends HadoopBean {
	HadoopTable hadoopTable;
	Object object;

	public HadoopEntity(Class type, HadoopTable hadoopTable) {
		super(type);
		this.hadoopTable = hadoopTable;
		try {
			Constructor constructor = type.getConstructor(null);
			constructor.setAccessible(true);
			object = constructor.newInstance();
		} catch (IllegalArgumentException e) {
			e.printStackTrace();
		} catch (InstantiationException e) {
			e.printStackTrace();
		} catch (IllegalAccessException e) {
			e.printStackTrace();
		} catch (InvocationTargetException e) {
			e.printStackTrace();
		} catch (SecurityException e) {
			e.printStackTrace();
		} catch (NoSuchMethodException e) {
			e.printStackTrace();
		}
	}

	public HadoopEntity(Object entity, HadoopTable hadoopTable) {
		super(entity.getClass());
		object = entity;
		this.hadoopTable = hadoopTable;
	}

	private void buildID() {
		try {
			if (!hasID()) {
				Long idV = UUID.randomUUID().getMostSignificantBits();
				id.set(object, idV);
			}
		} catch (IllegalArgumentException e) {
			e.printStackTrace();
		} catch (IllegalAccessException e) {
			e.printStackTrace();
		}
	}

	public void find(Object raw) {
		try {
			Get get = new Get(Bytes.toBytes((Long) raw));
			set(id, Bytes.toBytes((Long) raw));
			Result result = hadoopTable.getTable(type).get(get);
			for (Field field : fields) {
				set(field, result.getValue(Bytes.toBytes(field.getName()),
						Bytes.toBytes("")));
			}
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	public byte[] get(Field field) {
		byte[] bytes = null;
		Object value = null;
		try {
			value = field.get(object);
		} catch (IllegalArgumentException e) {
			e.printStackTrace();
		} catch (IllegalAccessException e) {
			e.printStackTrace();
		}
		if (value instanceof String) {
			bytes = Bytes.toBytes((String) value);
		} else if (value instanceof Long) {
			bytes = Bytes.toBytes((Long) value);
		} else if (value instanceof Date) {
			bytes = Bytes.toBytes(((Date) value).getTime());
		}
		return bytes;
	}

	public Object getObject() {
		return object;
	}

	private boolean hasID() {
		Object value = null;
		try {
			value = id.get(object);
		} catch (IllegalArgumentException e) {
			e.printStackTrace();
		} catch (IllegalAccessException e) {
			e.printStackTrace();
		}
		return value != null;
	}

	public void put() {
		try {
			buildID();
			Put p = new Put(Bytes.toBytes((Long) id.get(object)));
			for (Field field : fields) {
				String name = field.getName();
				p.add(Bytes.toBytes(name), Bytes.toBytes(""), get(field));
			}
			hadoopTable.getTable(type).put(p);
		} catch (IOException e) {
			e.printStackTrace();
		} catch (IllegalArgumentException e) {
			e.printStackTrace();
		} catch (IllegalAccessException e) {
			e.printStackTrace();
		}
	}

	public void remove() {
		if (!hasID()) {
			return;
		} else {
			try {
				Delete delete = new Delete(Bytes.toBytes((Long) id.get(object)));
				hadoopTable.getTable(type).delete(delete);
			} catch (IOException e) {
				e.printStackTrace();
			} catch (IllegalArgumentException e) {
				e.printStackTrace();
			} catch (IllegalAccessException e) {
				e.printStackTrace();
			}
		}
	}

	public void set(Field field, byte[] bytes) {
		try {
			if (Long.class.isAssignableFrom(field.getType())) {
				field.set(object, Bytes.toLong(bytes));
			} else if (String.class.isAssignableFrom(field.getType())) {
				field.set(object, Bytes.toString(bytes));
			} else if (Date.class.isAssignableFrom(field.getType())) {
				field.set(object, new Date(Bytes.toLong(bytes)));
			}
		} catch (IllegalArgumentException e) {
			e.printStackTrace();
		} catch (IllegalAccessException e) {
			e.printStackTrace();
		}
	}

	public void set(String key, byte[] value) {
		try {
			Field field = type.getDeclaredField(key);
			field.setAccessible(true);
			set(field, value);
		} catch (SecurityException e) {
			e.printStackTrace();
		} catch (NoSuchFieldException e) {
			e.printStackTrace();
		}

	}

}
