/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.oodt.cas.pge;

import com.google.common.base.Splitter;
import com.google.common.base.Strings;
import com.google.common.collect.Lists;
import org.apache.commons.lang.Validate;
import org.apache.oodt.cas.crawl.AutoDetectProductCrawler;
import org.apache.oodt.cas.crawl.ProductCrawler;
import org.apache.oodt.cas.crawl.StdProductCrawler;
import org.apache.oodt.cas.crawl.status.IngestStatus;
import org.apache.oodt.cas.crawl.structs.exceptions.CrawlerActionException;
import org.apache.oodt.cas.filemgr.structs.exceptions.CatalogException;
import org.apache.oodt.cas.filemgr.structs.exceptions.ConnectionException;
import org.apache.oodt.cas.filemgr.structs.exceptions.DataTransferException;
import org.apache.oodt.cas.metadata.Metadata;
import org.apache.oodt.cas.metadata.SerializableMetadata;
import org.apache.oodt.cas.metadata.exceptions.CasMetadataException;
import org.apache.oodt.cas.metadata.exceptions.MetExtractionException;
import org.apache.oodt.cas.metadata.exceptions.MetExtractorConfigReaderException;
import org.apache.oodt.cas.metadata.exceptions.NamingConventionException;
import org.apache.oodt.cas.metadata.filenaming.PathUtilsNamingConvention;
import org.apache.oodt.cas.pge.config.DynamicConfigFile;
import org.apache.oodt.cas.pge.config.OutputDir;
import org.apache.oodt.cas.pge.config.PgeConfig;
import org.apache.oodt.cas.pge.config.RegExprOutputFiles;
import org.apache.oodt.cas.pge.config.XmlFilePgeConfigBuilder;
import org.apache.oodt.cas.pge.exceptions.PGEException;
import org.apache.oodt.cas.pge.metadata.PgeMetadata;
import org.apache.oodt.cas.pge.metadata.PgeTaskMetKeys;
import org.apache.oodt.cas.pge.staging.FileManagerFileStager;
import org.apache.oodt.cas.pge.staging.FileStager;
import org.apache.oodt.cas.pge.writers.PcsMetFileWriter;
import org.apache.oodt.cas.pge.writers.SciPgeConfigFileWriter;
import org.apache.oodt.cas.workflow.exceptions.WorkflowException;
import org.apache.oodt.cas.workflow.metadata.CoreMetKeys;
import org.apache.oodt.cas.workflow.structs.WorkflowTaskConfiguration;
import org.apache.oodt.cas.workflow.structs.WorkflowTaskInstance;
import org.apache.oodt.cas.workflow.structs.exceptions.WorkflowTaskInstanceException;
import org.apache.oodt.cas.workflow.system.WorkflowManagerClient;
import org.apache.oodt.cas.workflow.system.rpc.RpcCommunicationFactory;
import org.apache.oodt.cas.workflow.util.ScriptFile;
import org.apache.oodt.commons.exceptions.CommonsException;
import org.apache.oodt.commons.exec.ExecUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.support.FileSystemXmlApplicationContext;

import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.net.MalformedURLException;
import java.net.URISyntaxException;
import java.net.URL;
import java.text.ParseException;
import java.util.Date;
import java.util.LinkedList;
import java.util.List;
import java.util.logging.FileHandler;
import java.util.logging.SimpleFormatter;
import java.util.regex.Pattern;

import static org.apache.oodt.cas.pge.metadata.PgeTaskMetKeys.*;
import static org.apache.oodt.cas.pge.metadata.PgeTaskStatus.CONF_FILE_BUILD;
import static org.apache.oodt.cas.pge.metadata.PgeTaskStatus.CRAWLING;
import static org.apache.oodt.cas.pge.metadata.PgeTaskStatus.RUNNING_PGE;
import static org.apache.oodt.cas.pge.metadata.PgeTaskStatus.STAGING_INPUT;
import static org.apache.oodt.cas.pge.util.GenericPgeObjectFactory.createConfigFilePropertyAdder;
import static org.apache.oodt.cas.pge.util.GenericPgeObjectFactory.createFileStager;
import static org.apache.oodt.cas.pge.util.GenericPgeObjectFactory.createPgeConfigBuilder;
import static org.apache.oodt.cas.pge.util.GenericPgeObjectFactory.createSciPgeConfigFileWriter;


