/********************************************************************
 * Copyright © 2005 VHA. All rights reserved
 ********************************************************************/

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

import java.util.List;

import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang.Validate;

import gov.va.med.esr.common.batchprocess.datasync.HECLegacyDataSynchronizationConstants;
import gov.va.med.esr.common.model.CommonEntityKeyFactory;
import gov.va.med.esr.common.model.ee.PrisonerOfWar;
import gov.va.med.esr.common.model.ee.PurpleHeart;
import gov.va.med.esr.common.model.person.Person;
import gov.va.med.esr.common.model.person.id.PersonEntityKey;
import gov.va.med.esr.common.model.registry.Registry;
import gov.va.med.esr.common.model.registry.RegistryTrait;
import gov.va.med.esr.service.PersonService;
import gov.va.med.esr.service.RegistrySearchCriteria;
import gov.va.med.esr.service.RegistrySearchResultBean;
import gov.va.med.esr.service.RegistryService;
import gov.va.med.fw.batchprocess.AbstractDataFileMREProcess;
import gov.va.med.fw.batchprocess.DataFileProcessExecutionContext;
import gov.va.med.fw.batchprocess.DataProcessExecutionContext;
import gov.va.med.fw.service.ServiceException;
import gov.va.med.fw.util.ThreadPool;
import gov.va.med.fw.util.builder.Builder;
import gov.va.med.fw.util.builder.BuilderException;


public class HandBookBatchRequestConsumerProcess extends
		AbstractDataFileMREProcess {
	private static final int DEFAULT_THREAD_POOL_SIZE = 1;
	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 HECLEGACY_DATASYNC_CONSUMER_THREAD_NAME = "HECLEGACYCONSUMERTHREAD";
	

	private int threadPoolSize = DEFAULT_THREAD_POOL_SIZE;

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

	private String spawnedTaskId;  
	
	private PersonService personService;

	private RegistryService registryService;

	private Builder mergeBuilder;

	
	private ThreadPool getThreadPool(DataProcessExecutionContext context) {
		ThreadPool threadPool = (ThreadPool) context.getContextData().get(CONTEXT_THREAD_POOL);
		if(threadPool == null) {
			threadPool = new ThreadPool("HandBookBatchRequestConsumerSpawnedThreadTask", 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));
		}
	}
	
	public void processEntityData(DataFileProcessExecutionContext context,
			List acquiredData) {
		
		if(context==null){
		 context = createDataFileProcessExecutionContext();
		}
		spawnThread(context, acquiredData);
		
	}

	
	public void afterPropertiesSet() {
		super.afterPropertiesSet();
		
	}

	
	/**
	 * @return Returns the spawnedTaskId.
	 */
	public String getSpawnedTaskId() {
		return spawnedTaskId;
	}

	/**
	 * @param spawnedTaskId The spawnedTaskId to set.
	 */
	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;
	}
	
	private boolean isThreaded() {
		return StringUtils.isNotBlank(spawnedTaskId);
	}
	
	 
	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("HECLegacyDataSynchronizationConsumerProcess 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);
    }
    
	private void spawnThread(DataFileProcessExecutionContext context,
			List acquiredData) {
		try {
			while (getTaskCount(context) >= throttleTaskCountThreshold) {
				Thread.sleep(spawnRetryPeriod);
			}
			initThreadCreatorAndIncrementTaskCount(context);
			ThreadPool threadPool = getThreadPool(context);
			// get a new Thread bean
			HandBookBatchRequestConsumerSpawnedThreadTask task =
				(HandBookBatchRequestConsumerSpawnedThreadTask) getApplicationContext().getBean(spawnedTaskId);
			task.setContext(context);
			task.setAuditInfo(getAuditInfo());
			task.setAcquiredData(acquiredData);
			
			threadPool.invokeLater(task);
		} catch(InterruptedException e) {
				throwIllegalStateException("HandBookBatchRequestConsumerProcess was interrupted while it was " +
						"spawning a thread. ", e);
		};
	}

	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;
	}
	protected boolean isNewEntityRow(String rowRawData) {
		return  false;
	}

	protected DataFileProcessExecutionContext createDataFileProcessExecutionContext() {
		DataFileProcessExecutionContext context = new DataFileProcessExecutionContext();
		return context;
	}
}
