D-Net

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

import java.util.Map;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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

import com.googlecode.sarasvati.Engine;

import com.googlecode.sarasvati.NodeToken;

import com.googlecode.sarasvati.env.Env;

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 TransformHadoopJobNode extends SubmitHadoopJobNode implements ProgressProvider, ProgressJobNode {

	private static final Log log = LogFactory.getLog(TransformHadoopJobNode.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 {

		// param 1 : hdfs path from

		// param 2 : hdfs path to

		// param 3 : transformation rule

		// param 4 : nameNode

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

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

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

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

		final String readingHdfsPath = "hdfs://mdstores/" + reading_mdId + "/" + reading_versionId;

		final String writingHdfsPath = "hdfs://mdstores/" + writing_mdId + "/" + writing_versionId;

		final String rule = "XXXXXXXX";

		final String nameNode = "TransformHadoop";

		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("readingHdfsPath", readingHdfsPath);

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

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

		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;

	}

}