/**
 * Runs a CAS-style Product Generation Executive based on the PCS Wrapper
 * Architecture from mattmann et al. on OCO.
 *
 * @author mattmann (Chris Mattmann)
 * @author bfoster (Brian Foster)
 */
public class PGETaskInstance implements WorkflowTaskInstance {

   protected Logger logger = LoggerFactory.getLogger(PGETaskInstance.class);

   /**
    * This JUL logger is kept for now to avoid large scale changes in logging dependencies.
    * Should be removed in future with an alternative to address the logging dependencies.
    */
   protected java.util.logging.Logger julLogger = java.util.logging.Logger.getLogger(PGETaskInstance.class.getName());

   private WorkflowManagerClient wmClient;
   private String workflowInstId;
   protected PgeMetadata pgeMetadata;
   protected PgeConfig pgeConfig;

   protected PGETaskInstance() {}

   @Override
   public void run(Metadata metadata, WorkflowTaskConfiguration config) throws WorkflowTaskInstanceException {
      logger.debug("Starting PGE Task instance...");
      try {
         // Initialize CAS-PGE.
         pgeMetadata = createPgeMetadata(metadata, config);
         pgeConfig = createPgeConfig();
         runPropertyAdders();
         workflowInstId = getWorkflowInstanceId();

         // use workflow ID specific logger from now on
         logger = LoggerFactory.getLogger(PGETaskInstance.class.getName() + "." + workflowInstId);
         logger.debug("Workflow instance ID is [{}]", workflowInstId);
         julLogger = createLogger();

         // Write out PgeMetadata.
         dumpMetadataIfRequested();

         // Setup the PGE.
         createExeDir();
         createOuputDirsIfRequested();
         updateStatus(CONF_FILE_BUILD.getWorkflowStatusName());
         createDynamicConfigFiles();
         updateStatus(STAGING_INPUT.getWorkflowStatusName());
         stageFiles();

         // Run the PGE.
         runPge();

         // Ingest products.
         ProductCrawler productCrawler = createProductCrawler();
         runIngestCrawler(productCrawler);
         productCrawler.shutdown();

         // Commit dynamic metadata.
         updateDynamicMetadata();
      } catch (Exception e) {
         logger.error("PGETask FAILED!!! Error occurred when running", e);
         throw new WorkflowTaskInstanceException("PGETask FAILED!!! : " + e.getMessage(), e);
      }
   }

   protected void updateStatus(String status) throws Exception {
      logger.info("Updating status to workflow as [" + status + "]");
      if (!getWorkflowManagerClient().updateWorkflowInstanceStatus(workflowInstId, status)) {
         throw new PGEException("Failed to update workflow status : client returned false");
      }
   }

   protected java.util.logging.Logger createLogger() throws IOException, PGEException {
      File logDir = new File(pgeConfig.getExeDir(), "logs");
      if (!(logDir.exists() || logDir.mkdirs())) {
         throw new PGEException("mkdirs for logs directory return false");
      }

      java.util.logging.Logger logger = java.util.logging.Logger.getLogger(PGETaskInstance.class.getName() + "." + workflowInstId);
      // TODO Need to find an alternative way to add a dynamic handler to write workflowInstance logs to a separate file
      FileHandler handler = new FileHandler(new File(logDir, createLogFileName()).getAbsolutePath());
      handler.setEncoding("UTF-8");
      handler.setFormatter(new SimpleFormatter());
      logger.addHandler(handler);
      return logger;
   }

