package gov.va.med.esr.common.batchprocess;

import gov.va.med.esr.service.LookupService;
import gov.va.med.fw.batchprocess.AbstractDataFileIncrementalProcess;
import gov.va.med.fw.batchprocess.CampLejeuneImportProcessStatistics;
import gov.va.med.fw.batchprocess.DataFileProcessExecutionContext;
import gov.va.med.fw.batchprocess.DataProcessExecutionContext;
import gov.va.med.fw.batchprocess.ProcessStatistics;
import gov.va.med.fw.util.ThreadPool;

import java.io.PrintWriter;
import java.io.StringWriter;
import java.util.ArrayList;
import java.util.List;

import org.apache.commons.lang.StringUtils;

/**
 * Batch process for importing CLEAR Database data and determining 
 * enrollment & eligibility for the records 
 */
public class CampLejeuneImportProcess extends
        AbstractDataFileIncrementalProcess
{

    LookupService lookupService;

    private static final int DEFAULT_THREAD_POOL_SIZE = 10;
	private static final int DEFAULT_THROTTLE_TASKCOUNT_THREASHOLD = 100;
	private static final int DEFAULT_SPAWN_RETRY_PERIOD = 3000;
	private static final int DEFAULT_EXCEPTION_UPDATE_INTERVAL = 20;
	
	public static final String CONTEXT_THREAD_CREATOR = "threadCreator";
	public static final String CONTEXT_TASK_COUNT = "taskCount";
	public static final String CONTEXT_THREAD_POOL = "threadPool";	

	private int threadPoolSize = DEFAULT_THREAD_POOL_SIZE;

	private int throttleTaskCountThreshold = DEFAULT_THROTTLE_TASKCOUNT_THREASHOLD;
	
	private int spawnRetryPeriod = DEFAULT_SPAWN_RETRY_PERIOD;

	private String spawnedTaskId;
    
    public String getSpawnedTaskId() {
		return spawnedTaskId;
	}

	public void setSpawnedTaskId(String spawnedTaskId) {
		this.spawnedTaskId = spawnedTaskId;
	}

	public int getSpawnRetryPeriod() {
		return spawnRetryPeriod;
	}

	public void setSpawnRetryPeriod(int spawnRetryPeriod) {
		this.spawnRetryPeriod = spawnRetryPeriod;
	}

	public int getThreadPoolSize() {
		return threadPoolSize;
	}

	public void setThreadPoolSize(int threadPoolSize) {
		this.threadPoolSize = threadPoolSize;
	}

	public int getThrottleTaskCountThreshold() {
		return throttleTaskCountThreshold;
	}

	public void setThrottleTaskCountThreshold(int throttleTaskCountThreshold) {
		this.throttleTaskCountThreshold = throttleTaskCountThreshold;
	}
	
	/** override so that the error message can be redirected to detail unsuccessful report */
	public boolean beanCreationFailure(DataFileProcessExecutionContext context, String nextRowRawData,  
			Exception e) {
		CampLejeuneImportProcessStatistics stats = (CampLejeuneImportProcessStatistics)context.getProcessStatistics();
		stats.incrementNumberOfErrorRecords();

		
		stats.addDetailUnsuccess(getIdData(context.getCurrentRowRawData()),  e.getMessage()); 
		
/*		if (logger.isErrorEnabled()) {
	        StringWriter sw = new StringWriter();
	        e.printStackTrace(new PrintWriter(sw)); //get the StackTrace
			logger.error(this.ERROR_KEYWORD + context.getCurrentRowRawData() + ".\n" + e.getMessage() + ".\n" + sw.toString());
		}*/
		//context.getExceptionData().add(ERROR_KEYWORD + context.getCurrentRowRawData() + ".\n" + e.getMessage() + ".\n" + e.getStackTrace().toString());
		return true;
	}
	
	private String getIdData(String rawData) {
		if (StringUtils.isEmpty(rawData))
				return "";
		
		String[] data = rawData.split("\\|");
		
		return new StringBuffer().append(CampLejeuneImportProcessStatistics.DELIMITER). //empty icn
			append(data[3]).append(CampLejeuneImportProcessStatistics.DELIMITER). //first name
			append(data[2]).append(CampLejeuneImportProcessStatistics.DELIMITER). //last name
			append(data[5]).append(CampLejeuneImportProcessStatistics.DELIMITER). //ssn
			append(data[7]).append(CampLejeuneImportProcessStatistics.DELIMITER). //dob
			append(data[8]).toString(); //gender
	}
	/*
	 * (non-Javadoc)
	 * 
	 * @see gov.va.med.fw.batchprocess.AbstractDataProcess#processData(gov.va.med.fw.batchprocess.DataProcessExecutionContext,
	 *      java.util.List)
	 */
	public  void processDataRecords(DataFileProcessExecutionContext context,
			List acquiredData) {

		if(acquiredData == null || acquiredData.isEmpty())
			return;
		
		//import header line: remove it from the acquiredData so it will not be processed
		//acquiredData.remove(0);
		Object oLine=acquiredData.get(0);
		String sLine=null;
		if(oLine!=null)
		{
			sLine=oLine.toString();
			if(sLine.length()>0)
			{
				if(sLine.contains("First_Name") &&(sLine.contains("Last_Name")))
				{
					acquiredData.remove(0);
				}
			}
			 
		}
	
         
    	spawnThread(context, acquiredData);
 	}
    
	private void spawnThread(DataFileProcessExecutionContext context,
			List acquiredData) {
		
		

		initThreadCreatorAndIncrementTaskCount(context);
		ThreadPool threadPool = getThreadPool(context);
		// get a new Thread bean
		CampLejeuneImportProcessSpawnedThreadTask task =
			(CampLejeuneImportProcessSpawnedThreadTask) getApplicationContext().getBean(spawnedTaskId);
		task.setContext(context);
		task.setAuditInfo(getAuditInfo());
		
		List taskData = new ArrayList();
		taskData.addAll(acquiredData);
		task.setDataRecords(taskData);
		
		threadPool.invokeLater(task);
	
	
	}
	
	protected void handleDataProcessCompleted(DataProcessExecutionContext context) {
		try {
			if (getTaskCount(context) != 0) {
				synchronized (this) {
					boolean stillProcessing = true;
					while (stillProcessing) {
						wait();
						if (getTaskCount(context) == 0)
							stillProcessing = false;
					}
				}
			}
		} catch (InterruptedException e) {
			throwIllegalStateException(
					"CampLejeuneImportProcess was interrupted while it was waiting for "
							+ "its spawned threads to complete", e);
		} finally {
			getThreadPool(context).stop();
			cleanThreadPool(context);
		}			

		super.handleDataProcessCompleted(context);
     
	}
	
	
    /**
     * stops the threapool and removes it for context so that next batch will 
     * start with a fresh threadpool.
     *
     */
    private void cleanThreadPool(DataProcessExecutionContext context)
    {
        ThreadPool threadPool = getThreadPool(context);
        threadPool.stop();
        context.getContextData().put(CONTEXT_THREAD_POOL, null);
        
		if (logger.isDebugEnabled()) {
			logger.debug("CampLejeuneImportProcessSpawnedThreadTask cleaned up the thread pool");
		}

    }
	
	private ThreadPool getThreadPool(DataProcessExecutionContext context) {
		ThreadPool threadPool = (ThreadPool) context.getContextData().get(CONTEXT_THREAD_POOL);
		if(threadPool == null) {
			threadPool = new ThreadPool("CampLejeuneImportProcessSpawnedThreadTask", threadPoolSize);
			context.getContextData().put(CONTEXT_THREAD_POOL, threadPool);
		}
		return threadPool;
	}
	
	private void initThreadCreatorAndIncrementTaskCount(DataFileProcessExecutionContext context) {
		if(!context.getContextData().containsKey(CONTEXT_THREAD_CREATOR))
			context.getContextData().put(CONTEXT_THREAD_CREATOR, this);
							
		adjustTaskCount(context, 1);
	}
	
	static int getTaskCount(DataProcessExecutionContext context) {
		Integer count = (Integer) context.getContextData().get(CONTEXT_TASK_COUNT);
		return count != null ? count.intValue() : 0;
	}

	static void adjustTaskCount(DataProcessExecutionContext context, int adjustment) {		
		synchronized(context) {
			context.getContextData().put(CONTEXT_TASK_COUNT, new Integer(getTaskCount(context) + adjustment));
		}
	}

    private boolean shouldWriteExceptionData(DataProcessExecutionContext context) {
        int exceptionDataSize = (context.getExceptionData()==null) ? 0 : context.getExceptionData().size();
        return exceptionDataSize != 0 ? (exceptionDataSize % DEFAULT_EXCEPTION_UPDATE_INTERVAL) == 0 : false;
    }
    
	protected ProcessStatistics createProcessStatistics() {
        return new CampLejeuneImportProcessStatistics ();
    } 

}