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

import gov.va.med.esr.service.LookupService;
import gov.va.med.fw.batchprocess.AbstractDataQueryIncrementalProcess;
import gov.va.med.fw.batchprocess.DataProcessExecutionContext;
import gov.va.med.fw.batchprocess.DataQueryProcessExecutionContext;
import gov.va.med.fw.batchprocess.ExceptionWriterProcessCompletedHandler;
import gov.va.med.fw.util.StringUtils;
import gov.va.med.fw.util.ThreadPool;

import java.math.BigDecimal;
import java.util.List;

/**
 * Deployment batch process for a new HBP rule, 
 */
public class HBPRuleDeploymentProcess extends
        AbstractDataQueryIncrementalProcess
{

    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;
	}
	
	/*
	 * (non-Javadoc)
	 * 
	 * @see gov.va.med.fw.batchprocess.AbstractDataProcess#processData(gov.va.med.fw.batchprocess.DataProcessExecutionContext,
	 *      java.util.List)
	 */
	protected  void processData(DataQueryProcessExecutionContext context,
			List acquiredData) {

	   if(logger.isDebugEnabled())
	       logger.debug(getProcessName() + ": Query increment result Size="+(acquiredData == null ? 0 : acquiredData.size()));

       if(acquiredData == null || acquiredData.isEmpty())
		return;
        	        
        // Add check for isInterrupted so don't have to wait for entire batch to finish
        for (int i = 0; i < acquiredData.size() && !isInterrupted(context); i++) {
            // Object[] returned from get when scrolling (idempotent is false)
            Object[] row = (Object[]) acquiredData.get(i);
            BigDecimal personId = (BigDecimal) row[0];
            
            if(logger.isDebugEnabled())
                logger.debug("Processing person # " + (i+1) + " with id = " + personId);
            
 			 processEntityData(context, personId);
        }
 	}
        
	public void processEntityData(DataQueryProcessExecutionContext context, BigDecimal acquiredData) {
		
		if(isThreaded()) {
			spawnThread(context, acquiredData);
		}
		
	}

	private boolean isThreaded() {
		return StringUtils.isNotBlank(spawnedTaskId);
	}

    
	private void spawnThread(DataQueryProcessExecutionContext context,
			BigDecimal acquiredData) {
		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))
	            {
	                ((ExceptionWriterProcessCompletedHandler)getDataProcessCompletedHandler()).appendExceptionData(context);
	            }
			}
			//Update the job result one in a while.
			if(shouldUpdateJobResult(context))
				this.updateJobResult(context);				


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

			initThreadCreatorAndIncrementTaskCount(context);
			ThreadPool threadPool = getThreadPool(context);
			// get a new Thread bean
			HBPRuleDeploymentProcessSpawnedThreadTask task =
				(HBPRuleDeploymentProcessSpawnedThreadTask) getApplicationContext().getBean(spawnedTaskId);
			task.setContext(context);
			task.setAuditInfo(getAuditInfo());
			
			task.setAcquiredData(acquiredData);	
			
			threadPool.invokeLater(task);
		} catch(InterruptedException e) {
				throwIllegalStateException("HBPRuleDeploymentProcessSpawnedThreadTask 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(
					"HBPRuleDeploymentProcess 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 once in a while.
		if (shouldUpdateJobResult((DataQueryProcessExecutionContext)context))
			this.updateJobResult(context);				

		// Write the last batch of exception data to the file
        if ( shouldWriteExceptionData(context))
        {
            ((ExceptionWriterProcessCompletedHandler)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("HBPRuleDeploymentProcessSpawnedThreadTask 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("HBPRuleDeploymentProcessSpawnedThreadTask", threadPoolSize);
			context.getContextData().put(CONTEXT_THREAD_POOL, threadPool);
		}
		return threadPool;
	}
	
	private void initThreadCreatorAndIncrementTaskCount(DataQueryProcessExecutionContext 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;
    }


}