   protected String createLogFileName() {
      String filenamePattern = pgeMetadata.getMetadata(LOG_FILENAME_PATTERN);
      if (filenamePattern != null) {
         return filenamePattern;
      } else {
         return pgeMetadata.getMetadata(NAME) + "." + System.currentTimeMillis() + ".log";
      }
   }

   protected PgeMetadata createPgeMetadata(Metadata dynMetadata, WorkflowTaskConfiguration config) {
      logger.info("Converting workflow configuration to static metadata");
      logger.debug("PGE Metadata Config: {}", config.getProperties().entrySet());
      logger.debug("PGE Metadata Dynamic Metadata: {}", dynMetadata.getMap());
      Metadata staticMetadata = new Metadata();
      for (Object objKey : config.getProperties().keySet()) {
         String key = (String) objKey;
         PgeTaskMetKeys metKey = PgeTaskMetKeys.getByName(key);
         if (metKey != null && metKey.isVector()) {
            List<String> values = Lists.newArrayList(
                  Splitter.on(",").trimResults()
                  .omitEmptyStrings()
                  .split(config.getProperty(key)));
            logger.debug("Adding static metadata: key = [{}] value = {}", key, values);
            staticMetadata.addMetadata(key, values);
         } else {
            String value = config.getProperty(key);
            logger.debug("Adding static metadata: key = [{}] value = {}", key, value);
            staticMetadata.addMetadata(key, value);
         }
      }

      logger.info("Loading workflow context metadata...");
      for (String key : dynMetadata.getAllKeys()) {
         logger.debug("Adding dynamic metadata: key = [{}] value = {}", key, dynMetadata.getAllMetadata(key));
      }
      return new PgeMetadata(staticMetadata, dynMetadata);
   }

   protected PgeConfig createPgeConfig() throws Exception {
      logger.info("Create PgeConfig...");
      String pgeConfigBuilderClass = pgeMetadata.getMetadata(PGE_CONFIG_BUILDER);
      if (pgeConfigBuilderClass != null) {
         logger.info("Using PgeConfigBuilder: {}", pgeConfigBuilderClass);
         return createPgeConfigBuilder(pgeConfigBuilderClass, logger).build(pgeMetadata);
      } else {
         logger.info("Using default PgeConfigBuilder: {}", XmlFilePgeConfigBuilder.class.getCanonicalName());
         return new XmlFilePgeConfigBuilder().build(pgeMetadata);
      }
   }

   protected void runPropertyAdders() throws PGEException {
      try {
         logger.info("Loading/Running property adders...");
         List<String> propertyAdders = pgeMetadata.getAllMetadata(PROPERTY_ADDERS);
         if (propertyAdders != null) {
            for (String propertyAdder : propertyAdders) {
               runPropertyAdder(loadPropertyAdder(propertyAdder));
            }
         } else {
            logger.info("No property adders specified");
         }
      } catch (Exception e) {
         logger.error("Error occurred when running property adders", e);
         throw new PGEException("Failed to instantiate/run Property Adders : " + e.getMessage(), e);
      }
   }

   protected ConfigFilePropertyAdder loadPropertyAdder(String propertyAdderClasspath) {
      logger.debug("Loading property adder: {}", propertyAdderClasspath);
      return createConfigFilePropertyAdder(propertyAdderClasspath, logger);
   }

   protected void runPropertyAdder(ConfigFilePropertyAdder propAdder) {
      logger.info("Running property adder: {}", propAdder.getClass().getCanonicalName());
      propAdder.addConfigProperties(pgeMetadata, pgeConfig.getPropertyAdderCustomArgs());
   }

