package org.kuali.ole.sys.batch;
   
   
   
   import java.io.File;
   import java.util.ArrayList;
   import java.util.Date;
   import java.util.HashMap;
   import java.util.Iterator;
  import java.util.List;
  import java.util.Map;
  import java.util.concurrent.Executor;
  import java.util.concurrent.Executors;
  
  import org.kuali.ole.sys.OLEConstants;
  import org.kuali.ole.sys.context.BatchContainerDirectory;
  import org.kuali.ole.sys.context.BatchLogger;
  import org.kuali.ole.sys.context.BatchStepExecutor;
  import org.kuali.ole.sys.context.BatchStepFileDescriptor;
  import org.kuali.ole.sys.context.ContainerStepListener;
  import org.kuali.rice.core.api.datetime.DateTimeService;
  import org.kuali.rice.coreservice.framework.parameter.ParameterService;
  
  /**
   * BatchContainerStep looks for .run files.
   * When one is found it deletes the file and creates a new thread (BatchStepExecutor) which executes the Step indicated by the .run file.
   * BatchContainerStep continues looking for .run files until it finds a stopBatchContainerStep.run file.
   * When BatchContainerStep begins it writes a .runlock file to indicate that the batch container is running, but will first look for an existing .runlock file
   * and if one is found it will exit immediately.
   *
   * BatchContainerStep adds a ConsoleAppender to its Logger if one hasn't been configured.
   *
   */
  public class BatchContainerStep extends AbstractStep implements ContainerStepListener {
      static org.apache.log4j.Logger LOG = org.apache.log4j.Logger.getLogger(BatchContainerStep.class);
  
      protected String batchContainerDirectory;
      protected Step batchContainerStopStep;
  
      protected BatchContainerDirectory directory;
  
      protected StringBuffer containerResults;
      protected Map<String, BatchStepFileDescriptor> startedSteps;
      protected List<BatchStepFileDescriptor[]> completedSteps;
  
      /**
       * This method begins an infinite loop in order to process semaphore files written by BatchStepTrigger (called via the brte scripts).
       * BatchStepTrigger writes .run files, BatchContainerStep reads .run files and calls BatchStepExecutor (its own thread) which will execute the
       * Step and write either a .success or .error result file which is then read by BatchStepTrigger.
       *
       * This method exits gracefully when it receives a .run file for the batchContainerStopStep.
       *
       */
  	@Override
      public boolean execute(String jobName, Date jobRunDate) throws InterruptedException {
  	    BatchLogger.addConsoleAppender(LOG);
  
  		LOG.info("Starting the batch container in Job: "+ jobName +" on "+ jobRunDate);
  
  		if (batchContainerDirectory == null) {
  			throw new RuntimeException("The batchContainerDirectory has not been specified.");
  		}
  		if (batchContainerStopStep == null) {
  			throw new RuntimeException("The batchContainerStopStep has not been specified.");
  		}
  
  		directory = new BatchContainerDirectory(batchContainerDirectory);
  
  		if (directory.isBatchContainerRunning()) {
  			//an instance of the batch container is already running - exit w/out trying to remove the batch container semaphore file
  			LOG.error("The BatchContainer is already running");
  			throw new RuntimeException("The BatchContainer is already running.");
  		}
  
  		initContainerResults();
  
  		try {
  			//write batch container run lock file to indicate the batch container is running
  			directory.writeBatchContainerSemaphore(jobName, getName());
  			directory.addShutdownHook();
  			LOG.info("The BatchContainer is running");
  
  	        ParameterService parameterService = getParameterService();
  	        DateTimeService dateTimeService = getDateTimeService();
  
  			Executor executor = Executors.newCachedThreadPool();
  	        while(true) {
  
  	            if (LOG.isDebugEnabled()) {
  	                LOG.debug("Looking for steps...");
  	            }
  	        	File[] stepRunFiles = directory.getStepRunFiles();
  
  	        	while (stepRunFiles != null && stepRunFiles.length > 0) {
  	        		LOG.info("Found "+ stepRunFiles.length +" steps to execute");
  
  	        		for(File stepRunFile : stepRunFiles) {
  	        			BatchStepFileDescriptor batchStepFile = new BatchStepFileDescriptor(stepRunFile);
  
 	        			Step step = getStep(batchStepFile);
 	        			if (step == null) {
 	        				directory.removeBatchStepFileFromSystem(batchStepFile);
 	        				directory.writeBatchStepErrorResultFile(batchStepFile, new IllegalArgumentException("Unable to find bean for step: "+ batchStepFile.getStepName()));
 	        			}
 	        			else {
 
 	        				if (isStopBatchContainerTriggered(step)) {
 	        					directory.removeBatchStepFileFromSystem(batchStepFile);
 	        					directory.writeBatchStepSuccessfulResultFile(batchStepFile);
 
 	        					//Stop BatchContainer
 	        					LOG.info("shutting down container");
 	        					return true;
 	        				}
 
 	        				//retrieve the stepIndex before the file is removed
 	        				int stepIndex = directory.getStepIndexFromFile(batchStepFile);
 
 	        				directory.removeBatchStepFileFromSystem(batchStepFile);
 
 	        				if (LOG.isDebugEnabled()) {
 	        				    LOG.debug("Creating new thread to run "+ batchStepFile);
 	        				}
 	        				BatchStepExecutor batchStepExecutor = new BatchStepExecutor(parameterService, dateTimeService, directory, batchStepFile, step, stepIndex);
                             batchStepExecutor.addContainerStepListener(this);
 	        				executor.execute(batchStepExecutor);
 
 	        			}
 	        		}
 
 	        		if (LOG.isDebugEnabled()) {
 	        		    LOG.debug("Looking for steps...");
 	        		}
 	            	stepRunFiles = directory.getStepRunFiles();
 
 	        	}
 
 	        	sleep();
 	    		if (!directory.isBatchContainerRunning()) {
 	    			//the batch container's runlock file no longer exists - exit
 	    			LOG.error("The BatchContainer runlock file no longer exists - exiting");
 	    			return false;
 	    		}
 	        }
 
 		} finally {
 			//remove batch container run lock file
 			directory.removeBatchContainerSemaphore();
 			LOG.info("The BatchContainer has stopped running");
 
 			logContainerResultsSummary();
 		}
 	}
 
 	/**
 	 * Notification that the Step started. Log the Step's information
 	 */
 	@Override
     public void stepStarted(BatchStepFileDescriptor runFile, String logFileName) {
 	    logStepStarted(runFile, logFileName);
 	}
 
 	/**
 	 * Notification that the Step finished. Log the Step's information
 	 */
 	@Override
     public void stepFinished(BatchStepFileDescriptor resultFile, String logFileName) {
 	    logStepFinished(resultFile, logFileName);
 	}
 
 	/**
 	 * Retrieves the Step bean from the SpringContext
 	 *
 	 * @param batchStepFile the file descriptor for the step to run
 	 * @return the Step bean from the SpringContext
 	 */
 	protected Step getStep(BatchStepFileDescriptor batchStepFile) {
 	  if (LOG.isDebugEnabled()) {
 	      LOG.debug("Converting step named in .run file into a Step class...");
 	  }
 
 	  Step step = null;
 	  try {
 	      step = BatchSpringContext.getStep(batchStepFile.getStepName());
 	  } catch (RuntimeException runtimeException) {
 	      LOG.error("Failed to getStep from spring context: ", runtimeException);
 	  }
       if (step == null) {
       	LOG.error("Unable to find bean for step: "+ batchStepFile.getStepName());
       	return null;
       }
 
       LOG.info("Found valid step: "+ step.getName());
       return step;
 	}
 
 	/**
 	 * @param step the Step specified by the .run file
 	 * @return true if the Step received is the step to stop the batch container, false otherwise
 	 */
     protected boolean isStopBatchContainerTriggered(Step step) {
     	if (step.getName().equals(batchContainerStopStep.getName())) {
     		LOG.info("Received Step: "+ batchContainerStopStep.getName() +". Stop listening for steps.");
     		return true;
     	}
     	return false;
  	}
 
     /**
      * Sleep for a specified amount of time before looking for more semaphore files to process
      */
     protected void sleep() {
         try {
             if (LOG.isDebugEnabled()) {
                 LOG.debug("Sleeping...");
             }
             Thread.sleep(getSemaphoreProcessingInterval());
         }
         catch (InterruptedException e) {
             throw new RuntimeException("BatchContainerStep encountered interrupt exception while trying to wait for the specified semaphore processing interval", e);
         }
     }
 
     /**
      * @return (in milliseconds) the amount of time to wait before looking for more semaphore files to process
      */
 	protected long getSemaphoreProcessingInterval() {
 		return Long.parseLong(getParameterService().getParameterValueAsString(BatchContainerStep.class, OLEConstants.SystemGroupParameterNames.BATCH_CONTAINER_SEMAPHORE_PROCESSING_INTERVAL));
 	}
 
 	/**
 	 * Initialize the structures necessary for logging the Steps' statistics
 	 */
     protected void initContainerResults() {
         containerResults = new StringBuffer("Container Results:\n");
         startedSteps = new HashMap<String, BatchStepFileDescriptor>();
         completedSteps = new ArrayList<BatchStepFileDescriptor[]>();
     }
 
     /**
      * Log the notification that the Step started to an internal buffer. Add the descriptor to the list of started steps.
      * The logFileName is used as a unique identifier in the list of started steps in order to identify it in the list of started steps on completion.
      *
      * @param runFile the step's run file descriptor
      * @param logFileName the name of the log created by the Step's executor.
      */
     protected void logStepStarted(BatchStepFileDescriptor runFile, String logFileName) {
         if (LOG.isDebugEnabled()) {
             LOG.debug("stepStarted: "+ runFile);
         }
 
         startedSteps.put(logFileName, runFile);
 
         containerResults.append("STARTED "+ runFile
                 +" "+ runFile.getStartedDate()
                 +" LOGFILE="+ logFileName
                 +"\n");
     }
 
     /**
      * Log the notification that the Step finished to an internal buffer. Remove the run descriptor from the list of started steps and add the run descriptor
      * and the result descriptor to the list of completed steps. The logFileName is used to locate the run descriptor from the list of started steps.
      *
      * @param resultFile the step's result file descriptor
      * @param logFileName the name of the log created by the Step's executor
      */
     protected void logStepFinished(BatchStepFileDescriptor resultFile, String logFileName) {
         if (LOG.isDebugEnabled()) {
             LOG.debug("stepFinished: "+ resultFile);
         }
 
         BatchStepFileDescriptor runFile = startedSteps.remove(logFileName);
 
         containerResults.append("COMPLETED "+ resultFile
                                 +" "+ resultFile.getCompletedDate()
                                 +" LOGFILE="+ logFileName
                                 +" STATUS="+ resultFile.getExtension()
                                 +(resultFile.isStepFileAnErrorResultFile() ? " EXCEPTION:\n"+ directory.getExceptionFromFile(resultFile) : "")
                                 +"\n");
 
         BatchStepFileDescriptor[] files = {runFile, resultFile};
         completedSteps.add(files);
     }
 
     /**
      * Print a summary of the steps that ran and the steps that haven't completed yet.
      */
 	protected void logContainerResultsSummary() {
 	    LOG.info("Printing container results...");
 
 	    containerResults.append("\n\nCompleted Steps: \n");
 	    if (completedSteps.isEmpty()) { containerResults.append("None"); }
 
 	    for(BatchStepFileDescriptor[] batchStepFile : completedSteps) {
 	        String status = batchStepFile[1].getExtension();
 	        Date startedDate = batchStepFile[0].getStartedDate();
 	        Date completedDate = batchStepFile[1].getCompletedDate();
 
             containerResults.append(batchStepFile[0] +"=" +status +"; S:"+ startedDate +" F:"+ completedDate +"\n");
 	    }
 
 	    containerResults.append("\n\nIncomplete Steps: \n");
 	    if (startedSteps.isEmpty()) { containerResults.append("None"); }
 
 	    for(Iterator<BatchStepFileDescriptor> iter = startedSteps.values().iterator(); iter.hasNext();) {
 	        BatchStepFileDescriptor batchStepFile = iter.next();
 
             Date startedDate = batchStepFile.getStartedDate();
 
             containerResults.append(batchStepFile +"; S:"+ startedDate +"\n");
 	    }
 
 	    LOG.info(containerResults);
 	}
 
 	/**
 	 * The path to the directory in which BatchContainerStep, BatchStepExecutor, and BatchStepTrigger will read and write the step semaphore files.
 	 *
 	 * @param batchContainerDirectory
 	 */
 	public void setBatchContainerDirectory(String batchContainerDirectory) {
 		this.batchContainerDirectory = batchContainerDirectory;
 	}
 
 	/**
 	 * The Step which indicates to the Batch Container to shut itself down.
 	 *
 	 * @param batchContainerStopStep
 	 */
 	public void setBatchContainerStopStep(Step batchContainerStopStep) {
 		this.batchContainerStopStep = batchContainerStopStep;
 	}
 
 	/**
 	 * @return the batchContainerStopStep
 	 */
 	public Step getBatchContainerStopStep() {
 	    return this.batchContainerStopStep;
 	}
 }