package org.jerrymouse.database.hadoop.jpa;

import java.io.IOException;
import java.lang.reflect.Field;
import java.util.List;

import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.MasterNotRunningException;
import org.apache.hadoop.hbase.TableNotFoundException;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.util.Bytes;

public class HadoopTable {
	public static boolean autoCreateTable;
	private HBaseAdmin hBaseAdmin;
	private HBaseConfiguration hBaseConfiguration;

	public HadoopTable(HBaseAdmin hBaseAdmin,
			HBaseConfiguration hBaseConfiguration) {
		super();
		this.hBaseAdmin = hBaseAdmin;
		this.hBaseConfiguration = hBaseConfiguration;
	}

	public void createTable(String tableName, List<Field> fields) {
		try {
			HTableDescriptor tableDescripter = new HTableDescriptor(Bytes
					.toBytes(tableName));
			for (Field field : fields) {
				tableDescripter
						.addFamily(new HColumnDescriptor(field.getName()));
			}
			hBaseAdmin.createTable(tableDescripter);
		} catch (MasterNotRunningException e) {
			e.printStackTrace();
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	void createTables(List<Class> list) {
		for (Class type : list) {
			HadoopBean bean = new HadoopBean(type);
			createTable(bean.getTableName(), bean.getFields());
		}
	}

	public void deleteTable(String tableName) {
		try {
			hBaseAdmin.disableTable(tableName);
			hBaseAdmin.deleteTable(tableName);
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	void deleteTables(List<Class> list) {
		for (Class type : list) {
			HadoopBean bean = new HadoopBean(type);
			deleteTable(bean.getTableName());
		}
	}

	HTable getTable(Class type) throws IOException {
		HTable hTable = null;
		HadoopBean hadoopBean = new HadoopBean(type);
		String tableName = hadoopBean.getTableName();
		try {
			hTable = new HTable(hBaseConfiguration, tableName);
		} catch (IOException e) {
			if (e instanceof TableNotFoundException) {
				createTable(tableName, hadoopBean.getFields());
				hTable=new HTable(hBaseConfiguration, tableName);
			} else
				e.printStackTrace();
		}
		return hTable;
	}

	void reCreateTables(List<Class> list) {
		deleteTables(list);
		createTables(list);
	}

	public boolean tableExists(String tableName) {
		try {
			return hBaseAdmin.tableExists(tableName);
		} catch (MasterNotRunningException e) {
			e.printStackTrace();
		}
		return false;
	}

	public void recreateTables(String entityPackage) {
		List<Class> list = HadoopBean.scanEntity(entityPackage);
	}

	public void checkTables(String entityPackage) {
		List<Class> list = HadoopBean.scanEntity(entityPackage);
		for (Class type : list) {
			HadoopBean bean = new HadoopBean(type);
			if (!tableExists(bean.getTableName()))
				createTable(bean.getTableName(), bean.getFields());
		}
	}
}
