D-Net

package eu.dnetlib.msro.workflows.procs;

import java.io.StringReader;

import java.util.HashMap;

import java.util.Map;

import java.util.concurrent.Executors;

import java.util.concurrent.TimeUnit;

import eu.dnetlib.enabling.common.Stoppable;

import eu.dnetlib.enabling.common.StoppableDetails;

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

import eu.dnetlib.msro.workflows.graph.Graph;

import eu.dnetlib.msro.workflows.graph.GraphLoader;

import eu.dnetlib.msro.workflows.util.ProcessCallback;

import eu.dnetlib.msro.workflows.util.WorkflowsConstants;

import eu.dnetlib.rmi.enabling.ISLookUpService;

import eu.dnetlib.rmi.manager.MSROException;

import org.apache.commons.lang3.StringUtils;

import org.apache.commons.lang3.math.NumberUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.dom4j.Document;

import org.dom4j.Element;

import org.dom4j.io.SAXReader;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.beans.factory.annotation.Required;

/**

 * Created by michele on 20/11/15.

 */

public class WorkflowExecutor implements Stoppable {

	private static final Log log = LogFactory.getLog(WorkflowExecutor.class);

	@Autowired

	private UniqueServiceLocator serviceLocator;

	private GraphLoader graphLoader;

	private ProcessRegistry processRegistry;

	private ProcessFactory processFactory;

	private ProcessEngine processEngine;

	private boolean paused = false;

	public void init() {

		Executors.newSingleThreadScheduledExecutor().scheduleAtFixedRate(new Runnable() {

			@Override

			public void run() {

				if (isPaused() || WorkflowExecutor.this.processRegistry.countRunningWfs() >= WorkflowsConstants.MAX_RUNNING_PROCS_SIZE) { return; }

				final WorkflowProcess process = WorkflowExecutor.this.processRegistry.nextProcessToStart();

				if (process != null) {

					WorkflowExecutor.this.processEngine.startProcess(process);

				} else {

					log.debug("WorkflowProcess queue is empty");

				}

			}

		}, 10, 10, TimeUnit.SECONDS);

	}

	public String startRepoHiWorkflow(final String profileId, final String dsId, final String iface, final ProcessCallback processCallback) throws Exception {

		if (isPaused()) {

			log.warn("Wf " + profileId + " not launched, because WorkflowExecutor is preparing for shutdown");

			throw new MSROException("WorkflowExecutor is preparing for shutdown");

		}

		try {

			final String profile = this.serviceLocator.getService(ISLookUpService.class).getResourceProfile(profileId);

			final Document doc = new SAXReader().read(new StringReader(profile));

			final String name = doc.valueOf("//WORKFLOW_NAME");

			final String family = doc.valueOf("//WORKFLOW_FAMILY");

			final int priority = NumberUtils.toInt("//WORKFLOW_PRIORITY", WorkflowsConstants.DEFAULT_WF_PRIORITY);

			final boolean isReady = doc.valueOf("//CONFIGURATION/@status").equals(WorkflowsConstants.WorkflowStatus.EXECUTABLE.toString());

			final boolean isDisabled = doc.valueOf("//CONFIGURATION/@start").equals("disabled");

			if (!isReady || isDisabled) {

				log.warn("Wf " + profileId + " not launched, because it is not ready to start or it is disabled");

				throw new MSROException("Workflow " + profileId + " is not ready to start");

			}

			final Map<String, String> globalParams = new HashMap<String, String>();

			for (final Object o : doc.selectNodes("//CONFIGURATION/PARAMETERS/PARAM")) {

				final Element p = (Element) o;

				globalParams.put(p.valueOf("@name"), p.getTextTrim());

			}

			final Graph graph = this.graphLoader.loadGraph(doc, globalParams);

			final WorkflowProcess process =

					this.processFactory.newProcess(name, family, dsId, iface, graph, priority, profileId, false, globalParams, processCallback);

			return this.processRegistry.registerProcess(process, profileId);

		} catch (final Exception e) {

			log.error("Error parsing workflow: " + profileId, e);

			throw new MSROException("Error parsing workflow");

		}

	}

