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

import java.io.PrintWriter;
import java.io.StringWriter;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Hashtable;
import java.util.Iterator;
import java.util.List;
import java.util.Set;

import org.apache.commons.lang.Validate;

import gov.va.med.esr.common.model.CommonEntityKeyFactory;
import gov.va.med.esr.common.model.ivmdm.IVMDMExtractBatch;
import gov.va.med.esr.common.model.ivmdm.IVMDMExtractRecord;
import gov.va.med.esr.common.model.lookup.IVMDMType;
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.person.id.VPIDEntityKey;
import gov.va.med.esr.common.model.system.SystemParameter;
import gov.va.med.esr.service.IVMDMService;
import gov.va.med.esr.service.LookupService;
import gov.va.med.esr.service.PSDelegateService;
import gov.va.med.esr.service.SystemParameterService;
import gov.va.med.fw.batchprocess.ConcurrentProcess;
import gov.va.med.fw.batchprocess.DataQueryDetail;
import gov.va.med.fw.batchprocess.DataQueryProcessExecutionContext;
import gov.va.med.fw.scheduling.ScheduledProcessInvocationContext;
import gov.va.med.fw.service.ServiceException;
import gov.va.med.fw.util.StopWatchLogger;


/**
 * IVMDataSynchronizationProducerProcess
 * Created Apr 14, 2006 6:22:43 PM
 * @author DNS   MANSOG
 *
 */