   protected WorkflowManagerClient getWorkflowManagerClient() throws MalformedURLException {
      if (this.wmClient == null) {
         String url = pgeMetadata.getMetadata(WORKFLOW_MANAGER_URL);
         logger.info("Creating WorkflowManager client for url [" + url + "]");
         Validate.notNull(url, "Must specify " + WORKFLOW_MANAGER_URL);
         this.wmClient = RpcCommunicationFactory.createClient(new URL(url));
      }

      return this.wmClient;
   }

   protected String getWorkflowInstanceId() {
      String instanceId = pgeMetadata.getMetadata(CoreMetKeys.WORKFLOW_INST_ID);
      logger.debug("Workflow instanceId is [{}]", instanceId);
      Validate.notNull(instanceId, "Must specify " + CoreMetKeys.WORKFLOW_INST_ID);
      return instanceId;
   }

   protected void dumpMetadataIfRequested() throws IOException {
      if (Boolean.parseBoolean(pgeMetadata
            .getMetadata(DUMP_METADATA))) {
         new SerializableMetadata(pgeMetadata.asMetadata())
               .writeMetadataToXmlStream(new FileOutputStream(
                     getDumpMetadataPath()));
      }
   }

   protected String getDumpMetadataPath()  {
      return new File(pgeConfig.getExeDir()).getAbsolutePath() + "/"
            + getDumpMetadataName();
   }

   protected String getDumpMetadataName() {
      return "pgetask-metadata.xml";
   }

   protected void createExeDir() throws PGEException {
      logger.info("Creating PGE execution working directory: [{}]", pgeConfig.getExeDir());
      File executionDir = new File(pgeConfig.getExeDir());
      if (!(executionDir.exists() || executionDir.mkdirs())) {
         logger.warn("Unable to create execution working directory: {}", pgeConfig.getExeDir());
         throw new PGEException("mkdirs returned false for creating [" + pgeConfig.getExeDir() + "]");
      }
   }

   protected void createOuputDirsIfRequested() throws PGEException {
      for (OutputDir outputDir : pgeConfig.getOuputDirs()) {
         if (outputDir.isCreateBeforeExe()) {
            logger.info("Creating PGE file ouput directory: [{}]", outputDir.getPath());
            File dir = new File(outputDir.getPath());
            if (!(dir.exists() || dir.mkdirs())) {
               logger.warn("Unable to create output dirs: {}", outputDir.getPath());
               throw new PGEException("mkdir returned false for creating [" + outputDir.getPath() + "]");
            }
         }
      }
   }

   protected void stageFiles()
       throws PGEException, IOException, ConnectionException, CatalogException, URISyntaxException,
       DataTransferException, InstantiationException {
      if (pgeConfig.getFileStagingInfo() != null) {
         FileStager fileStager = getFileStager();
         logger.info("Starting file staging...");
         fileStager.stageFiles(pgeConfig.getFileStagingInfo(), pgeMetadata, logger);
      } else {
         logger.info("No files to stage.");
      }
   }

   protected FileStager getFileStager() {
      String fileStagerClass = pgeMetadata.getMetadata(FILE_STAGER);
      if (fileStagerClass != null) {
         logger.info("Loading FileStager [{}]", fileStagerClass);
         return createFileStager(fileStagerClass, logger);
      } else {
         logger.info("Using default FileStager ["
               + FileManagerFileStager.class.getCanonicalName() + "]");
         return new FileManagerFileStager();
      }
   }

   protected void createDynamicConfigFiles() throws IOException, PGEException {
      logger.info("Starting creation of sci pge config files...");
      for (DynamicConfigFile dynamicConfigFile : pgeConfig
            .getDynamicConfigFiles()) {
         createDynamicConfigFile(dynamicConfigFile);
      }
      logger.info("Successfully wrote all sci pge config files!");
   }

