D-Net

package eu.dnetlib.msro.controllers;

import java.io.IOException;

import java.util.Arrays;

import java.util.HashMap;

import java.util.List;

import java.util.Map;

import java.util.Set;

import java.util.stream.Collectors;

import org.apache.commons.lang3.StringUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.dom4j.DocumentHelper;

import org.dom4j.Element;

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

import org.springframework.web.bind.annotation.PathVariable;

import org.springframework.web.bind.annotation.RequestMapping;

import org.springframework.web.bind.annotation.RequestMethod;

import org.springframework.web.bind.annotation.RestController;

import com.fasterxml.jackson.databind.ObjectMapper;

import eu.dnetlib.clients.msro.MsroWorkerRunningInstance;

import eu.dnetlib.clients.msro.TerminatedWfDesc;

import eu.dnetlib.enabling.annotations.DnetService;

import eu.dnetlib.enabling.annotations.DnetServiceType;

import eu.dnetlib.msro.exceptions.MSROException;

import eu.dnetlib.msro.workers.aggregation.collect.plugins.CollectorPlugin;

import eu.dnetlib.msro.workers.aggregation.collect.plugins.DnetCollectorParam;

import eu.dnetlib.msro.workers.aggregation.collect.plugins.DnetCollectorPlugin;

import eu.dnetlib.msro.workers.aggregation.collect.plugins.ProtocolParameterType;

import eu.dnetlib.msro.workflows.ProcessStatus;

import eu.dnetlib.msro.workflows.WorkflowInstance;

import eu.dnetlib.msro.workflows.WorkflowProcessInfo;

import eu.dnetlib.msro.workflows.nodes.collect.CollectorPluginEnumerator;

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

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

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

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

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

import eu.dnetlib.services.BaseService;

import eu.dnetlib.services.ServiceRunningInstance;

import eu.dnetlib.services.ServiceStatus;

import eu.dnetlib.services.async.AsyncMethodException;

import eu.dnetlib.services.async.AsyncRunnable;

import eu.dnetlib.services.async.AsyncServerCallback;

import eu.dnetlib.services.async.HasAsyncMethods;

@RestController

@RequestMapping("/worker")

@DnetService(DnetServiceType.msroWorker)

public class MsroWorkerController extends BaseService implements HasAsyncMethods {

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

	private static final long MAX_NUMBER_OF_TERMINATED_WFS = 5;

	@Autowired

	private ProcessFactory processFactory;

	@Autowired

	private ProcessRegistry processRegistry;

	@Autowired

	private ProcessExecutor processExecutor;

	@Autowired

	private NodeHelper nodeHelper;

	@Autowired

	private CollectorPluginEnumerator collectorPlugins;

	@RequestMapping(value = "nodes", method = RequestMethod.GET)

	public Set<String> nodes() {

		return nodeHelper.getValidTypes().keySet();

	}

	@RequestMapping(value = "collector/plugins", method = RequestMethod.GET)

	public List<String> plugins() {

		return collectorPlugins.getAll().stream().map(CollectorPlugin::getProtocol).sorted().collect(Collectors.toList());

	}

	@RequestMapping(value = "pause", method = RequestMethod.GET)

	public ServiceRunningInstance pause() {

		processExecutor.setPaused(true);

		return getStatus();

	}

	@RequestMapping(value = "resume", method = RequestMethod.GET)

	public ServiceRunningInstance resume() {

		processExecutor.setPaused(false);

		return getStatus();

	}

	@RequestMapping(value = "proc/{procId}", method = RequestMethod.GET)

	public WorkflowProcessInfo processInfo(@PathVariable final String procId) {

		return processRegistry.findProcess(procId);

	}

	@RequestMapping(value = "kill/{procId}", method = RequestMethod.GET)

	public boolean killProcess(@PathVariable final String procId) {

		processRegistry.findProcess(procId).kill();

		return true;

	}

	@Override

	public List<Element> geXmlProfileSections() {

		final Element nodes = DocumentHelper.createElement("WORKFLOW_NODES");

		final Element collPlugins = DocumentHelper.createElement("COLLECTOR_PLUGINS");

		nodeHelper.getValidTypes().entrySet().forEach(e -> {

			final Element n = nodes.addElement("NODE");

			n.addAttribute("type", e.getKey());

			n.addAttribute("class", e.getValue().getName());

		});

		collectorPlugins.getAll().stream()

				.map(o -> o.getClass().getAnnotation(DnetCollectorPlugin.class))

				.forEach(ann -> {

					final Element plugin = collPlugins.addElement("PLUGIN");

					plugin.addAttribute("protocol", ann.value());

					for (final DnetCollectorParam ap : ann.parameters()) {

						final Element p = plugin.addElement("PARAM");

						p.addAttribute("name", ap.value());

						p.addAttribute("type", ap.type().toString());

						p.addAttribute("optional", Boolean.toString(ap.optional()));

						if (StringUtils.isNotBlank(ap.regex()) && (ap.type() == ProtocolParameterType.TEXT)) {

							p.addAttribute("regex", ap.regex());

						}

					}

				});

		return Arrays.asList(nodes, collPlugins);

	}

	@Override

	public ServiceRunningInstance getStatus() {

		return MsroWorkerRunningInstance.newInstance(

				getProfileId(),

				getBaseUrl(),

				baseDir(), processExecutor.isPaused() ? ServiceStatus.PAUSED : ServiceStatus.ACTIVE,

				processRegistry.countRunningWfs(),

				processRegistry.countQueuedWfs(),

				processRegistry.listTerminated()

						.stream()

						.limit(MAX_NUMBER_OF_TERMINATED_WFS)

						.map(wf -> new TerminatedWfDesc(wf.getId(), wf.getStatus() == ProcessStatus.SUCCESS, wf.getEndDate()))

						.collect(Collectors.toList()));

	}

	@Override

	public AsyncRunnable createAsyncJob(final String method, final String jsonParams, final AsyncServerCallback callback) throws AsyncMethodException {

		final ObjectMapper mapper = new ObjectMapper();

		try {

			switch (method) {

			case "startWorkflow":

				return new AsyncRunnable() {

					private WorkflowProcess proc;

					@Override

					public Map<String, String> prepare() throws AsyncMethodException {

						try {

							proc = processFactory.newProcess(getBaseUrl(), mapper.readValue(jsonParams, WorkflowInstance.class), callback);

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

							map.put("procId", proc.getId());

							return map;

						} catch (final MSROException | IOException e) {

							throw new AsyncMethodException(e);

						}

					}

					@Override

					public void execute() throws AsyncMethodException {

						try {

							processExecutor.execute(proc);

						} catch (final MSROException e) {

							throw new AsyncMethodException(e);

						}

					}

				};

			default:

				log.warn("Invalid method: " + method);

				throw new AsyncMethodException("Invalid method: " + method);

			}

		} catch (final Throwable e) {

			log.warn("Error executing method: " + method, e);

			throw new AsyncMethodException("Error executing method: " + method, e);

		}

	}

}