D-Net

package eu.dnetlib.msro.workflows.procs;

import java.util.*;

import java.util.stream.Collectors;

import com.google.common.base.Throwables;

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

import eu.dnetlib.miscutils.datetime.DateUtils;

import eu.dnetlib.msro.logging.DnetLogger;

import eu.dnetlib.msro.notification.EmailDispatcher;

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

import eu.dnetlib.msro.workflows.nodes.ProcessNode;

import eu.dnetlib.msro.workflows.procs.WorkflowProcess.Status;

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

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

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

import eu.dnetlib.rmi.enabling.ISRegistryService;

import org.antlr.stringtemplate.StringTemplate;

import org.apache.commons.io.IOUtils;

import org.apache.commons.lang3.StringEscapeUtils;

import org.apache.commons.lang3.StringUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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

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

/**

 * Created by michele on 26/11/15.

 */

public class ProcessEngine {

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

	@Autowired

	private UniqueServiceLocator serviceLocator;

	private NodeHelper nodeHelper;

	private DnetLogger dnetLogger;

	private EmailDispatcher emailDispatcher;

	public void startProcess(final WorkflowProcess process) {

		log.info(process.getGraph());

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

		final long now = DateUtils.now();

		process.setStatus(WorkflowProcess.Status.EXECUTING);

		process.setStartDate(now);

		process.setLastActivityDate(now);

		try {

			for (final GraphNode node : process.getGraph().startNodes()) {

				final ProcessNode pNode = this.nodeHelper.newProcessNode(node, process);

				final Token token = new Token(node.getName(), newNodeTokenCallback(process, node));

				token.getEnv().addAttributes(process.getEnv().getAttributes());

				process.getTokens().add(token);

				pNode.execute(token);

			}

		} catch (final Throwable e) {

			log.error("WorkflowProcess node instantiation failed", e);

			process.setStatus(WorkflowProcess.Status.FAILURE);

		}

	}

	public void releaseToken(final WorkflowProcess process, final GraphNode oldGraphNode, final Token oldToken) {

		process.setLastActivityDate(DateUtils.now());

		try {

			for (final GraphNode node : process.getGraph().nextNodes(oldGraphNode, oldToken.getNextArc())) {

				if (node.isJoin() || node.isSucessNode()) {

					if (!process.getPausedJoinNodeTokens().containsKey(node.getName())) {

						process.getPausedJoinNodeTokens().put(node.getName(), new ArrayList<Token>());

					}

					final List<Token> list = process.getPausedJoinNodeTokens().get(node.getName());

					list.add(oldToken);

					if (list.size() == process.getGraph().getNumberOfIncomingArcs(node)) {

						final ProcessNode pNode = this.nodeHelper.newProcessNode(node, process);

						final Token token = new Token(node.getName(), newNodeTokenCallback(process, node));

						token.getEnv().addAttributes(mergeEnvParams(list.toArray(new Token[list.size()])));

						process.getTokens().add(token);

						process.setLastActivityDate(DateUtils.now());

						if (node.isSucessNode()) {

							markAsCompleted(process, token);

						} else {

							pNode.execute(token);

						}

					}

				} else {

					final ProcessNode pNode = this.nodeHelper.newProcessNode(node, process);

					final Token token = new Token(node.getName(), newNodeTokenCallback(process, node));

					token.getEnv().addAttributes(oldToken.getEnv().getAttributes());

					process.getTokens().add(token);

					process.setLastActivityDate(DateUtils.now());

					pNode.execute(token);

				}

			}

		} catch (final Throwable e) {

			log.error("WorkflowProcess node instantiation failed", e);

			process.setStatus(WorkflowProcess.Status.FAILURE);

			process.setError(e.getMessage());

			process.setErrorStacktrace(Throwables.getStackTraceAsString(e));

			process.setLastActivityDate(DateUtils.now());

		}

	}

	private NodeTokenCallback newNodeTokenCallback(final WorkflowProcess process, final GraphNode node) {

		return new NodeTokenCallback() {

			@Override

			public void onSuccess(final Token token) {

				releaseToken(process, node, token);

			}

			@Override

			public void onFail(final Token token) {

				completeProcess(process, token);

			}

		};

	}

	private Map<String, Object> mergeEnvParams(final Token... tokens) {

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

		Arrays.stream(tokens).forEach(t -> map.putAll(t.getEnv().getAttributes()));

		return map;

	}

	private void markAsCompleted(final WorkflowProcess process, final Token token) {

		completeProcess(process, token);

	}