	public String startWorkflow(final String profileId, final ProcessCallback processCallback) throws Exception {

		if (isPaused()) {

			log.warn("Wf " + profileId + " not launched, because WorkflowExecutor is preparing for shutdown");

			throw new MSROException("WorkflowExecutor is preparing for shutdown");

		}

		try {

			final String profile = this.serviceLocator.getService(ISLookUpService.class).getResourceProfile(profileId);

			final Document doc = new SAXReader().read(new StringReader(profile));

			final String name = doc.valueOf("//WORKFLOW_NAME");

			final String family = doc.valueOf("//WORKFLOW_FAMILY");

			final int priority = NumberUtils.toInt("//WORKFLOW_PRIORITY", WorkflowsConstants.DEFAULT_WF_PRIORITY);

			final boolean isReady = doc.valueOf("//CONFIGURATION/@status").equals(WorkflowsConstants.WorkflowStatus.EXECUTABLE.toString());

			final boolean isDisabled = doc.valueOf("//CONFIGURATION/@start").equals("disabled");

			final String dsId = doc.valueOf("//DATASOURCE/@id");

			final String iface = doc.valueOf("//DATASOURCE/@interface");

			if (!isReady || isDisabled) {

				log.warn("Wf " + profileId + " not launched, because it is not ready to start or it is disabled");

				throw new MSROException("Workflow " + profileId + " is not ready to start");

			}

			final Map<String, String> globalParams = new HashMap<String, String>();

			for (final Object o : doc.selectNodes("//CONFIGURATION/PARAMETERS/PARAM")) {

				final Element p = (Element) o;

				globalParams.put(p.valueOf("@name"), p.getTextTrim());

			}

			final Graph graph = this.graphLoader.loadGraph(doc, globalParams);

			final WorkflowProcess process =

					this.processFactory.newProcess(name, family, dsId, iface, graph, priority, profileId, false, globalParams, processCallback);

			return this.processRegistry.registerProcess(process, profileId);

		} catch (final Exception e) {

			log.error("Error parsing workflow: " + profileId, e);

			throw new MSROException("Error parsing workflow");

		}

	}

	public String startWorkflowTemplate(final String profileId,

			final String name,

			final String family,

			final int priority,

			final String dsId,

			final String iface,

			final Map<String, String> params,

			final ProcessCallback processCallback) throws Exception {

		if (isPaused()) {

			log.warn("Wf template " + profileId + " not launched, because WorkflowExecutor is preparing for shutdown");

			throw new MSROException("WorkflowExecutor is preparing for shutdown");

		}

		try {

			final String profile = this.serviceLocator.getService(ISLookUpService.class).getResourceProfile(profileId);

			final Document doc = new SAXReader().read(new StringReader(profile));

			final Map<String, String> globalParams = new HashMap<String, String>();

			for (final Object o : doc.selectNodes("//CONFIGURATION/PARAMETERS/PARAM")) {

				final Element p = (Element) o;

				final String pname = p.valueOf("@name");

				if (StringUtils.isNotBlank(params.get(pname))) {

					globalParams.put(pname, params.get(pname));

				} else if (p.selectSingleNode("@default") != null) {

					globalParams.put(pname, p.valueOf("@default"));

				} else if (!StringUtils.equalsIgnoreCase(p.valueOf("@required"), "true")) {

					globalParams.put(pname, "");

				} else {

					throw new MSROException("A required parameter is missing in wf template:" + pname);

				}

			}

			final Graph graph = this.graphLoader.loadGraph(doc, globalParams);

			final WorkflowProcess process =

					this.processFactory.newProcess(name, family, dsId, iface, graph, priority, profileId, true, globalParams, processCallback);

			return this.processRegistry.registerProcess(process);

		} catch (final Exception e) {

			log.error("Error parsing workflow template: " + profileId, e);

			throw new MSROException("Error parsing workflow template", e);

		}

	}

	@Override

	public void stop() {

		this.paused = true;

	}

	@Override

	public void resume() {

		this.paused = false;

	}

	@Override

	public StoppableDetails getStopDetails() {

		final int count = this.processRegistry.countRunningWfs();

		final StoppableDetails.StopStatus status =

				isPaused() ? (count == 0 ? StoppableDetails.StopStatus.STOPPED : StoppableDetails.StopStatus.STOPPING) : StoppableDetails.StopStatus.RUNNING;

		return new StoppableDetails("D-NET workflow manager", "Running workflows: " + count, status);

	}

	public ProcessRegistry getProcessRegistry() {

		return this.processRegistry;

	}

	@Required

	public void setProcessRegistry(final ProcessRegistry processRegistry) {

		this.processRegistry = processRegistry;

	}

	public GraphLoader getGraphLoader() {

		return this.graphLoader;

	}

	@Required

	public void setGraphLoader(final GraphLoader graphLoader) {

		this.graphLoader = graphLoader;

	}

	public ProcessFactory getProcessFactory() {

		return this.processFactory;

	}

	@Required

	public void setProcessFactory(final ProcessFactory processFactory) {

		this.processFactory = processFactory;

	}

	public ProcessEngine getProcessEngine() {

		return this.processEngine;

	}

	@Required

	public void setProcessEngine(final ProcessEngine processEngine) {

		this.processEngine = processEngine;

	}

	public boolean isPaused() {

		return this.paused;

	}

	public void setPaused(final boolean paused) {

		this.paused = paused;

	}

}