D-Net

package eu.dnetlib.msro.openaireplus.workflows.nodes.hadoop;

import java.util.Map;

import java.util.Optional;

import java.util.stream.Collectors;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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

import com.google.gson.Gson;

import com.googlecode.sarasvati.Engine;

import com.googlecode.sarasvati.NodeToken;

import com.googlecode.sarasvati.env.Env;

import eu.dnetlib.collector.worker.model.ApiDescriptor;

import eu.dnetlib.enabling.datasources.common.ApiParam;

import eu.dnetlib.enabling.datasources.common.LocalDatasourceManager;

import eu.dnetlib.enabling.tools.blackboard.BlackboardJob;

import eu.dnetlib.message.Message;

import eu.dnetlib.msro.message.DnetMessageManager;

import eu.dnetlib.msro.workflows.hadoop.SubmitHadoopJobNode;

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

import eu.dnetlib.msro.workflows.nodes.blackboard.BlackboardWorkflowJobListener;

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

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

public class SubmitDnetHadoopJobNode extends SubmitHadoopJobNode implements ProgressProvider, ProgressJobNode {

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

	@Autowired

	DnetMessageManager dnetMessageManager;

	@Autowired

	private LocalDatasourceManager<?, ?> dsManager;

	private boolean ongoing = true;

	private int currentValue;

	private String wfId;

	@Override

	protected void prepareJob(final BlackboardJob job, final NodeToken token) throws Exception {

		final String dsId = token.getEnv().getAttribute(WorkflowsConstants.DATAPROVIDER_ID);

		final String apiId = token.getEnv().getAttribute(WorkflowsConstants.DATAPROVIDER_INTERFACE);

		final String mdId = token.getEnv().getAttribute("mdId");

		final String versionId = token.getEnv().getAttribute("versionId");

		final Optional<ApiDescriptor> opt = dsManager.getApis(dsId).stream().filter(a -> a.getId().equals(apiId)).map(a -> {

			final ApiDescriptor res = new ApiDescriptor();

			res.setBaseUrl(a.getBaseurl());

			res.setId(a.getId());

			res.setProtocol(a.getProtocol());

			res.getParams().putAll(a.getApiParams().stream().map(o -> {

				return (ApiParam) o;

			}).collect(Collectors.toMap(ApiParam::getParam, ApiParam::getValue)));

			return res;

		}).findFirst();

		if (!opt.isPresent()) { return; }

		final ApiDescriptor api = opt.get();

		final String hdfsPath = "hdfs://mdstores/" + mdId + "/" + versionId;

		final String nameNode = "SubmitDnetHadoop";

		log.info("Collection param 1 : hdfsPath = " + hdfsPath);

		log.info("Collection param 2 : api descriptor json = " + new Gson().toJson(api));

		log.info("Collection param 3 : nameNode = " + nameNode);

		wfId = token.getProcess().getEnv().getAttribute("system:processId");

		job.getParameters().put("wfId", wfId);

		final Runnable runnable = () -> {

			while (ongoing) {

				final Message mess = dnetMessageManager.getOnGoingMessages(wfId);

				if (mess != null && mess.getBody() != null && mess.getBody().containsKey("progressCount")) {

					try {

						currentValue = Integer.parseInt(mess.getBody().get("progressCount"));

						Thread.sleep(1000);

					} catch (final Throwable e) {

						e.printStackTrace();

					}

				}

			}

		};

		job.getParameters().put("hdfsPath", hdfsPath);

		job.getParameters().put("nameNode", nameNode);

		job.getParameters().put("api", new Gson().toJson(api));

		new Thread(runnable).start();

		super.prepareJob(job, token);

	}

	@Override

	protected BlackboardWorkflowJobListener generateBlackboardListener(final Engine engine, final NodeToken token) {

		return new BlackboardWorkflowJobListener(engine, token) {

			@Override

			protected void onFailed(final BlackboardJob job) {

				ongoing = false;

				log.warn("Blackboard workflow node FAILED: " + job.getError());

				token.getEnv().setAttribute(WorkflowsConstants.SYSTEM_HAS_FAILED, true);

				token.getEnv().setAttribute(WorkflowsConstants.SYSTEM_ERROR, job.getError());

				complete(job, "abort");

			}

			@Override

			protected void populateEnv(final Env env, final Map<String, String> responseParams) {

				ongoing = false;

				final Message report = dnetMessageManager.getReport(wfId, "Collection");

				if (report != null) {

					report.getBody().forEach(env::setAttribute);

				}

			}

		};

	}

	@Override

	public int getTotalValue() {

		return 0;

	}

	@Override

	public int getCurrentValue() {

		return currentValue;

	}

	@Override

	public boolean isInaccurate() {

		return false;

	}

	@Override

	public ProgressProvider getProgressProvider() {

		return this;

	}

}