D-Net

package eu.dnetlib.msro.workflows.procs;

import java.util.ArrayList;

import java.util.Arrays;

import java.util.HashMap;

import java.util.List;

import java.util.Map;

import java.util.stream.Collectors;

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.stereotype.Component;

import com.google.common.base.Throwables;

import eu.dnetlib.miscutils.datetime.DateUtils;

import eu.dnetlib.msro.workflows.GraphNode;

import eu.dnetlib.msro.workflows.ProcessStatus;

import eu.dnetlib.msro.workflows.WorkflowsConstants;

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

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

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

import eu.dnetlib.services.async.AsyncMethodStatus;

import eu.dnetlib.services.async.AsyncResponse;

/**

 * Created by michele on 26/11/15.

 */

@Component

public class ProcessEngine {

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

	@Autowired

	private NodeHelper nodeHelper;

	public void startProcess(final WorkflowProcess process) {

		log.info(process.getGraph());

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

		final long now = DateUtils.now();

		process.setStatus(ProcessStatus.EXECUTING);

		process.setStartDate(now);

		process.setLastActivityDate(now);

		try {

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

				final AbstractProcessNode pNode = nodeHelper.newProcessNode(node, process, process.getEnv());

				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(ProcessStatus.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())) {

				log.info("Executing node: " + node.getName());

				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 Token token = new Token(node.getName(), newNodeTokenCallback(process, node));

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

						final AbstractProcessNode pNode = nodeHelper.newProcessNode(node, process, token.getEnv());

						process.getTokens().add(token);

						process.setLastActivityDate(DateUtils.now());

						if (node.isSucessNode()) {

							markAsCompleted(process, token);

						} else {

							pNode.execute(token);

						}

					}

				} else {

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

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

					final AbstractProcessNode pNode = nodeHelper.newProcessNode(node, process, token.getEnv());

					process.getTokens().add(token);

					process.setLastActivityDate(DateUtils.now());

					pNode.execute(token);

				}

			}

		} catch (final Throwable e) {

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

			process.setStatus(ProcessStatus.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.releaseAsCompleted();

			}

		}

		final long now = token.getEndDate();

		process.setLastActivityDate(now);

		process.setEndDate(now);

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

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

		final AsyncResponse asyncResponse = new AsyncResponse();

		if (token.isFailed()) {

			asyncResponse.setStatus(AsyncMethodStatus.FAILED);

			process.setStatus(ProcessStatus.FAILURE);

			process.setError(token.getError());

			process.setErrorStacktrace(token.getErrorStackTrace());

			process.setLastActivityDate(DateUtils.now());

			process.getOutputParams().put(WorkflowsConstants.LOG_SYSTEM_ERROR, token.getError());

			process.getOutputParams().put(WorkflowsConstants.LOG_SYSTEM_ERROR_STACKTRACE, token.getErrorStackTrace());

		} else {

			asyncResponse.setStatus(AsyncMethodStatus.SUCCESS);

		}

		asyncResponse.prepareResponse(process.getOutputParams());

		process.getCallback().notify(asyncResponse);

	}

	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())));

	}

}