   protected void createDynamicConfigFile(DynamicConfigFile dynamicConfigFile)
       throws PGEException, IOException {
      Validate.notNull(dynamicConfigFile, "dynamicConfigFile cannot be null");
      logger.debug("Starting creation of sci pge config file: {}", dynamicConfigFile.getFilePath());

      // Create parent directory if it doesn't exist.
      File parentDir = new File(dynamicConfigFile.getFilePath()).getParentFile();
      if (!(parentDir.exists() || parentDir.mkdirs())) {
         logger.warn("Unable to create directory({}) with sci pge config file", dynamicConfigFile.getFilePath());
         throw new PGEException("Failed to create directory where sci pge config file ["
               + dynamicConfigFile.getFilePath() + "] was to be written");
      }

      // Load writer and write file.
      logger.debug("Loading writer class for sci pge config file [{}]", dynamicConfigFile.getFilePath());
      SciPgeConfigFileWriter writer = createSciPgeConfigFileWriter(dynamicConfigFile.getWriterClass(), logger);
      logger.debug("Loaded writer [{}] for sci pge config file [{}]",
              writer.getClass().getCanonicalName(),dynamicConfigFile.getFilePath());
      logger.info("Writing sci pge config file [{}]", dynamicConfigFile.getFilePath());
      File configFile = writer.createConfigFile(dynamicConfigFile.getFilePath(),
              pgeMetadata.asMetadata(), dynamicConfigFile.getArgs());
      if (!configFile.exists()) {
         logger.warn("Failed to create config file '{}'. File doesn't exist", configFile);
         throw new PGEException("Writer failed to create config file [" + configFile + "], exists returned false");
      }
   }

   protected ScriptFile buildPgeRunScript() {
      logger.debug("Creating PGE run script for shell [{}] with contents: {}",
              pgeConfig.getShellType(), pgeConfig.getExeCmds());
      ScriptFile sf = new ScriptFile(pgeConfig.getShellType());
      sf.setCommands(pgeConfig.getExeCmds());
      return sf;
   }

   protected File getScriptPath() {
      File script = new File(pgeConfig.getExeDir(), getPgeScriptName());
      logger.debug("Script file with be written to [{}]", script);
      return script;
   }

   protected String getPgeScriptName() {
      String pgeScriptName = "sciPgeExeScript_" + pgeMetadata.getMetadata(NAME);
      logger.debug("Generated script file name [{}]", pgeScriptName);
      return pgeScriptName;
   }

   protected void runPge() throws Exception {
      ScriptFile sf = null;
      try {
         long startTime = System.currentTimeMillis();
         logger.info("PGE start time [{}]", new Date(startTime));

         // create script to run
         sf = buildPgeRunScript();
         sf.writeScriptFile(getScriptPath().getAbsolutePath());

         // run script and evaluate whether success or failure
         updateStatus(RUNNING_PGE.getWorkflowStatusName());
         logger.debug("Starting execution of PGE: {}", sf.getCommands());
         if (!wasPgeSuccessful(ExecUtils.callProgram(
               pgeConfig.getShellType() + " " + getScriptPath(), julLogger,
               new File(pgeConfig.getExeDir()).getAbsoluteFile()))) {
            logger.error("PGE didn't finish successfully: {}", sf);
            throw new RuntimeException("Pge didn't finish successfully");
         } else {
            logger.info("Successfully completed running script file: '{}'", sf);
         }

         long endTime = System.currentTimeMillis();
         logger.info("PGE end time [{}]", new Date(startTime));

         long runTime = endTime - startTime;
         logger.info("PGE runtime in millis [{}]", runTime);

         pgeMetadata.replaceMetadata(PGE_RUNTIME, Long.toString(runTime));
      } catch (WorkflowException | IOException e) {
         logger.error("Error when executing PGE commands: {}", sf);
        throw new PGEException("Exception when executing PGE commands '" +
                (sf.getCommands()) + "' : " + e.getMessage(), e);
      }
   }

   protected boolean wasPgeSuccessful(int returnCode) {
      return returnCode == 0;
   }

