D-Net

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

import eu.dnetlib.msro.rmi.MSROException;

import eu.dnetlib.msro.rmi.MSRORuntimeException;

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 eu.dnetlib.dhp.message.Message;

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

import eu.dnetlib.msro.openaireplus.workflows.nodes.dhp.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 java.util.Objects;

import java.util.Optional;

public class SubmitDnetHadoopJobNode extends SubmitHadoopJobNode implements ProgressProvider, ProgressJobNode {

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

	@Autowired

	private DnetMessageManager dnetMessageManager;

	private int currentValue = 0;

	private int totalValue = 0;

	private boolean accurate = false;

	private String wfId;

	@Override

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

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

		super.prepareJob(job, token);

	}

	private void updateProgressProvider() {

		final Message mess = dnetMessageManager.findOngoingMessage(wfId);

		if (mess != null && mess.getBody() != null) {

			if (mess.getBody().containsKey(Message.CURRENT_PARAM)) {

				try {

					setCurrentValue(Integer.parseInt(mess.getBody().get(Message.CURRENT_PARAM)));

				} catch (final Throwable e) {

					log.error("Error parsing value", e);

				}

			}

			if (mess.getBody().containsKey(Message.TOTAL_PARAM)) {

				try {

					setTotalValue(Integer.parseInt(mess.getBody().get(Message.TOTAL_PARAM)));

					setAccurate(true);

				} catch (final Throwable e) {

					log.error("Error parsing value", e);

				}

			}

		}

	}

	@Override

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

		return new BlackboardWorkflowJobListener(engine, token) {

			@Override

			protected void onDone(BlackboardJob job) {

				super.onDone(getOozieWorkflowReport(job));

			}

			@Override

			protected void onFailed(final BlackboardJob job) {

				super.onFailed(getOozieWorkflowReport(job));

			}

			private BlackboardJob getOozieWorkflowReport(BlackboardJob job) {

				final Message report = dnetMessageManager.findReportMessage(wfId);

				if (Objects.isNull(report)) {

					final String msg = "cannot find report for workflow id: " + wfId;

					log.error(msg);

					throw new MSRORuntimeException(msg);

				}

				report.getBody().forEach((k, v) -> log.info(String.format("%s - %s", k, v)));

				job.getParameters().putAll(report.getBody());

				return job;

			}

		};

	}

	@Override

	public ProgressProvider getProgressProvider() {

		return this;

	}

	public String getWfId() {

		return wfId;

	}

	public void setWfId(final String wfId) {

		this.wfId = wfId;

	}

	public boolean isAccurate() {

		return accurate;

	}

	public void setAccurate(final boolean accurate) {

		this.accurate = accurate;

	}

	@Override

	public int getCurrentValue() {

		updateProgressProvider();

		return currentValue;

	}

	public void setCurrentValue(final int currentValue) {

		this.currentValue = currentValue;

	}

	@Override

	public int getTotalValue() {

		return totalValue;

	}

	public void setTotalValue(final int totalValue) {

		this.totalValue = totalValue;

	}

	@Override

	public boolean isInaccurate() {

		return !accurate;

	}

}