D-Net

package eu.dnetlib.msro.worker;

import java.io.File;

import java.io.FileOutputStream;

import java.io.StringReader;

import java.util.Comparator;

import java.util.Map;

import java.util.concurrent.PriorityBlockingQueue;

import org.apache.commons.io.IOUtils;

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

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.dom4j.Document;

import org.dom4j.DocumentException;

import org.dom4j.io.SAXReader;

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

import com.googlecode.sarasvati.Graph;

import eu.dnetlib.common.ifaces.BlackboardExecutionCallback;

import eu.dnetlib.common.ifaces.Stoppable;

import eu.dnetlib.common.ifaces.StoppableDetails;

import eu.dnetlib.common.ifaces.StoppableDetails.StopStatus;

import eu.dnetlib.miscutils.DateUtils;

import eu.dnetlib.msro.worker.sarasvati.DnetGraphProcess;

import eu.dnetlib.msro.worker.sarasvati.InjectableMemEngine;

import eu.dnetlib.rmi.blackboard.LaunchWorkflowMessage;

import eu.dnetlib.rmi.soap.exceptions.ManagerServiceException;

public class WorkflowLauncher implements Stoppable {

	private InjectableMemEngine engine;

	private LocalWorkflowRegistry workflowRegistry;

	private boolean paused = false;

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

	private PriorityBlockingQueue<DnetGraphProcess> pendingProcs = new PriorityBlockingQueue<DnetGraphProcess>(20, new Comparator<DnetGraphProcess>() {

		@Override

		public int compare(final DnetGraphProcess p1, final DnetGraphProcess p2) {

			return NumberUtils.compare(p1.getPriority(), p2.getPriority());

		}

	});

	public void startsWfsInQueue() {

		if (pendingProcs.isEmpty() || isPaused()) { return; }

		int k = WorkflowConstants.MAX_WF_THREADS - workflowRegistry.countRunningWfs();

		for (int i = 0; i < k && !pendingProcs.isEmpty(); i++) {

			final DnetGraphProcess process = pendingProcs.poll();

			if (process != null) {

				log.info("Starting workflow: " + process);

				final long now = DateUtils.now();

				process.getEnv().setAttribute(WorkflowConstants.SYSTEM_START_DATE, now);

				process.getEnv().setAttribute(WorkflowConstants.SYSTEM_START_HUMAN_DATE, DateUtils.getDate_ISO8601(now));

				engine.startProcess(process);

			} else {

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

			}

		}

	}

	public void enqueueWf(final LaunchWorkflowMessage message,

			final Map<String, String> params,

			final BlackboardExecutionCallback<LaunchWorkflowMessage> callback)

			throws Exception {

		if (isPaused()) {

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

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

		}

		if (pendingProcs.size() > WorkflowConstants.MAX_PENDING_PROCS_SIZE) {

			log.warn("Wf not launched, Max number of pending procs reached: " + WorkflowConstants.MAX_PENDING_PROCS_SIZE);

			throw new ManagerServiceException("Max number of pending procs reached: " + WorkflowConstants.MAX_PENDING_PROCS_SIZE);

		}

		final File tmpFile = File.createTempFile("wftfs", null);

		try {

			final Graph graph = loadGraph(message.getWfXml());

			final DnetGraphProcess process = new DnetGraphProcess(graph, message.getWfPriority(), message, callback);

			workflowRegistry.registerProcess(process);

			process.getEnv().setAttribute(WorkflowConstants.SYSTEM_WF_PROCESS_ID, message.getProcId());

			process.getEnv().setAttribute(WorkflowConstants.SYSTEM_WF_NAME, graph.getName());

			process.getEnv().setAttribute(WorkflowConstants.SYSTEM_WF_PRIORITY, message.getWfPriority());

			if (params != null) {

				for (Map.Entry<String, String> e : params.entrySet()) {

					process.getEnv().setAttribute(e.getKey(), e.getValue());

				}

			}

			log.info("Process " + process + " in queue, priority=" + message.getWfPriority());

			pendingProcs.put(process);

		} catch (Exception e) {

			log.error("Error parsing workflow xml: " + message.getWfXml(), e);

			throw new IllegalArgumentException("Error parsing workflow");

		} finally {

			tmpFile.delete();

		}

	}

	private Graph loadGraph(final String xml) throws Exception {

		final File tmpFile = File.createTempFile("wftfs", null);

		try {

			IOUtils.copy(new StringReader(xml), new FileOutputStream(tmpFile));

			final SAXReader reader = new SAXReader();

			final Document doc = reader.read(tmpFile);

			final String graphName = doc.valueOf("/process-definition/@name");

			getEngine().getLoader().load(tmpFile);

			final Graph graph = getEngine().getRepository().getLatestGraph(graphName);

			if (graph == null) { throw new IllegalArgumentException("graph called " + graphName + " doesn't exist"); }

			return graph;

		} catch (DocumentException e) {

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

			throw e;

		} finally {

			tmpFile.delete();

		}

	}

	public boolean isPaused() {

		return paused;

	}

	public void setPaused(final boolean paused) {

		this.paused = paused;

	}

	@Override

	public StoppableDetails getStopDetails() {

		final int count = workflowRegistry.countRunningWfs();

		StoppableDetails.StopStatus status = StopStatus.RUNNING;

		if (isPaused()) {

			status = count == 0 ? StopStatus.STOPPED : StopStatus.STOPPING;

		}

		// graphProcessRegistry.listProcIds();

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

	}

	@Override

	public void stop() {

		this.paused = true;

	}

	@Override

	public void resume() {

		this.paused = false;

	}

	public InjectableMemEngine getEngine() {

		return engine;

	}

	@Required

	public void setEngine(final InjectableMemEngine engine) {

		this.engine = engine;

	}

	public LocalWorkflowRegistry getWorkflowRegistry() {

		return workflowRegistry;

	}

	@Required

	public void setWorkflowRegistry(final LocalWorkflowRegistry workflowRegistry) {

		this.workflowRegistry = workflowRegistry;

	}

}