   protected void processOutput() throws IOException {
      logger.debug("Processing output");
      for (final OutputDir outputDir : this.pgeConfig.getOuputDirs()) {
         File[] createdFiles = new File(outputDir.getPath()).listFiles();
         if (createdFiles != null) {
            for (File createdFile : createdFiles) {
               Metadata outputMetadata = new Metadata();
               for (RegExprOutputFiles regExprFiles : outputDir
                       .getRegExprOutputFiles()) {
                  if (Pattern.matches(regExprFiles.getRegExp(), createdFile
                          .getName())) {
                     try {
                        PcsMetFileWriter writer = (PcsMetFileWriter) Class
                                .forName(regExprFiles.getConverterClass())
                                .newInstance();
                        outputMetadata.replaceMetadata(this.getMetadataForFile(
                                (regExprFiles.getRenamingConv() != null)
                                        ? createdFile = this.renameFile(createdFile, regExprFiles.getRenamingConv())
                                        : createdFile, writer, regExprFiles.getArgs()));
                     } catch (Exception e) {
                        logger.error("Failed to create metadata file for '{}'", createdFile, e);
                     }
                  }
               }

               if (outputMetadata.getAllKeys().size() > 0) {
                  this.writeFromMetadata(outputMetadata, createdFile.getAbsolutePath()
                          + "." + this.pgeMetadata.getMetadata(MET_FILE_EXT));
               }
            }
         }
      }
   }

	protected File renameFile(File file, PathUtilsNamingConvention renamingConv)
        throws NamingConventionException {
		Metadata curMetadata = this.pgeMetadata.asMetadata();
		curMetadata.replaceMetadata(renamingConv.getTmpReplaceMet());
		return renamingConv.rename(file, curMetadata);
	}

	protected Metadata getMetadataForFile(File sciPgeCreatedDataFile,
			PcsMetFileWriter writer, Object[] args)
        throws PGEException, MetExtractorConfigReaderException, ParseException, MetExtractionException,
        CommonsException, CasMetadataException, FileNotFoundException {
		return writer.getMetadataForFile(sciPgeCreatedDataFile,
				this.pgeMetadata, args);
	}

	protected void writeFromMetadata(Metadata metadata, String toMetFilePath)
			throws IOException {
		new SerializableMetadata(metadata, "UTF-8", false)
				.writeMetadataToXmlStream(new FileOutputStream(toMetFilePath));
	}

   protected ProductCrawler createProductCrawler()
           throws MalformedURLException, IllegalAccessException, CrawlerActionException, MetExtractionException,
           InstantiationException, FileNotFoundException, ClassNotFoundException {
      /* create a ProductCrawler based on whether or not the output dir specifies a MIME_EXTRACTOR_REPO */
      logger.info("Configuring ProductCrawler...");
      ProductCrawler crawler;
      if (pgeMetadata.getMetadata(MIME_EXTRACTOR_REPO) != null &&
              !pgeMetadata.getMetadata(MIME_EXTRACTOR_REPO).equals("")) {
         crawler = new AutoDetectProductCrawler();
         ((AutoDetectProductCrawler) crawler).
                 setMimeExtractorRepo(pgeMetadata.getMetadata(MIME_EXTRACTOR_REPO));
      } else {
         crawler = new StdProductCrawler();
      }

      crawler.setClientTransferer(pgeMetadata
              .getMetadata(INGEST_CLIENT_TRANSFER_SERVICE_FACTORY));
      crawler.setFilemgrUrl(pgeMetadata.getMetadata(INGEST_FILE_MANAGER_URL));
      String crawlerConfigFile = pgeMetadata.getMetadata(CRAWLER_CONFIG_FILE);
      if (!Strings.isNullOrEmpty(crawlerConfigFile)) {
         crawler.setApplicationContext(
                 new FileSystemXmlApplicationContext(crawlerConfigFile));
         List<String> actionIds = pgeMetadata.getAllMetadata(ACTION_IDS);
         if (actionIds != null) {
            crawler.setActionIds(actionIds);
         }
      }
      crawler.setRequiredMetadata(pgeMetadata.getAllMetadata(REQUIRED_METADATA));
      crawler.setCrawlForDirs(Boolean.parseBoolean(pgeMetadata
              .getMetadata(CRAWLER_CRAWL_FOR_DIRS)));
      crawler.setNoRecur(!Boolean.parseBoolean(
              pgeMetadata.getMetadata(CRAWLER_RECUR)));
      logger.debug("Passing Workflow Metadata to CAS-Crawler as global metadata . . .");
      crawler.setGlobalMetadata(pgeMetadata.asMetadata(PgeMetadata.Type.DYNAMIC));
      logger.debug("Created ProductCrawler [{}]", crawler.getClass().getCanonicalName());
      return crawler;
   }

