dnet-applications/libs/dnet-wf-service/src/main/java/eu/dnetlib/manager/wf/workflows/procs/ProcessEngine.java

package eu.dnetlib.manager.wf.workflows.procs;

import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import com.google.common.base.Throwables;

import eu.dnetlib.manager.history.WorkflowLogger;
import eu.dnetlib.manager.wf.nodes.ProcessNode;
import eu.dnetlib.manager.wf.notification.EmailSender;
import eu.dnetlib.manager.wf.workflows.graph.GraphNode;
import eu.dnetlib.manager.wf.workflows.util.NodeCallback;
import eu.dnetlib.manager.wf.workflows.util.NodeHelper;

@Service
public class ProcessEngine {

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

	@Autowired
	private NodeHelper nodeHelper;
	@Autowired
	private EmailSender emailSender;
	@Autowired
	private WorkflowLogger wfLogger;

	public void startProcess(final WorkflowProcess process) {
		log.info(process.getGraph());

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

		final LocalDateTime now = LocalDateTime.now();
		process.setStatus(WorkflowProcess.Status.running);
		process.setStartDate(now);
		process.setLastActivityDate(now);

		try {
			for (final GraphNode graphNode : process.getGraph().startNodes()) {
				final ProcessNode pNode = nodeHelper.newProcessNode(graphNode, process);
				final Token token = new Token();
				token.getEnv().addAttributes(process.getEnv().getAttributes());
				process.getTokens().add(token);

				pNode.initInputParams(token.getEnv());
				pNode.execute(token, newNodeCallback(process, pNode, graphNode));
			}
		} catch (final Throwable e) {
			log.error("WorkflowProcess node instantiation failed", e);
			process.setStatus(WorkflowProcess.Status.failure);
		}
	}

	private NodeCallback newNodeCallback(final WorkflowProcess process, final ProcessNode pNode, final GraphNode graphNode) {
		return new NodeCallback() {

			@Override
			public void onComplete(final Token t) {
				pNode.saveOutputParams(t.getEnv());
				releaseToken(process, graphNode, t);
			}

			@Override
			public void onFail(final Token t) {
				pNode.saveOutputParams(t.getEnv());
				completeProcess(process, t);
			}
		};
	}

	public void releaseToken(final WorkflowProcess process, final GraphNode oldGraphNode, final Token oldToken) {
		process.setLastActivityDate(LocalDateTime.now());

		try {
			for (final GraphNode graphNode : process.getGraph().nextNodes(oldGraphNode, oldToken.getEnv())) {
				if (graphNode.isJoin() || graphNode.isSucessNode()) {
					if (!process.getPausedJoinNodeTokens().containsKey(graphNode.getName())) {
						process.getPausedJoinNodeTokens().put(graphNode.getName(), new ArrayList<Token>());
					}

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

					list.add(oldToken);

					if (list.size() == process.getGraph().getNumberOfIncomingArcs(graphNode)) {
						final Token token = new Token();
						token.getEnv().addAttributes(mergeEnvParams(list.toArray(new Token[list.size()])));
						final ProcessNode pNode = nodeHelper.newProcessNode(graphNode, process);

						process.getTokens().add(token);
						process.setLastActivityDate(LocalDateTime.now());

						if (graphNode.isSucessNode()) {
							completeProcess(process, token);
						} else {
							pNode.execute(token, newNodeCallback(process, pNode, graphNode));
						}
					}
				} else {
					final Token token = new Token();
					token.getEnv().addAttributes(oldToken.getEnv().getAttributes());
					final ProcessNode pNode = nodeHelper.newProcessNode(graphNode, process);

					process.getTokens().add(token);
					process.setLastActivityDate(LocalDateTime.now());
					pNode.execute(token, newNodeCallback(process, pNode, graphNode));
				}
			}
		} 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(LocalDateTime.now());
		}

	}

	private Map<String, Object> mergeEnvParams(final Token... tokens) {
		final Map<String, Object> map = new HashMap<>();
		Arrays.stream(tokens).forEach(t -> map.putAll(t.getEnv().getAttributes()));
		return map;
	}

	private void completeProcess(final WorkflowProcess process, final Token token) {
		token.checkStatus();
		process.complete(token);
		wfLogger.saveLog(process.asLog());
		emailSender.sendMails(process);
	}

}