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

import gov.va.med.esr.service.DemographicService;
import gov.va.med.esr.service.LookupService;
import gov.va.med.fw.batchprocess.AbstractDataFileSingleRowIncrementProcess;
import gov.va.med.fw.batchprocess.DataFileProcessExecutionContext;
import gov.va.med.fw.batchprocess.DataProcessExecutionContext;
import gov.va.med.fw.util.ThreadPool;

import java.util.List;

/**
 * Batch process for the SSN Verification Response from SSA use case
 * 
 * @author Rajiv Patnaik Created on Mar 6, 2006
 * @version 1.0
 * 
 * Copyright © 2006 VHA. All rights reserved
 */
public class SSNVerificationResponseProcess extends
        AbstractDataFileSingleRowIncrementProcess
{

    LookupService lookupService;

    DemographicService demographicService;

    
	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";
	public static final String SSN_VER_RES_PROC_THREAD_NAME = "SSNVERRESPROCTHREAD";
	

	private int threadPoolSize = DEFAULT_THREAD_POOL_SIZE;

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

	private String spawnedTaskId;
    
    
    /**
     * @return Returns the demographicService.
     */
    public DemographicService getDemographicService()
    {
        return demographicService;
    }

    /**
     * @param demographicService
     *            The demographicService to set.
     */
    public void setDemographicService(DemographicService demographicService)
    {
        this.demographicService = demographicService;
    }

    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;
	}
	
	
	/*
	 * (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.isEmpty())
			return;
		
		String rawData = context.getCurrentRowRawData();
		try {
			Object obj = acquiredData.get(0);
			if (processDataRecord(context, obj)) {
				if(isKeepProcessedData())
					context.getProcessedData().add(obj);
			} else {
				context.getExceptionData().add(createExceptionObject(rawData));
			}
		} catch (RuntimeException e) {
			if(logger.isErrorEnabled())
				logger.error("Error processing data record: " + rawData, e);
			context.getProcessStatistics().incrementNumberOfErrorRecords();
			context.getExceptionData().add(createExceptionObject(rawData));
		}
	}

	/*
     * (non-Javadoc)
     * 
     * @see gov.va.med.fw.batchprocess.AbstractDataFileSingleRowIncrementProcess#processDataRecord(java.lang.Object)
     */
    protected boolean processDataRecord(
            DataFileProcessExecutionContext context, Object bean)
    {
        boolean success = false;
        SSASSNVerificationData dataRecord = (SSASSNVerificationData) bean;

         
    	spawnThread(context, dataRecord);
    	success = true;
    	return success;
    }
    
	private void spawnThread(DataFileProcessExecutionContext context,
			SSASSNVerificationData dataRecord) {
		try {
			while (getTaskCount(context) >= throttleTaskCountThreshold) {
				Thread.sleep(spawnRetryPeriod);
				
				//Update the job result and exception data while 
				// waiting.
				if(shouldUpdateJobResult(context))
					this.updateJobResult(context);				

	            if ( shouldWriteExceptionData(context))
	            {
	                ((SSNVerificationResponseProcessrProcessCompletedHandler)getDataProcessCompletedHandler()).appendExceptionData(context);
	            }
			}
			//Update the job result one in a while.
			if(shouldUpdateJobResult(context))
				this.updateJobResult(context);				


            if ( shouldWriteExceptionData(context))
            {
                ((SSNVerificationResponseProcessrProcessCompletedHandler)getDataProcessCompletedHandler()).appendExceptionData(context);
            }

			
			initThreadCreatorAndIncrementTaskCount(context);
			ThreadPool threadPool = getThreadPool(context);
			// get a new Thread bean
			SSNVerificationResponseProcessSpawnedThreadTask task =
				(SSNVerificationResponseProcessSpawnedThreadTask) getApplicationContext().getBean(spawnedTaskId);
			task.setContext(context);
			task.setAuditInfo(getAuditInfo());
			task.setDataRecord(dataRecord);
			
			threadPool.invokeLater(task);
		} catch(InterruptedException e) {
				throwIllegalStateException("SSNVerificationResponseProcessSpawnedThreadTask was interrupted while it was " +
						"spawning a thread. ", e);
		};
	}
	
	
	
	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(
					"HECLegacyDataSynchronizationConsumerProcess was interrupted while it was waiting for "
							+ "its spawned threads to complete", e);
		} finally {
			getThreadPool(context).stop();
			cleanThreadPool(context);
		}

		// update the job stats and write the exception data 
		//Update the job result one in a while.
		if(shouldUpdateJobResult(context))
			this.updateJobResult(context);				

		// Write the last batch of exception data to the file
        if ( shouldWriteExceptionData(context))
        {
            ((SSNVerificationResponseProcessrProcessCompletedHandler)getDataProcessCompletedHandler()).appendExceptionData(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("SSNVerificationResponseProcessSpawnedThreadTask 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("SSNVerificationResponseProcessSpawnedThreadTask", 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;
    }


}