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

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.batchprocess.ProcessStatistics;
import gov.va.med.fw.scheduling.ScheduledProcessInvocationContext;
import gov.va.med.fw.util.StringUtils;
import gov.va.med.fw.util.ThreadPool;

import java.util.List;

import org.apache.commons.lang.Validate;

public class MVIAddressExportProcess extends AbstractDataQueryIncrementalProcess {
	
   //multithreading    
    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; 
 
	
	protected void executeProcess(DataProcessExecutionContext context) throws Exception {		
		String args = (String)context.getExecutionArguments();
		if (!StringUtils.isEmpty(args)) {
			setThreadPoolSize(args);
		}
		super.executeProcess(context);
	}
	
	private void setThreadPoolSize(String args) throws Exception {	
		try {
			Integer customizedPoolSize = new Integer(args);
			this.setThreadPoolSize(customizedPoolSize.intValue());
			if (logger.isDebugEnabled()) {
				logger.debug("MVIAddressExportProcess -- Setting threadpool size to " + customizedPoolSize);
			}
		} catch(Exception e) {
			if (logger.isErrorEnabled()) {
				logger.error("MVIAddressExportProcess -- Invalid process argument: " + args + ".", e);
			}
		}
	}
	
	protected void processData(DataQueryProcessExecutionContext context,
			List acquiredData) {

		if (acquiredData == null){
			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[] row = (Object[]) acquiredData.get(i);
			processEntityData(context, row);
 		}	
	}

	public void processEntityData(DataQueryProcessExecutionContext context, Object[] acquiredData) {
		
		if(isThreaded()) {
			spawnThread(context, acquiredData);
		}
		
	}
	private void spawnThread(DataQueryProcessExecutionContext context,
			Object[] 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 once 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
			MVIAddressExportProcessSpawnedThreadTask task =
				(MVIAddressExportProcessSpawnedThreadTask) getApplicationContext().getBean(spawnedTaskId);

			task.setContext(context);
			task.setAcquiredData(acquiredData);	
			task.setAuditInfo(getAuditInfo()); 

			threadPool.invokeLater(task);
		} catch(InterruptedException e) {
				throwIllegalStateException("MVIAddressExportProcessSpawnedThreadTask was interrupted while it was " +
						"spawning a thread. ", e); 
		};
	}

	 private void cleanThreadPool(DataProcessExecutionContext context)
	    {
	        ThreadPool threadPool = getThreadPool(context);
	        threadPool.stop();
	        context.getContextData().put(CONTEXT_THREAD_POOL, null);
	    }
	 
	protected ProcessStatistics createProcessStatistics() {
		return new MVIAddressExportProcessStatistics();
	}	

	protected void handleDataProcessCompleted(DataProcessExecutionContext  context) {
			if(isThreaded()) {
				try {
					if(getTaskCount(context) != 0) {
						synchronized(this) {
							boolean stillProcessing = true;
							while(stillProcessing) {
								wait();
								if(getTaskCount(context) == 0)
									stillProcessing = false;
							}
						}
					}
				} catch(InterruptedException e) {
					throwIllegalStateException("MVIAddressExportProcess 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 
			this.updateJobResult(context);				
            ((ExceptionWriterProcessCompletedHandler)getDataProcessCompletedHandler()).appendExceptionData(context);
			
	       super.handleDataProcessCompleted(context);
		}
	
	protected boolean shouldUpdateJobResult(DataQueryProcessExecutionContext context) {
			return context.getProcessStatistics().isTotalNumberMod(DEFAULT_JOB_RESULT_UPDATE_INTERVAL);
	}

    private boolean shouldWriteExceptionData(DataQueryProcessExecutionContext context) {
        int exceptionDataSize = (context.getExceptionData()==null) ? 0 : context.getExceptionData().size();
	        return exceptionDataSize != 0 ? (exceptionDataSize % DEFAULT_EXCEPTION_UPDATE_INTERVAL) == 0 : false;
	    }


	public int getThrottleTaskCountThreshold() {
		return throttleTaskCountThreshold;
	}

	public void setThrottleTaskCountThreshold(int throttleTaskCountThreshold) {
		this.throttleTaskCountThreshold = throttleTaskCountThreshold;
	}

	public int getSpawnRetryPeriod() {
		return spawnRetryPeriod;
	}

	public void setSpawnRetryPeriod(int spawnRetryPeriod) {
		this.spawnRetryPeriod = spawnRetryPeriod;
	}
	
	private ThreadPool getThreadPool(DataProcessExecutionContext context) {
		ThreadPool threadPool = (ThreadPool) context.getContextData().get(CONTEXT_THREAD_POOL);
		if(threadPool == null) {
			threadPool = new ThreadPool("MVIAddressExportProcessSpawnedThreadTask", 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 isThreaded() {
		return StringUtils.isNotBlank(spawnedTaskId);
	}

	public String getSpawnedTaskId() {
		return spawnedTaskId;
	}

	public void setSpawnedTaskId(String spawnedTaskId) {
		this.spawnedTaskId = spawnedTaskId;
	}
	
	/**
	 * @return Returns the threadPoolSize.
	 */
	public int getThreadPoolSize() {
		return threadPoolSize;
	}

	/**
	 * @param threadPoolSize The threadPoolSize to set.
	 */
	public void setThreadPoolSize(int threadPoolSize) {
		this.threadPoolSize = threadPoolSize;
	}

	
}
