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

package gov.va.med.fw.batchprocess;

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

import gov.va.med.fw.batchprocess.model.JobConfig;
import gov.va.med.fw.batchprocess.model.JobResult;
import gov.va.med.fw.batchprocess.model.JobStatus;
import gov.va.med.fw.model.EntityKey;
import gov.va.med.fw.model.EntityKeyFactory;
import gov.va.med.fw.service.AbstractComponent;

/**
 * Worker implementation for AbstractDataProcess instances that need parallelism
 * (eg, processing JMS messages across a cluster).
 * 
 * Created Sep 21, 2006 3:13:12 PM
 * 
 * @author DNS   BOHMEG
 */
public abstract class AbstractDataProcessWorker extends AbstractComponent {
	private ProcessStatisticsHandler processStatisticsHandler;
	private BatchProcessService batchProcessService;
	
	public final void process(DataProcessWorkerInput in) {
		if(logger.isInfoEnabled())
			logger.info("Processing data for process: " + in.getProcessName() + " with a payload count of: " + in.getPayloadCount() +
					" and a total count of: " + in.getTotalCount());
		try {
			EntityKey jobResultKey = null;
			if(in.getId() != null)
				jobResultKey = EntityKeyFactory.createEntityKey(in.getId().toString(), JobResult.class);
			// 1) verify if should still process data
			JobResult jobResult = batchProcessService.getJobResult(jobResultKey);
			if (shouldProcessData(in, jobResult)) {
				// 2) process data
				ProcessStatistics workerStats = processData(in.getPayload());
				workerStats.setProcessName(in.getProcessName());
				JobResultData data = batchProcessService.saveAndOverlayJobResult(jobResultKey, workerStats, in.getTotalCount());
				processStatistics(data);
			}
		} catch (Exception e) {
			if (logger.isErrorEnabled())
				logger.error("Failed processing data", e);
		}
	}
	
	protected final void processStatistics(JobResultData data) throws Exception {
		ProcessStatistics jobStats = data.getProcessStatistics();
		JobResult jobResult = data.getJobResult();
		
		if(JobStatus.CANCELLED.equals(jobResult.getStatus()))
			jobStats.setWasInterrupted(true);
	
		if(!JobStatus.IN_PROCESS.equals(jobResult.getStatus())) {			
			if(logger.isInfoEnabled())
				logger.info("ProcessStatistics [" + jobStats + "]");			
			if(processStatisticsHandler != null) {
				JobConfig config = this.batchProcessService.getJobConfig(jobResult.getJobName(), jobResult.getJobGroup());
				if(config != null && StringUtils.isNotBlank(config.getEmailDistributionList())) {
					processStatisticsHandler.processStatisticsWithDynamicRecipients(jobStats, config.getEmailsAsList());
				} else {
					processStatisticsHandler.processStatistics(jobStats);
				}
			}
			
		}
	}	

	protected boolean shouldProcessData(DataProcessWorkerInput in,
			JobResult currentJobResult) {
		return JobStatus.IN_PROCESS.getCode().equals(
				currentJobResult.getStatus().getCode());
	}

	// TODO: change this to be protected
	// for now, keeping this public allows for easy testing of worker
	public abstract ProcessStatistics processData(Object o) throws Exception;

	/**
	 * @return Returns the processStatisticsHandler.
	 */
	public ProcessStatisticsHandler getProcessStatisticsHandler() {
		return processStatisticsHandler;
	}

	/**
	 * @param processStatisticsHandler The processStatisticsHandler to set.
	 */
	public void setProcessStatisticsHandler(
			ProcessStatisticsHandler processStatisticsHandler) {
		this.processStatisticsHandler = processStatisticsHandler;
	}

	/**
	 * @return Returns the batchProcessService.
	 */
	public BatchProcessService getBatchProcessService() {
		return batchProcessService;
	}

	/**
	 * @param batchProcessService The batchProcessService to set.
	 */
	public void setBatchProcessService(BatchProcessService batchProcessService) {
		this.batchProcessService = batchProcessService;
	}
	
	public void afterPropertiesSet() {
		Validate.notNull(this.batchProcessService, "batchProcessService is required");
	}
}