public class IVMDataSynchronizationProducerProcess 
         extends AbstractDataSynchronizationProducerProcess {

    private static final String KEY_LAST_EXTRACTED_PERSON = "LAST_EXTRACTED_PERSON";
    private static final String KEY_BATCH_PERSON_HASH_TABLE = "KEY_BATCH_PERSON_HASH_TABLE";
    private static final String KEY_BATCH_PERSON_SITES_HASH_TABLE = "KEY_BATCH_PERSON_SITES_HASH_TABLE";
    public static final String EXTRACTSTARTDATEARG = "-extractstartdate=";
    private static final int DEFAULT_EXCEPTION_UPDATE_INTERVAL = 20;

    private PSDelegateService psDelegateService = null;
    private SystemParameterService systemParameterService = null;
    private LookupService lookupService = null;
    private IVMDMType ivmDmType = null;
    private IVMDataSynchronizationProducer ivmDataSyncProducer = null;
    private IVMDMService ivmDMService;
    private ConcurrentProcess personLoader = null;
    private int batchSize = 1000;
    private int numberofprocesses = 10;

    private static final int DEFAULT_BATCH_SIZE = 1000;
    private static final int DEFAULT_NUM_PROCESSES = 10;
    
    
   
    protected List executeQuery(DataQueryProcessExecutionContext context) throws Exception {
        
    	setEntityCountPerFile(getIvmMaxRecordsPerFile());
    	// read any execution args here.
        //expecting only these args -extractstartdate=yyyyMMddHHmmss
        String[] args = null;
        
        if (context.getExecutionArguments() != null )
        {
            if (context.getExecutionArguments() instanceof String[]) {
                args = (String[])context.getExecutionArguments();
            }
            else
            {
                //must be a single argument as String
                args = new String[]{(String)context.getExecutionArguments()};
            }
        }
        
        if ( args != null && args.length > 0 )
        {
            // parse the date
            String startDateStr = args[0].substring(EXTRACTSTARTDATEARG.length());
            Date extractStartDate = new SimpleDateFormat("yyyyMMddHHmmss").parse(startDateStr);
            
            if (extractStartDate != null )
            {
                context.getContextData().put(EXTRACTSTARTDATEARG, extractStartDate);
            }
        }
        
        return super.executeQuery(context);
    }
    
    
    /**
     * @see {@link gov.va.med.esr.common.batchprocess.datasync.AbstractDataSynchronizationProducerProcess#processData(DataQueryProcessExecutionContext, List)}
     */
    protected void processData(DataQueryProcessExecutionContext context, List acquiredData) {
        Iterator itr = acquiredData.iterator();
        Object acquiredObject = null;
        Object processedObject = null;
        Object transformedData = null;
        List data = new ArrayList();

        List shouldProcessAquiredObjects = new ArrayList();
        Set personKeys = new HashSet();

       if (logger.isDebugEnabled())
            logger.debug("processData() start acquiredData size : "   + acquiredData.size());

        while (itr.hasNext()) {
            acquiredObject = itr.next();
            String personKey = (String) ((Object[]) acquiredObject)[3];

            try {
                boolean shouldProcessAcquiredObject = shouldProcessAcquiredObject(context, acquiredObject);
    
                if (shouldProcessAcquiredObject) {
                    personKeys.add(personKey);
                    shouldProcessAquiredObjects.add(acquiredObject);
                } else {
                    continue;
                }
            } catch (Exception e) {
                
                handleFailure(context, acquiredObject, "Exception in shouldProcessAcquiredObject:", e);
                
            }


        }

        if (logger.isDebugEnabled())
            logger
                    .debug("processData() finished should process step with size :"   + shouldProcessAquiredObjects.size()   + " outof total :" + acquiredData.size());
        
        
        int batchId = 0;
        while (shouldProcessAquiredObjects.size() > 0) {

            // Get the first batch of the aquiredObjects.
            List currBatchOfAquiredObjects = new ArrayList();
            Set personKeysSet = new HashSet();
            if (shouldProcessAquiredObjects.size() > batchSize) {
                currBatchOfAquiredObjects.addAll(shouldProcessAquiredObjects.subList(0, batchSize));
            } else {
                // get all
                currBatchOfAquiredObjects.addAll(shouldProcessAquiredObjects);
            }
            
            shouldProcessAquiredObjects.removeAll(currBatchOfAquiredObjects);

            Iterator curBatchItr = currBatchOfAquiredObjects.iterator();

            while (curBatchItr.hasNext()) {
                acquiredObject = curBatchItr.next();
                String personKey = (String) ((Object[]) acquiredObject)[0];
                personKeysSet.add(personKey);
            }

            // now distribute this among 10 processes.

            int numPerProcess = personKeysSet.size() / numberofprocesses;

            // create processes.
            Set processes = new HashSet(numberofprocesses);
            Hashtable personsMap = new Hashtable();
            Hashtable personSites = new Hashtable();
            
            ArrayList personKeysArray = (new ArrayList());
            personKeysArray.addAll(personKeysSet);
            String perfId = "Start processing the batch : " + batchId;
            StopWatchLogger watch = new StopWatchLogger(perfId);
            watch.start();
            
            for (int i = 0; i < numberofprocesses; i++) {
                PersonLoaderProcess loadProcess = new PersonLoaderProcess();
                loadProcess.setPersonService(getPersonService());
                loadProcess.setPsDelegateService(getPsDelegateService());
                loadProcess.setPersons(personsMap);
                loadProcess.setPersonSites(personSites);
                if (numPerProcess > 0) {
                    int startPos = i * numPerProcess;
                    int endPos = startPos + numPerProcess;

                    // if this is the last process in the set then dump all the
                    // remaining keys here.
                    if (i == (numberofprocesses - 1))
                        endPos = (personKeysSet.size());
                    loadProcess.setPersonKeys(new HashSet(personKeysArray
                            .subList(startPos, endPos)));
                } else {
                    if (i < personKeysArray.size()) {
                        loadProcess.setPersonKeys(new HashSet(personKeysArray
                                .subList(i, i)));
                    }
                }

                loadProcess.setProcessId(i);
                
                loadProcess.setAuditId("BatchId :" + batchId + " ProcessId :" + i);
                processes.add(loadProcess);
            }

            // here we have the personObjects for all the persons in this batch
            getPersonLoader().setProcesses(processes);

            try {
                getPersonLoader().invoke(new ScheduledProcessInvocationContext());
            } catch (Exception e) {
                // just log and continue
                if (logger.isErrorEnabled())
                    logger.error("Exception in invoking the PersonLoaderProcess: "
                            , e);
            }

            // sleep until all the threads are complete.
            while (true) {
                try {
                    Thread.sleep(100);
                } catch (Exception e) {
                }

                boolean isComplete = true;
                Iterator processItr = processes.iterator();

                while (processItr.hasNext()) {
                    PersonLoaderProcess pesonLoader = (PersonLoaderProcess) processItr.next();

                    if (!pesonLoader.isComplete()) {
                        isComplete = false;
                        break;
                    }
                }

                if (isComplete) {
                    break;
                }
            }

            watch.stopAndLog(perfId);
            
            
            
            context.getContextData().put(KEY_BATCH_PERSON_HASH_TABLE, personsMap);
            context.getContextData().put(KEY_BATCH_PERSON_SITES_HASH_TABLE,personSites);
            batchId++;

            Iterator currBatchItr = currBatchOfAquiredObjects.iterator();

            //reset the personyearstats object to new 
            //PersonYearStats personYearStats = getPersonYearStats(context);
            //personYearStats = new PersonYearStats();
            resetPersonYearStats(context);

            
            while (currBatchItr.hasNext()) {
                acquiredObject = currBatchItr.next();
                try {
                   populatePersonStats(context, acquiredObject)  ;
                   processedObject = getProcessedObject(context,acquiredObject);

                    if (processedObject == null) {
                        if (logger.isWarnEnabled())
                            logger.warn("Unable to retrieve processed object for data sync...it was retrieved as null....ignoring this entry");
                        continue;
                    }
                    
                    // note that one entity could be transformed into multiple
                    // objects
                    transformedData = transformObject(context, processedObject,acquiredObject);
                    writeToFile(context, transformedData, data,processedObject, acquiredObject);
                } catch (Exception e) {
                    handleFailure(context, acquiredObject, "Exception while wrting to the File", e);
                    data.clear();
                }

                // inside this loop, update JobResult
                if (context.getProcessStatistics().isTotalNumberMod(DEFAULT_JOB_RESULT_UPDATE_INTERVAL))
                    updateJobResult(context);
            }

            // remove the list of objects processed from the parent list
            shouldProcessAquiredObjects.remove(currBatchOfAquiredObjects);
        }
    }

    private void populatePersonStats(DataQueryProcessExecutionContext context, Object acquiredObject) {
        PersonYearStats personYearStats = getPersonYearStats(context);
        PersonEntityKey personKey = CommonEntityKeyFactory
        .createPersonIdEntityKey(getPersonIdFromAcquiredObject(acquiredObject));
        
        Set years = getAllIncomeYearsFromAcquiredObject(acquiredObject);
        
        personYearStats.getYearsToProcess().clear();
        if (personYearStats.getPersonKey() == null
                || !personKey.getKeyValueAsString().equals(
                        personYearStats.getPersonKey()
                                .getKeyValueAsString())) {
            // no prev person exist, or different person
            personYearStats.setPersonKey(personKey);
            personYearStats.getProcessedYears().clear(); // clear prev
            // year years
            personYearStats.getYearsToProcess().addAll(years);
        } else {
            // exists, just add years
            for (Iterator iter = years.iterator(); iter.hasNext();) {
                Integer year = (Integer) iter.next();
                if (!personYearStats.getProcessedYears().contains(year)) {
                    personYearStats.getYearsToProcess().add(year);
                }
            }
        }
    }


    
    
    
    /**
     * @see gov.va.med.fw.batchprocess.AbstractDataQueriesProcess#prepareDataQueryDetail(gov.va.med.fw.batchprocess.DataQueryProcessExecutionContext, gov.va.med.fw.batchprocess.DataQueryDetail)
     */
    protected void prepareDataQueryDetail(DataQueryProcessExecutionContext context, DataQueryDetail query) {
        if (query.getQuery().getParamNames() != null) {
            Date lastExtractDate = IVMExtractDateProcessor.getLastExtractTime(context);
            Date extractStartTime = IVMExtractDateProcessor.getExtractStartTime(context);
            query.getQuery().setParamValues(new Object[] {lastExtractDate, extractStartTime});
        }
    }

    /**
     * @see gov.va.med.esr.common.batchprocess.datasync.AbstractDataSynchronizationProducerProcess#shouldProcessAcquiredObject(gov.va.med.fw.batchprocess.DataQueryProcessExecutionContext, java.lang.Object)
     */
    protected boolean shouldProcessAcquiredObject(DataQueryProcessExecutionContext context, Object acquiredObject) {
        PersonEntityKey personKey = CommonEntityKeyFactory.createPersonIdEntityKey(getPersonIdFromAcquiredObject(acquiredObject));
        boolean process = false;
        PersonYearStats personYearStats = getPersonYearStats(context);
        if (isRemigrate(acquiredObject)) {
            process = true;
        }
        else {
            Integer incomeYear = (Integer)getIncomeYearFromAcquiredObject(acquiredObject);
            try {
                VPIDEntityKey vpidKey = CommonEntityKeyFactory.createVPIDEntityKey(getVpidValueFromAcquiredObject(acquiredObject));
                
                Set incomeYears = ivmDMService.findExtractIncomeYears(personKey, incomeYear, vpidKey);
                if (incomeYears.size()== 0)
                    return false;
                
                if (incomeYear == null) {
                    ((Object[])acquiredObject)[1] = incomeYears;
                }
                process = true;
            } catch (ServiceException ex) {
                IllegalStateException newEx = new IllegalStateException("Error while checking the ivm selection criteria.");
                newEx.initCause(ex);
                throw newEx;
            }
        }
        if (process) {
            Set years = getAllIncomeYearsFromAcquiredObject(acquiredObject);
            personYearStats.getYearsToProcess().clear();
            if (personYearStats.getPersonKey() == null || !personKey.getKeyValueAsString().equals(personYearStats.getPersonKey().getKeyValueAsString())) {
                //no prev person exist, or different person
                personYearStats.setPersonKey(personKey);
                personYearStats.getProcessedYears().clear(); //clear prev year years
                personYearStats.getYearsToProcess().addAll(years);
                personYearStats.getProcessedYears().addAll(years);
            }
            else {
                //exists, just add years
                for (Iterator iter = years.iterator(); iter.hasNext();) {
                    Integer year = (Integer) iter.next();
                    if (!personYearStats.getProcessedYears().contains(year)) {
                        personYearStats.getYearsToProcess().add(year);
                        personYearStats.getProcessedYears().add(year);
                    }
                }
            }
            return personYearStats.getYearsToProcess().size() > 0;
        }
        return process;
    }
    
    /**
     * @see gov.va.med.esr.common.batchprocess.datasync.AbstractDataSynchronizationProducerProcess#getProcessedObject(gov.va.med.fw.batchprocess.DataQueryProcessExecutionContext, java.lang.Object)
     */
    protected Object getProcessedObject(DataQueryProcessExecutionContext context, Object acquiredObject) throws ServiceException {
        // retrieve Person
        PersonEntityKey personKey = CommonEntityKeyFactory.createPersonIdEntityKey(getPersonIdFromAcquiredObject(acquiredObject));
        if(logger.isDebugEnabled())
            logger.debug("Synchronizing data for Person: " + personKey.getKeyValueAsString());
        
        return getPersonYearStats(context).getPerson((Hashtable) context.getContextData().get(
                KEY_BATCH_PERSON_HASH_TABLE));
    }
    
    /**
     * @see gov.va.med.esr.common.batchprocess.AbstractDataSynchronizationProducer#transformPerson(java.lang.Object, java.lang.Object)
     */
    protected Object transformObject(DataQueryProcessExecutionContext context, Object processedObject, Object acquiredObject) throws ServiceException {
        //Person person = (Person) processedObject; 
        PersonYearStats personYearStats = getPersonYearStats(context);
        Person person = personYearStats.getPerson((Hashtable) context.getContextData().get(
                KEY_BATCH_PERSON_HASH_TABLE));
        Set incomeYears = personYearStats.getYearsToProcess();
        List data = new ArrayList();
        for (Iterator iter = incomeYears.iterator(); iter.hasNext();) {
            Integer year = (Integer) iter.next();
            if (!personYearStats.getProcessedYears().contains(year)) {
                Set sites = getSitesFromContext(context, person.getVPIDEntityKey());
                data.addAll(ivmDataSyncProducer.getIVMRawData(person, year, sites));
                personYearStats.getProcessedYears().add(year);
            }
        }
        return data;
    }

    /**
     * retives the sites from the cache that was stored in the context as hahtable. if this is npt found then 
     * go to the database.
     * @param context
     * @param vpEntityKey
     * @return
     * @throws ServiceException
     */
    private Set getSitesFromContext(DataQueryProcessExecutionContext context,VPIDEntityKey vpEntityKey ) throws ServiceException
    {
        Hashtable personSitesHashtable =       (Hashtable)context.getContextData().get(KEY_BATCH_PERSON_SITES_HASH_TABLE);
        Set sites = null;
        if (personSitesHashtable != null  )
        {
            sites = (Set)personSitesHashtable.get(vpEntityKey); 
        }
        
        if ( sites == null )
        {
            sites = psDelegateService.getSites(vpEntityKey);
        }
        return sites;
    }

    /**
     * @see gov.va.med.esr.common.batchprocess.datasync.AbstractDataSynchronizationProducerProcess#postProcessDataFile(gov.va.med.fw.batchprocess.DataQueryProcessExecutionContext, gov.va.med.esr.common.batchprocess.datasync.DataSynchronizationProducerStatistics)
     */
    protected void postProcessDataFile(DataQueryProcessExecutionContext context, DataSynchronizationProducerStatistics stats) {
    	//Moved this conversion from end of processData method to here to fix CR 8727
    	changeFileExtensions("TMP", "TXT");
    	List fileData = stats.getFileData();
        if (fileData == null || fileData.size() == 0) {
            return;
        }
        IVMDMExtractBatch ivmBatch = new IVMDMExtractBatch();
        ivmBatch.setMigrationType(getIVMDMType());
        ivmBatch.setExtractFileName(stats.getFileName());
        
        //fix CCR 8013 setting the file creation date as job start date. To be more accurate it will be the 
        //value given to extractStartTime in the query as this will be the starting point for the next time.
        //ivmBatch.setExtractStartDate(stats.getStartTime());
        ivmBatch.setExtractStartDate(IVMExtractDateProcessor.getExtractStartTime(context));
        ivmBatch.setExtractEndDate(stats.getEndTime());
        ivmBatch.setExtractRecordCount(new Integer(stats.getEntityCount()));
        
        for (Iterator iter = fileData.iterator(); iter.hasNext();) {
            IVMDMExtractRecord extractRecord = (IVMDMExtractRecord) iter.next();
            ivmBatch.addExtractRecord(extractRecord);
        }
        try {
            ivmDMService.saveIVMDMExtractBatch(ivmBatch);
        } catch (ServiceException ex) {
            throw new IllegalArgumentException("Error saving the IVM Extract statistics. Reason: "+ex.getMessage());
        }
    }
    
    /**
     * @see gov.va.med.esr.common.batchprocess.datasync.AbstractDataSynchronizationProducerProcess#postProcessAcquiredObject(gov.va.med.fw.batchprocess.DataQueryProcessExecutionContext, gov.va.med.esr.common.batchprocess.datasync.DataSynchronizationProducerStatistics, java.lang.Object, java.lang.Object)
     */
    protected void postProcessAcquiredObject(DataQueryProcessExecutionContext context, DataSynchronizationProducerStatistics stats, Object processedObject, Object acquiredObject) {
        Person person = (Person) processedObject;
        Set incomeYears = getPersonYearStats(context).getYearsToProcess();
        
        for (Iterator iter = incomeYears.iterator(); iter.hasNext();) {
            Integer year = (Integer) iter.next();
            Date extractStartTime = IVMExtractDateProcessor.getExtractStartTime(context);
            String icn = person.getVPIDEntityKey().getShortVPID();
            PersonEntityKey personKey = person.getPersonEntityKey();
            try {
                ivmDMService.saveIVMMigration(extractStartTime, personKey, icn, year);
            } catch (ServiceException ex) {
                throw new IllegalArgumentException("Error saving the IVM Extract. Reason: "+ex.getMessage());
            }
        }
    }

    protected void addFileData(DataQueryProcessExecutionContext context, DataSynchronizationProducerStatistics stats, Object processedObject, Object acquiredObject) {
        Person person = (Person) processedObject;
        Set incomeYears = getPersonYearStats(context).getYearsToProcess();

        String icn =  person.getVPIDEntityKey().getShortVPID();
        for (Iterator iter = incomeYears.iterator(); iter.hasNext();) {
            Integer year = (Integer) iter.next();

            IVMDMExtractRecord ivmRecord = new IVMDMExtractRecord();
            ivmRecord.setIcn(icn);
            ivmRecord.setIncomeYear(year);
            
            stats.addFileData(ivmRecord);
        }
    }

    
    
    private void handleFailure(DataQueryProcessExecutionContext context,
            Object acquiredObject, String errorMessage, Exception e) {
        
        try
        {
            String exceptionText = null;
            logger.error("handleFailure ", e);
            StringBuffer identityData = new StringBuffer();;
            identityData.append("Current TimeStamp :").append(new Date());
            identityData.append("\tPersonId :").append(getPersonIdFromAcquiredObject(acquiredObject));
            identityData.append("\tVpidValue :").append(getVpidValueFromAcquiredObject(acquiredObject));
            

            try 
            {
                Object incomeYearObj = getIncomeYearFromAcquiredObject(acquiredObject);
                Integer incomeYear = null;
                if (incomeYearObj instanceof Integer)
                {
                    incomeYear = (Integer)incomeYearObj;
                }
                identityData.append("\tIncome Year :").append(incomeYear);
            }
            catch(Exception debugEx)
            {
                logger.error("is This class Cast Exception ", debugEx);
            }
            
            if (logger.isErrorEnabled())
            {
                logger.error(exceptionText);
                logger.error("Unable to synchronize Entity: "
                        + identityData.toString() + " due to exception", e);
            }
            
            context.getProcessStatistics().incrementNumberOfErrorRecords();
            context.getExceptionData().add(identityData.toString());
            
           if ( e != null )
           {
               StringWriter sw = new StringWriter();
               e.printStackTrace(new PrintWriter(sw));
               String trackTrace =  sw.toString();
               context.getExceptionData().add(trackTrace);
           }
    
            
            
            if ( shouldWriteExceptionData(context))
            {
                ((IVMDataSynchronizationProducerProcessCompletedHandler)getDataProcessCompletedHandler()).appendExceptionData(context);
            }
        }
        catch(Exception eX)
        {
            //eating the exception
        }
    }

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

    private PersonYearStats getPersonYearStats(DataQueryProcessExecutionContext context) {
        PersonYearStats personStats = (PersonYearStats)context.getContextData().get(KEY_LAST_EXTRACTED_PERSON);
        if (personStats == null) {
            personStats = new PersonYearStats();
            context.getContextData().put(KEY_LAST_EXTRACTED_PERSON, personStats);
        }
        return personStats;
    }

    private String getPersonIdFromAcquiredObject(Object acquiredObject) {
        return (String)((Object[])acquiredObject)[0];
    }
    private Object getIncomeYearFromAcquiredObject(Object acquiredObject) {
        return ((Object[])acquiredObject)[1];
    }
    private Set getAllIncomeYearsFromAcquiredObject(Object acquiredObject) {
        Set years = new HashSet();
        Object obj = ((Object[])acquiredObject)[1];
        if (obj instanceof Set) {
            years.addAll(((Set)obj));
        }
        else {
            years.add(obj);
        }
        return years;
    }
    private boolean isRemigrate(Object acquiredObject) {
        return Boolean.TRUE.equals((Boolean)((Object[])acquiredObject)[2]);
    }
    private String getVpidValueFromAcquiredObject(Object acquiredObject) {
        return (String)((Object[])acquiredObject)[3];
    }    
   
    /**
     * @return Returns the psDelegateService.
     */
    public PSDelegateService getPsDelegateService() {
        return psDelegateService;
    }
    
    /**
     * @param psDelegateService The psDelegateService to set.
     */
    public void setPsDelegateService(PSDelegateService psDelegateService) {
        this.psDelegateService = psDelegateService;
    }
    
    /**
     * @return Returns the systemParameterService.
     */
    public SystemParameterService getSystemParameterService() {
        return systemParameterService;
    }
    
    /**
     * @param systemParameterService The systemParameterService to set.
     */
    public void setSystemParameterService(SystemParameterService systemParameterService) {
        this.systemParameterService = systemParameterService;
    }
    
    /**
     * @return Returns the lookupService.
     */
    public LookupService getLookupService() {
        return lookupService;
    }

    /**
     * @param lookupService The lookupService to set.
     */
    public void setLookupService(LookupService lookupService) {
        this.lookupService = lookupService;
    }
    
    /**
     * @return Returns the ivmDataSyncProducer.
     */
    public IVMDataSynchronizationProducer getIvmDataSyncProducer() {
        return ivmDataSyncProducer;
    }
    
    /**
     * @param ivmDataSyncProducer The ivmDataSyncProducer to set.
     */
    public void setIvmDataSyncProducer(IVMDataSynchronizationProducer ivmDataSyncProducer) {
        this.ivmDataSyncProducer = ivmDataSyncProducer;
    }
    
    /**
     * @return Returns the ivmDMService.
     */
    public IVMDMService getIvmDMService() {
        return ivmDMService;
    }

    /**
     * @param ivmDMService The ivmDMService to set.
     */
    public void setIvmDMService(IVMDMService ivmDMService) {
        this.ivmDMService = ivmDMService;
    }

    public int getBatchSize() {
        return batchSize;
    }

    public void setBatchSize(int batchSize) {
        this.batchSize = batchSize;
    }

    public int getNumberofprocesses() {
        return numberofprocesses;
    }

    public void setNumberofprocesses(int numberofprocesses) {
        this.numberofprocesses = numberofprocesses;
    }

    public ConcurrentProcess getPersonLoader() {
        return personLoader;
    }

    public void setPersonLoader(ConcurrentProcess personLoader) {
        this.personLoader = personLoader;
    }

    /**
     * @see gov.va.med.esr.common.batchprocess.AbstractDataSynchronizationProducer#afterPropertiesSet()
     */
    public void afterPropertiesSet() {
        super.afterPropertiesSet();
        Validate.notNull(psDelegateService, "PSDelegateService is required.");
        Validate.notNull(lookupService, "LookupService is required.");
        Validate.notNull(systemParameterService, "SystemParameterService is required.");
        Validate.notNull(ivmDMService, "IVMDMService is required.");
        Validate.notNull(ivmDataSyncProducer, "IVMDataSynchronizationProducer is required.");
        Validate.notNull(personLoader, "personLoader is required");
        setDefaultsConcurrentProps();
        //Fix for CR_8739, moving below method call into the executeQuery method.
        //setEntityCountPerFile(getIvmMaxRecordsPerFile());

    }

    private void setDefaultsConcurrentProps()
    {
        if ( this.batchSize <= 0 )
        {
            setBatchSize(DEFAULT_BATCH_SIZE);
        }
        
        if ( this.numberofprocesses <= 0 )
        {
            setNumberofprocesses(DEFAULT_NUM_PROCESSES);
        }
    }

    private IVMDMType getIVMDMType() {
        if (ivmDmType == null) {
            try {
                ivmDmType =  (IVMDMType)lookupService.getByCode(IVMDMType.class, IVMDMType.ONGOING.getCode());
            } catch (Exception ex) {
                throw new IllegalStateException("Missing IVM DM type "+IVMDMType.ONGOING.getCode()+" Reason: "+ex.getMessage());
            }
        }
        return ivmDmType;
    }

    private int getIvmMaxRecordsPerFile() {
        int ivmRecordsPerFile = 0; //default
        try {
            SystemParameter ivmMaxRecords = systemParameterService.getByName(SystemParameter.IVM_DM_EXPORTER_MAX_RECORDS);
            if (ivmMaxRecords != null && ivmMaxRecords.getValue() != null)
                ivmRecordsPerFile = Integer.parseInt(ivmMaxRecords.getValue());
        } catch (Exception ex) {
            ex.printStackTrace();
        }
        return ivmRecordsPerFile;
    }
    
    private void resetPersonYearStats(DataQueryProcessExecutionContext context) {
        HashMap contextData = (HashMap)context.getContextData();
        contextData.put(KEY_LAST_EXTRACTED_PERSON, null);
    }
    
    /**
     * A inner class to hold single person / years statistics during the process.
     * @date Apr 27, 2006 10:44:54 PM
     * @author DNS   MANSOG
     */
    private class PersonYearStats {
        private PersonEntityKey personKey = null;
        private Person person = null;
        private Set processedYears = new HashSet();
        private Set yearsToProcess = new HashSet();
        
        /**
         * @return Returns the personKey.
         */
        public PersonEntityKey getPersonKey() {
            return personKey;
        }
        
        /**
         * @param personKey The personKey to set.
         */
        public void setPersonKey(PersonEntityKey personKey) {
            this.personKey = personKey;
            if (person != null && !person.getEntityKey().getKeyValueAsString().equals(personKey.getKeyValueAsString())) {
                person = null;
            }
        }


        public Person getPerson(Hashtable personsMap) throws ServiceException {
            if (person == null) {
                if (personsMap != null) {
                    person = (Person) personsMap.get(personKey
                            .getKeyValueAsString());
                }
                
                //If the person is not cached yet
                if (person == null)
                {
                    person = getPersonService().getPerson(personKey);
                }
            }
            return person;
        }

        /**
         * @return Returns the years.
         */
        public Set getProcessedYears() {
            return processedYears;
        }
        
        /**
         * @return Returns the toProcess.
         */
        public Set getYearsToProcess() {
            return yearsToProcess;
        }
    }
}