	private void completeProcess(final WorkflowProcess process, final Token token) {

		if (token.isActive()) {

			if (StringUtils.isNotBlank(token.getError())) {

				token.releaseAsFailed(token.getError());

			} else {

				token.release();

			}

		}

		final long now = token.getEndDate();

		process.setLastActivityDate(now);

		process.setEndDate(now);

		process.setStatus(token.isFailed() ? WorkflowProcess.Status.FAILURE : WorkflowProcess.Status.SUCCESS);

		if (token.isFailed()) {

			process.setStatus(Status.FAILURE);

			process.setError(token.getError());

			process.setErrorStacktrace(token.getErrorStackTrace());

			process.setLastActivityDate(DateUtils.now());

		}

		if (process.getCallback() != null) {

			if (token.isFailed()) {

				process.getCallback().onFail();

			} else {

				process.getCallback().onSuccess();

			}

		}

		process.setOutputParams(filterOutputParams(token.getEnv().getAttributes()));

		final String profileId = process.getProfileId();

		if (!process.isTemplate() && StringUtils.isNotBlank(profileId)) {

			try {

				final String template = IOUtils.toString(getClass().getResourceAsStream("/eu/dnetlib/msro/workflows/templates/workflow_status.xml.st"));

				final StringTemplate st = new StringTemplate(template);

				st.setAttribute("procId", StringEscapeUtils.escapeXml11(process.getId()));

				st.setAttribute("date", StringEscapeUtils.escapeXml11(DateUtils.calculate_ISO8601(now)));

				st.setAttribute("params", process.getOutputParams());

				if (process.getStatus() == Status.FAILURE) {

					st.setAttribute("error", StringEscapeUtils.escapeXml11(process.getError()));

				}

				this.serviceLocator.getService(ISRegistryService.class).updateProfileNode(profileId, "//STATUS", st.toString());

			} catch (final Exception e) {

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

				process.setStatus(WorkflowProcess.Status.FAILURE);

				process.setError("Error updating workflow profile: " + profileId);

				process.setErrorStacktrace(Throwables.getStackTraceAsString(e));

			}

		}

		this.dnetLogger.newLogMessage()

				.addDetails(process.getOutputParams())

				.addDetail(WorkflowsConstants.LOG_WF_NAME, process.getName())

				.addDetail(WorkflowsConstants.LOG_WF_FAMILY, process.getFamily())

				.addDetail(WorkflowsConstants.LOG_WF_PRIORITY, "" + process.getPriority())

				.addDetail(WorkflowsConstants.LOG_WF_PROCESS_ID, process.getId())

				.addDetail(WorkflowsConstants.LOG_WF_PROCESS_STATUS, process.getStatus().toString())

				.addDetail(WorkflowsConstants.LOG_WF_PROCESS_START_DATE, Long.toString(process.getStartDate()))

				.addDetail(WorkflowsConstants.LOG_WF_PROCESS_END_DATE, Long.toString(process.getEndDate()))

				.addDetail(WorkflowsConstants.LOG_WF_PROFILE_ID, process.isTemplate() ? null : process.getProfileId())

				.addDetail(WorkflowsConstants.LOG_WF_PROFILE_TEMPLATE_ID, process.isTemplate() ? process.getProfileId() : null)

				.addDetail(WorkflowsConstants.LOG_DATASOURCE_ID, process.getDsId())

				.addDetail(WorkflowsConstants.LOG_DATASOURCE_NAME, process.getDsName())

				.addDetail(WorkflowsConstants.LOG_DATASOURCE_INTERFACE, process.getDsInterface())

				.addDetail(WorkflowsConstants.LOG_SYSTEM_ERROR, process.getError())

				.addDetail(WorkflowsConstants.LOG_SYSTEM_ERROR_STACKTRACE, process.getErrorStacktrace())

				.flush();

		this.emailDispatcher.sendMails(process);

	}

	private Map<String, String> filterOutputParams(final Map<String, Object> map) {

		return map == null ? new HashMap<>() : map.entrySet().stream()

				.filter(e -> StringUtils.isNotBlank(e.getKey()))

				.filter(e -> e.getValue() != null)

				.filter(e -> e.getKey().startsWith(WorkflowsConstants.DATASOURCE_PREFIX) || e.getKey().startsWith(WorkflowsConstants.MAIN_LOG_PREFIX))

				.collect(Collectors.toMap(

						e -> StringEscapeUtils.escapeXml11(e.getKey()),

						e -> StringEscapeUtils.escapeXml11(e.getValue().toString())));

	}

	public DnetLogger getDnetLogger() {

		return this.dnetLogger;

	}

	@Required

	public void setDnetLogger(final DnetLogger dnetLogger) {

		this.dnetLogger = dnetLogger;

	}

	public NodeHelper getNodeHelper() {

		return this.nodeHelper;

	}

	@Required

	public void setNodeHelper(final NodeHelper nodeHelper) {

		this.nodeHelper = nodeHelper;

	}

	public EmailDispatcher getEmailDispatcher() {

		return this.emailDispatcher;

	}

	@Required

	public void setEmailDispatcher(final EmailDispatcher emailDispatcher) {

		this.emailDispatcher = emailDispatcher;

	}

}