   protected void runIngestCrawler(ProductCrawler crawler) throws Exception {
      // Determine if we need to create Metadata files
	   if (crawler instanceof StdProductCrawler){
		   this.processOutput();
	   }
	   
	   // Determine directories to crawl.
      List<File> crawlDirs = new LinkedList<File>();
      for (OutputDir outputDir : pgeConfig.getOuputDirs()) {
         crawlDirs.add(new File(outputDir.getPath()));
      }

      // Start crawlin...
      updateStatus(CRAWLING.getWorkflowStatusName());
      boolean attemptIngestAll = Boolean.parseBoolean(pgeMetadata
            .getMetadata(ATTEMPT_INGEST_ALL));
      for (File crawlDir : crawlDirs) {
         logger.info("Crawling for products in [" + crawlDir + "]");
         crawler.crawl(crawlDir);
         if (!attemptIngestAll) {
            verifyIngests(crawler);
         }
      }
      if (attemptIngestAll) {
         verifyIngests(crawler);
      }
   }

   protected void verifyIngests(ProductCrawler crawler) throws PGEException {
      logger.debug("Verifying ingests successful...");
      boolean ingestsSuccess = true;
      String exceptionMsg = "";
      for (IngestStatus status : crawler.getIngestStatus()) {
         if (status.getResult().equals(IngestStatus.Result.FAILURE)) {
            exceptionMsg += (exceptionMsg.equals("") ? "" : " : ")
                  + "Failed to ingest product [file='"
                  + status.getProduct().getAbsolutePath() + "',result='"
                  + status.getResult() + "',msg='" + status.getMessage() + "']";
            ingestsSuccess = false;
         } else if (!status.getResult().equals(IngestStatus.Result.SUCCESS)) {
            logger.warn("Product was not ingested [file='{}', result='{}', msg='{}'",
                    status.getProduct().getAbsolutePath(), status.getResult(),status.getMessage());
         }
      }

      if (!ingestsSuccess) {
         logger.error("Ingest wasn't successful: {}", exceptionMsg);
         throw new PGEException(exceptionMsg);
      } else {
         logger.info("Ingests were successful");
      }
   }

   protected void updateDynamicMetadata() throws Exception {
      logger.debug("Updating dynamic metadata ...");
      pgeMetadata.commitMarkedDynamicMetadataKeys();
      getWorkflowManagerClient()
              .updateMetadataForWorkflow(workflowInstId, pgeMetadata.asMetadata(PgeMetadata.Type.DYNAMIC));
   }

   public String getWorkflowInstId() {
      return workflowInstId;
   }

   public void setWorkflowInstId(String workflowInstId) {
      logger.debug("Set workflow instance ID: {}", workflowInstId);
      this.workflowInstId = workflowInstId;
   }

   public void setWmClient(WorkflowManagerClient wmClient) {
      this.wmClient = wmClient;
   }

   @Override
   public void finalize() throws IOException {
      logger.debug("Finalizing ...");
      if (wmClient != null) {
         wmClient.close();
         logger.debug("Workflow manager client closed");
      }
   }
}