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

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

import java.util.List;

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

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;

/**
 * Process that consumes objects built from input file from HEC Legacy. It
 * accepts data for one Person/Registry setting at a time.
 * 
 * Created Mar 7, 2006 3:30:33 PM
 * 
 * @author DNS   BOHMEG
 */
public class HECLegacyDataSynchronizationConsumerProcess extends
		AbstractDataFileMREProcess {
	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 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;

	protected boolean isNewEntityRow(String rowRawData) {
		return StringUtils.isNotBlank(rowRawData) ? rowRawData
				.startsWith(HECLegacyDataSynchronizationConstants.DataTypes.IDENTITY_DATA_TYPE) : false;
	}

	private ThreadPool getThreadPool(DataProcessExecutionContext context) {
		ThreadPool threadPool = (ThreadPool) context.getContextData().get(CONTEXT_THREAD_POOL);
		if(threadPool == null) {
			threadPool = new ThreadPool("HECLegacyDataSynchronizationConsumerSpawnedThreadTask", 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(isThreaded()) {
			spawnThread(context, acquiredData);
		} else {
			// ordering implied here for the first element (must be identity
			// information)
			HECLegacyIdentityFileData identityData = (HECLegacyIdentityFileData) acquiredData
					.get(0);
			try {
				if (identityData.isPersonUpdate()) {
					// retrieve Person
					Person person = matchPerson(identityData);
					// overlay data
					overlayFileDataOnPerson(person, acquiredData, identityData.isCommsOnly());
					// call Rules
					// Code CR7502 - If Comms Only change, do not trigger E&E calculations
					if (!identityData.isCommsOnly()) {
						executeRules(person);
					}
				} else {
					// retrieve Registry
					Registry registry = matchRegistry(identityData);
					// overlay data
					overlayFileDataOnRegistry(registry, acquiredData);
					// save Registry
					registryService.updateRegistry(registry);
				}
				context.getProcessStatistics().incrementNumberOfSuccessfulRecords();			
			} catch (Exception e) {
				handleFailure(context, identityData, e);
			}
		}
		
		// write out the exception list on a regular basis to avoid holding up memory space
		if (shouldWriteExceptionData(context)) {
			((HECLegacyDataSynchronizationConsumerProcessCompletedHandler)getDataProcessCompletedHandler()).appendExceptionData(context);
		}
	}

	private boolean shouldWriteExceptionData(DataFileProcessExecutionContext context) {
		int exceptionDataSize = (context.getExceptionData()==null) ? 0 : context.getExceptionData().size();
		return exceptionDataSize != 0 ? (exceptionDataSize % DEFAULT_EXCEPTION_UPDATE_INTERVAL) == 0 : false;
	}
	
	private void overlayFileDataOnPerson(Person person, List acquiredData, boolean isCommsOnly)
			throws Exception {
		Object[] args = new Object[3];
		args[0] = person;
		args[1] = acquiredData;
		args[2] = Boolean.valueOf(isCommsOnly);
		mergeBuilder.build(args);
	}

	private void overlayFileDataOnRegistry(Registry registry, List acquiredData)
			throws Exception {
		Object[] args = new Object[3];
		args[0] = registry;
		args[1] = acquiredData;
		mergeBuilder.build(args);
	}

	private void executeRules(Person person) throws ServiceException {
		personService.processDataSyncFromHECLegacy(person);
	}

	private Person matchPerson(HECLegacyIdentityFileData identityData)
			throws Exception {
		PersonEntityKey vpid = CommonEntityKeyFactory
				.createVPIDEntityKey(identityData.getICN());
		if (logger.isDebugEnabled())
			logger.debug("Retrieving Person by short-form VPID (ICN): " + vpid);
		// retrieve Person
		Person person = personService.getPerson(vpid);
		if (person == null)
			throw new RuntimeException("Unable to find Person for VPID: "
					+ vpid);
		return person;
	}

	private Registry matchRegistry(HECLegacyIdentityFileData identityData) throws Exception {
		RegistrySearchCriteria query = new RegistrySearchCriteria();
		query.setSsn(identityData.getSSN());
		query.setLastName(identityData.getLegalName().getFamilyName());
		query.setFirstName(identityData.getLegalName().getGivenName());
		query.setRegistryType(identityData.getRegistryType());
		
		Registry onFile = null;
		// TODO: make these 2 RegistryService api calls into one!
		List registryData = registryService.search(query);
		if(registryData.size() == 0) {
			if(identityData.getRegistryType() != null) {
				if(logger.isInfoEnabled())
					logger.info("Unable to find a matching Registry entry in ESR.....creating a new one");
				if(identityData.getRegistryType().isPrisonerOfWar())
					onFile = new PrisonerOfWar();
				else if(identityData.getRegistryType().isPurpleHeart())
					onFile = new PurpleHeart();
				else {
					throw new IllegalStateException("Unable to find a matching Registry entry in ESR...unable to create a new one " +
						"unsupported RegistryType");					
				}
				onFile.setRegistryTrait(new RegistryTrait());
				onFile.getRegistryTrait().setRegistryType(identityData.getRegistryType());				
			} else {
				throw new IllegalStateException("Unable to find a matching Registry entry in ESR...unable to create a new one " +
						"since HECLegacy sent null for RegistryType");
			}			
		} else if(registryData.size() == 1) {
			RegistrySearchResultBean searchResult = (RegistrySearchResultBean) registryData.get(0);
			onFile = registryService.getRegistryById(searchResult.getEntityKey(), searchResult.getRegistryType());			
		} else if(registryData.size() > 1) {
			throw new IllegalStateException("Can not find a singular match by Registry traits");
		}

		return onFile;
	}

	private void handleFailure(DataFileProcessExecutionContext context,
			HECLegacyIdentityFileData identityData, Exception e) {
		
		String exceptionText = null;
		
		if (e == null) {
			exceptionText = "Error: Unable to synchronize data for identity data: " + identityData;
		}
		else {
			if(e instanceof BuilderException) {
				// do this since reflection is being used...
				Throwable cause = e.getCause(); // typically this is a non-helpful reflection exception 
				Throwable causesCause = cause != null ? cause.getCause() : null;
				exceptionText = "Error: Unable to synchronize data for identity data: "
						+ identityData;
				if (causesCause != null) {
					exceptionText += " because of BuilderException with cause: " + cause + " and causes cause: " + causesCause;
				}
			}
			else {
				exceptionText = "Error: Unable to synchronize data for identity data: "
						+ identityData + " because of exception: " + e;		
			}
		}
		
		if (logger.isErrorEnabled()) {
			logger.error(exceptionText);
		}
		
		context.getProcessStatistics().incrementNumberOfErrorRecords();
		context.getExceptionData().add(identityData);
		context.getExceptionData().add(exceptionText + "\n\n");
	}

	/**
	 * @return Returns the personService.
	 */
	public PersonService getPersonService() {
		return personService;
	}

	/**
	 * @param personService
	 *            The personService to set.
	 */
	public void setPersonService(PersonService personService) {
		this.personService = personService;
	}

	public void afterPropertiesSet() {
		super.afterPropertiesSet();
		Validate.notNull(personService);
		Validate.notNull(registryService);
		Validate.notNull(mergeBuilder);
	}

	/**
	 * @return Returns the mergeBuilder.
	 */
	public Builder getMergeBuilder() {
		return mergeBuilder;
	}

	/**
	 * @param mergeBuilder
	 *            The mergeBuilder to set.
	 */
	public void setMergeBuilder(Builder mergeBuilder) {
		this.mergeBuilder = mergeBuilder;
	}

	/**
	 * @return Returns the registryService.
	 */
	public RegistryService getRegistryService() {
		return registryService;
	}

	/**
	 * @param registryService
	 *            The registryService to set.
	 */
	public void setRegistryService(RegistryService registryService) {
		this.registryService = registryService;
	}

	/**
	 * @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
			HECLegacyDataSynchronizationConsumerSpawnedThreadTask task =
				(HECLegacyDataSynchronizationConsumerSpawnedThreadTask) getApplicationContext().getBean(spawnedTaskId);
			task.setContext(context);
			task.setAuditInfo(getAuditInfo());
			task.setAcquiredData(acquiredData);
			
			threadPool.invokeLater(task);
		} catch(InterruptedException e) {
				throwIllegalStateException("HECLegacyDataSynchronizationConsumerProcess 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;
	}
    
	
}
