package org.keedio.flume.source;

import java.util.List;

import org.hibernate.Session;
import org.hibernate.SessionFactory;
import org.hibernate.boot.registry.StandardServiceRegistryBuilder;
import org.hibernate.cfg.Configuration;
import org.hibernate.service.ServiceRegistry;
import org.hibernate.transform.Transformers;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Helper class to manage hibernate sessions and perform queries
 * 
 * @author <a href="mailto:mvalle@keedio.com">Marcelo Valle</a>
 *
 */
public class HibernateHelper {

	private static final Logger LOG = LoggerFactory
			.getLogger(HibernateHelper.class);

	private static SessionFactory factory;
	private Session session;
	private Configuration config;
	private SQLSourceHelper sqlSourceHelper;

	/**
	 * Constructor to initialize hibernate configuration parameters
	 * @param sqlSourceHelper Contains the configuration parameters from flume config file
	 */
	public HibernateHelper(SQLSourceHelper sqlSourceHelper) {

		this.sqlSourceHelper = sqlSourceHelper;

		config = new Configuration()
				.setProperty("hibernate.connection.url", sqlSourceHelper.getConnectionURL())
				.setProperty("hibernate.connection.username", sqlSourceHelper.getUser())
				.setProperty("hibernate.connection.password", sqlSourceHelper.getPassword());
		
		if (sqlSourceHelper.getHibernateDialect() != null)
			config.setProperty("hibernate.dialect", sqlSourceHelper.getHibernateDialect());
		if (sqlSourceHelper.getHibernateDriver() != null)
			config.setProperty("hibernate.connection.driver_class", sqlSourceHelper.getHibernateDriver());
	}

	/**
	 * Connect to database using hibernate
	 */
	public void establishSession() {

		LOG.info("Opening hibernate session");

		ServiceRegistry serviceRegistry = new StandardServiceRegistryBuilder()
				.applySettings(config.getProperties()).build();
		factory = config.buildSessionFactory(serviceRegistry);
		session = factory.openSession();
	}

	/**
	 * Close database connection
	 */
	public void closeSession() {

		LOG.info("Closing hibernate session");

		session.close();
	}

	/**
	 * Execute the selection query in the database
	 * @return The query result. Each Object is a cell content. <p>
	 * The cell contents use database types (date,int,string...), 
	 * keep in mind in case of future conversions/castings.
	 */
	@SuppressWarnings("unchecked")
	public List<List<Object>> executeQuery() {

		List<List<Object>> rowsList = session
				.createSQLQuery(sqlSourceHelper.getQuery())
				.setFirstResult(sqlSourceHelper.getCurrentIndex())
				.setMaxResults(sqlSourceHelper.getMaxRows())
				.setResultTransformer(Transformers.TO_LIST).list();				

		sqlSourceHelper.setCurrentIndex(sqlSourceHelper.getCurrentIndex()
				+ rowsList.size());

		return rowsList;
	}

	public void resetConnectionAndSleep() throws InterruptedException {
		
		long startTime = System.currentTimeMillis();
		
		session.close();
		session = factory.openSession();
		
		long execTime = System.currentTimeMillis() - startTime;
		
		if (execTime < sqlSourceHelper.getRunQueryDelay())
			Thread.sleep(sqlSourceHelper.getRunQueryDelay() - execTime);
	}	
}

@@ -704,16 +704,58 @@
 ession;%0A
+%09private ServiceRegistry serviceRegistry;%0A
 %09private
@@ -777,16 +777,16 @@
 config;%0A
-
 %09private
@@ -1779,32 +1779,16 @@
 n%22);%0A%0A%09%09
-ServiceRegistry 
 serviceR
@@ -2110,24 +2110,43 @@
 on.close();%0A
+%09%09factory.close();%0A
 %09%7D%0A%0A%09/**%0A%09 *
@@ -2705,20 +2705,16 @@
 .list();
-%09%09%09%09
 %0A%0A%09%09sqlS
@@ -2966,38 +2966,44 @@
 se();%0A%09%09
-session = factory.open
+factory.close();%0A%09%09establish
 Session(
@@ -3190,10 +3190,9 @@
 e);%0A
+
 %09%7D
-%09
 %0A%7D%0A

Fix #15, avoided multiple connections to database