D-Net

package eu.dnetlib.msro.workflows.util;

import java.util.HashMap;

import java.util.List;

import java.util.Map;

import java.util.Optional;

import java.util.stream.Collectors;

import javax.annotation.PostConstruct;

import org.apache.commons.lang3.StringUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.springframework.beans.BeansException;

import org.springframework.beans.PropertyAccessorFactory;

import org.springframework.context.ApplicationContext;

import org.springframework.context.ApplicationContextAware;

import org.springframework.stereotype.Component;

import com.fasterxml.jackson.core.type.TypeReference;

import com.fasterxml.jackson.databind.ObjectMapper;

import eu.dnetlib.exceptions.DnetGenericRuntimeException;

import eu.dnetlib.msro.annotations.ProcessNode;

import eu.dnetlib.msro.exceptions.MSROException;

import eu.dnetlib.msro.workflows.GraphNode;

import eu.dnetlib.msro.workflows.GraphNodeParameter;

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

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

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

import eu.dnetlib.msro.workflows.procs.Env;

import eu.dnetlib.msro.workflows.procs.ProcessAware;

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

/**

 * Created by michele on 19/11/15.

 */

@Component

public class NodeHelper implements ApplicationContextAware {

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

	private ApplicationContext applicationContext;

	private Map<String, Class<?>> validTypes = new HashMap<>();

	@PostConstruct

	public void init() {

		validTypes = applicationContext.getBeansWithAnnotation(ProcessNode.class)

				.values()

				.stream()

				.filter(bean -> bean instanceof AbstractProcessNode)

				.map(bean -> (AbstractProcessNode) bean)

				.collect(Collectors.toMap(

						bean -> bean.getClass().getAnnotation(ProcessNode.class).value(),

						Object::getClass));

	}

	public AbstractProcessNode newProcessNode(final GraphNode node, final WorkflowProcess process, final Env env) throws MSROException {

		if (node.isSucessNode()) {

			return new SuccessNode();

		} else if (StringUtils.isBlank(node.getType())) {

			return new DefaultJobNode(node.getName());

		} else {

			final Optional<AbstractProcessNode> optional = applicationContext.getBeansWithAnnotation(ProcessNode.class)

					.values()

					.stream()

					.filter(bean -> bean.getClass().getAnnotation(ProcessNode.class).value().equals(node.getType()))

					.filter(bean -> bean instanceof AbstractProcessNode)

					.map(bean -> (AbstractProcessNode) bean)

					.findFirst();

			if (optional.isPresent()) {

				final AbstractProcessNode pnode = optional.get();

				pnode.setNodeName(node.getName());

				// I invoke the setter methods using the static params of the graph node

				try {

					PropertyAccessorFactory.forBeanPropertyAccess(pnode).setPropertyValues(resolveParams(node, env));

				} catch (final Throwable e) {

					throw new MSROException(String.format("error setting parameters in wfNode %s", node.getName()), e);

				}

				if (pnode instanceof ProcessAware) {

					((ProcessAware) pnode).setProcess(process);

				}

				return pnode;

			} else {

				log.error("cannot find node " + node.getType() + ", valid types are: " + validTypes);

				throw new MSROException("cannot find node " + node.getType());

			}

		}

	}

	public Map<String, Object> resolveParamsWithNoEnv(final GraphNode node) {

		return resolveParams(node, null);

	}

	public Map<String, Object> resolveParams(final GraphNode node, final Env env) {

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

		final ObjectMapper mapper = new ObjectMapper();

		try {

			if (node.getNodeParams() != null) {

				for (final Map.Entry<String, GraphNodeParameter> e : node.getNodeParams().entrySet()) {

					final String pName = e.getKey();

					final GraphNodeParameter param = e.getValue();

					if (param.isEnvParam()) {

						params.put(pName, resolveFromEnv(param, env));

					} else if (param.isMap()) {

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

						final Map<String, GraphNodeParameter> unresolved =

								mapper.readValue(param.getValue(), new TypeReference<Map<String, GraphNodeParameter>>() {});

						for (final Map.Entry<String, GraphNodeParameter> entry : unresolved.entrySet()) {

							final String k = entry.getKey();

							final Object v = entry.getValue().isEnvParam() ? resolveFromEnv(entry.getValue(), env) : entry.getValue().getValue();

							map.put(k, v);

						}

						/*

						 * inputParam.entrySet() .stream() .collect(Collectors.toMap( e1 -> e1.getKey(), e1 -> e1.getValue().isEnvParam() ?

						 * : e1getValue().getValue()));

						 */

						params.put(pName, map);

					} else if (param.isList()) {

						final List<GraphNodeParameter> list =

								mapper.readValue(param.getValue(), new TypeReference<List<GraphNodeParameter>>() {});

						params.put(pName, list

								.stream()

								.map(p -> p.isEnvParam() ? resolveFromEnv(p, env) : p.getValue())

								.collect(Collectors.toList()));

					} else {

						params.put(pName, param.getValue());

					}

				}

			}

			return params;

		} catch (final Exception e) {

			throw new DnetGenericRuntimeException(e);

		}

	}

	private Object resolveFromEnv(final GraphNodeParameter param, final Env env) {

		return env != null ? env.getAttribute(param.getValue()) : "[this value will be resolved using the runtime ENV]";

	}

	@Override

	public void setApplicationContext(final ApplicationContext applicationContext) throws BeansException {

		this.applicationContext = applicationContext;

	}

	public synchronized boolean isValidType(final String type) {

		return validTypes.containsKey(type);

	}

	public Map<String, Class<?>> getValidTypes() {

		return validTypes;

	}

}