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


package gov.va.med.fw.batchprocess;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

import org.apache.commons.lang.Validate;

import gov.va.med.fw.batchprocess.model.JobResult;
import gov.va.med.fw.model.EntityKey;
import gov.va.med.fw.service.trigger.TriggerEvent;
import gov.va.med.fw.service.trigger.TriggerRouter;
import gov.va.med.fw.util.InvalidConfigurationException;

/**
 * Specialization of AbstractDataQueryIncrementalProcess that will publish query results to JMS
 * queues for instances of AbstractDataProcessWorker to asynchronously process.
 * 
 * <p>Useful for "large" queries that need parallel processing.
 * 
 * <p>If a worker is injected, it will serially process data using that worker.  Useful for
 * configuration changes to determine performance impact.
 * 
 * Created Sep 25, 2006 10:32:31 AM
 * @author DNS   BOHMEG
 */
public abstract class AbstractPublishQueryResultsToWorkers extends
		AbstractDataQueryIncrementalProcess {
	protected static final String CURRENT_RECORDS_PROCESSED = "totalRecordsProcessed";
	protected static final String MAX_RECORDS_QUERIED = "maxRecordsQueries";
	protected static final String TOTAL_QUERY_COUNT = "totalQueryCount";	
	private String totalCountQueryName;
	private TriggerRouter triggerRouter;
	private int messageSize;
	/** allows for synchronous/serial usage of the workers, does not use the trigger router */
	private AbstractDataProcessWorker worker;
	
	protected boolean shouldUpdateJobResult(DataQueryProcessExecutionContext context) {
		return worker != null;		
	}
	
	protected boolean ownsJobCompletion(DataProcessExecutionContext context) {		
		boolean val = true;
		try {
			val = (worker == null && getTotalCount((DataQueryProcessExecutionContext) context) != 0) ? false : true;			
		} catch(Exception e) {
			this.throwIllegalStateException("Unable to determine if ownsJobCompletion", e);
		}
		return val;		
	}	

	private List checkAgainstMax(DataQueryProcessExecutionContext context, List acquiredData) throws Exception {
		int realizedTotalCount = this.getTotalCount(context);
		if(context.getContextData().containsKey(MAX_RECORDS_QUERIED)) {
			int maxTotalCount = ((Integer) context.getContextData().get(MAX_RECORDS_QUERIED)).intValue();
			if(maxTotalCount < realizedTotalCount) {
				realizedTotalCount = maxTotalCount; 
				context.getContextData().put(TOTAL_QUERY_COUNT, new Integer(maxTotalCount));
			}
		}
		
		int currentRecordsQueried = 0;
		if(context.getContextData().containsKey(CURRENT_RECORDS_PROCESSED)) {
			currentRecordsQueried = ((Integer) context.getContextData().get(CURRENT_RECORDS_PROCESSED)).intValue();		
		}
		
		if(currentRecordsQueried >= realizedTotalCount) {
			return null; // already exceeded, no more data for you
		}
		
		if((currentRecordsQueried + acquiredData.size()) > realizedTotalCount) {
			int totalToRemove = (currentRecordsQueried + acquiredData.size()) - realizedTotalCount;
			int lastIndex = acquiredData.size() - 1;
			for(int i=lastIndex; i > (lastIndex - totalToRemove); i--)
				acquiredData.remove(i);
		}

		return acquiredData;
	}
	

	/* Will be called for each increment of "large" query.
	 * 
	 * @see gov.va.med.fw.batchprocess.AbstractDataQueryIncrementalProcess#processData(gov.va.med.fw.batchprocess.DataQueryProcessExecutionContext, java.util.List)
	 */
	protected void processData(DataQueryProcessExecutionContext context,
			List acquiredData) {			
		try {
			// re-adjust to abide by any custom total desired
			acquiredData = checkAgainstMax(context, acquiredData);
			if(acquiredData == null)
				return;
			
			// keep track of number of records visited
			Integer currentRecordsProcessed = (Integer) context.getContextData().get(CURRENT_RECORDS_PROCESSED);
			if(currentRecordsProcessed == null)
				currentRecordsProcessed = new Integer(acquiredData.size());
			else
				currentRecordsProcessed = new Integer(currentRecordsProcessed.intValue() + acquiredData.size());
			context.getContextData().put(CURRENT_RECORDS_PROCESSED, currentRecordsProcessed);
						
			// ok to proceed and process
			if(worker == null) {
				// no worker, publish JMS messages to async workers
				JobResult jobResult = getJobResult(context);
				DataProcessWorkerInput wrappedPayload = new DataProcessWorkerInput();
				wrappedPayload.setProcessName(getProcessName());
				EntityKey key = jobResult.getEntityKey();
				if(key != null)
					wrappedPayload.setId(key.getKeyValueAsString());
				wrappedPayload.setContext(jobResult.getContext());
				wrappedPayload.setTotalCount(getTotalCount(context));
				
				// create events with payloads equal to messageSize
				if(messageSize == getFetchSize()) {
					// create a single JMS Message for each fetch
					wrappedPayload.setPayloadCount(acquiredData.size());
					TriggerEvent event = getTriggerEvent(context, acquiredData);
					wrappedPayload.setPayload(event.getPayload());
					event.setPayload(wrappedPayload);
					triggerRouter.processTriggerEvent(event);					
				} else {
					// iterate through acquiredData and send smaller messages
					Iterator itr = acquiredData.iterator();
					List messageData = new ArrayList();
					while(itr.hasNext()) {
						messageData.add(itr.next());
						if(messageData.size() == messageSize) {
							wrappedPayload.setPayloadCount(messageData.size());
							TriggerEvent event = getTriggerEvent(context, messageData);
							wrappedPayload.setPayload(event.getPayload());
							event.setPayload(wrappedPayload);
							triggerRouter.processTriggerEvent(event);					
							
							messageData.clear();
						}
					}
					// check for last batch (partial)
					if(!messageData.isEmpty()) {
						wrappedPayload.setPayloadCount(messageData.size());
						TriggerEvent event = getTriggerEvent(context, messageData);
						wrappedPayload.setPayload(event.getPayload());
						event.setPayload(wrappedPayload);
						triggerRouter.processTriggerEvent(event);											
					}
				}
			} else {
				// get event to get the "formatted" payload
				TriggerEvent event = getTriggerEvent(context, acquiredData);
				
				// yes worker, worker will process serially internally				
				ProcessStatistics workerStats = worker.processData(event.getPayload());
				workerStats.setProcessName(getProcessName());
				context.getProcessStatistics().overlayStats(workerStats);
			}
		} catch(Exception e) {
			super.throwIllegalStateException("Failed to process", e);
		}						
	}
	
	private final int getTotalCount(DataQueryProcessExecutionContext context) throws Exception {
		Number count = null;
		synchronized(context) {
			count = (Number) context.getContextData().get(TOTAL_QUERY_COUNT);
			if(count == null) {
				count = getTotalCount();
				context.getContextData().put(TOTAL_QUERY_COUNT, count);
			}
		}
		return count.intValue();
	}
	
	private Number getTotalCount() throws Exception {
		return (Number) this.getDao().find(totalCountQueryName).get(0);
	}
	
	protected abstract TriggerEvent getTriggerEvent(DataQueryProcessExecutionContext context,
			List acquiredData);

	/**
	 * @return Returns the triggerRouter.
	 */
	public TriggerRouter getTriggerRouter() {
		return triggerRouter;
	}

	/**
	 * @param triggerRouter The triggerRouter to set.
	 */
	public void setTriggerRouter(TriggerRouter triggerRouter) {
		this.triggerRouter = triggerRouter;
	}
	
	public void afterPropertiesSet() {
		super.afterPropertiesSet();
		Validate.notNull(triggerRouter, "triggerRouter is required");
		Validate.notNull(totalCountQueryName, "totalCountQueryName is required");
		if(messageSize <= 0)
			messageSize = getFetchSize();
		if(messageSize > getFetchSize())
			throw new InvalidConfigurationException("JMS Message size can not be larger than query fetch size");
		if((getFetchSize() % messageSize) != 0)
			throw new InvalidConfigurationException("JMS Message must be fetchSize mod 0");
	}

	/**
	 * @return Returns the worker.
	 */
	public AbstractDataProcessWorker getWorker() {
		return worker;
	}

	/**
	 * @param worker The worker to set.
	 */
	public void setWorker(AbstractDataProcessWorker worker) {
		this.worker = worker;
	}

	/**
	 * @return Returns the totalCountQueryName.
	 */
	public String getTotalCountQueryName() {
		return totalCountQueryName;
	}

	/**
	 * @param totalCountQueryName The totalCountQueryName to set.
	 */
	public void setTotalCountQueryName(String totalCountQueryName) {
		this.totalCountQueryName = totalCountQueryName;
	}

	/**
	 * @return Returns the messageSize.
	 */
	public int getMessageSize() {
		return messageSize;
	}

	/**
	 * @param messageSize The messageSize to set.
	 */
	public void setMessageSize(int messageSize) {
		this.messageSize = messageSize;
	}
}
