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


import java.math.BigDecimal;

import java.util.List;

import org.apache.commons.lang.StringUtils;

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.util.ThreadPool;


public class GenerateQRYZ21Process extends AbstractDataQueryIncrementalProcess{    
    
    
    
    //multithreading    
    private static final int DEFAULT_THREAD_POOL_SIZE = 10;
	private static final int DEFAULT_THROTTLE_TASKCOUNT_THREASHOLD = 500;
	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 GENERATE_QUERYZ21_PROCESS_THREAD_NAME = "GENQUERYZ21THREAD";
	

	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 processData(DataQueryProcessExecutionContext context, List acquiredData) {
        if(logger.isDebugEnabled())
            logger.debug(/*Generate QRY-Z21:*/ getProcessName() + ": Query increment result Size="+(acquiredData == null ? 0 : acquiredData.size()));
        
        if (acquiredData == null)
            return;
        
        Object args = context.getExecutionArguments();
        if(args != null){
        	super.setFetchSize(1);
        	String singleArg = (String) args;
        	BigDecimal[] personArray ;
        	personArray = new BigDecimal[2];
        	personArray[0] = new BigDecimal(singleArg);        	
        	acquiredData.clear();
        	acquiredData.add(personArray); 
        }
        
        // 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);
            
            // Update statistics more frequently than once per batch (default size 500)
            if(context.getProcessStatistics().isTotalNumberMod(DEFAULT_JOB_RESULT_UPDATE_INTERVAL))
                this.updateJobResult(context); 
            
            //interupt after the first set of fetch if execution was done with arguments
            if(args != null){
            	context.setInterrupted(true);            	
            }
        }
        
        if(logger.isDebugEnabled()) {
            logger.debug(/*Generate QRY-Z21:*/ getProcessName() + ": Processing of current batch complete.");
            logger.debug(/*Generate QRY-Z21:*/ getProcessName() + ": Successful records count = "+context.getProcessStatistics().getNumberOfSuccessfulRecords());
            logger.debug(/*Generate QRY-Z21:*/ getProcessName() + ": Failure records count ="+context.getProcessStatistics().getNumberOfErrorRecords());
        }
    } 
	
	public void processEntityData(DataQueryProcessExecutionContext context, BigDecimal acquiredData) {
		
		if(isThreaded()) {
			spawnThread(context, acquiredData);
		}
		
	}
	
	private void spawnThread(DataQueryProcessExecutionContext context,
			BigDecimal acquiredData) {
		try {
			while (getTaskCount(context) >= throttleTaskCountThreshold) {
				Thread.sleep(spawnRetryPeriod);
			}
			initThreadCreatorAndIncrementTaskCount(context);
			ThreadPool threadPool = getThreadPool(context);
			// get a new Thread bean
			GenerateQRYZ21ProcessSpawnedThreadTask task =
				(GenerateQRYZ21ProcessSpawnedThreadTask) getApplicationContext().getBean(spawnedTaskId);
			task.setContext(context);
			task.setAuditInfo(getAuditInfo());
			task.setAcquiredData(acquiredData);	
			task.setProcessName(getProcessName());
			
			threadPool.invokeLater(task);
		} catch(InterruptedException e) {
				throwIllegalStateException("GenerateQRYZ21ProcessSpawnedThreadTask 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 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("GenerateQRYZ21Process was interrupted while it was waiting for " +
							"its spawned threads to complete", e);
				} finally {
					getThreadPool(context).stop();
	                cleanThreadPool(context);
				}
			}

	       super.handleDataProcessCompleted(context);
	     
		}
	
	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("GenerateQRYZ21ProcessSpawnedThreadTask", 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;
	}

	
    
    
}
