D-Net

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

import com.googlecode.sarasvati.Engine;

import com.googlecode.sarasvati.NodeToken;

import com.googlecode.sarasvati.env.Env;

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

import eu.dnetlib.message.Message;

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 eu.dnetlib.msro.workflows.util.WorkflowsConstants;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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

import java.util.List;

import java.util.Map;

public class SubmitDnetHadoopJobNode extends SubmitHadoopJobNode implements ProgressProvider, ProgressJobNode {

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

    @Autowired

    DnetMessageManager dnetMessageManager;

    private boolean ongoing = true;

    private int currentValue;

    private String wfId;

    @Override

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

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

        Runnable r = () -> {

            while (ongoing) {

                Message mess = dnetMessageManager.getOnGoingMessages(wfId);

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

                    try {

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

                        Thread.sleep(1000);

                    } catch (Throwable e) {

                        log.error("Error ono receiving messages ", e);

                    }

                }

            }

        };

        new Thread(r).start();

        super.prepareJob(job, token);

    }

    @Override

    protected BlackboardWorkflowJobListener generateBlackboardListener(Engine engine, 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(Env env, Map<String, String> responseParams) {

                ongoing = false;

                List<Message> reports = dnetMessageManager.getReport(wfId);

                if (reports == null) {

                    int numberOftries = 0;

                    try {

                        while (reports == null && numberOftries < 3) {

                            reports = dnetMessageManager.getReport(wfId);

                            Thread.sleep(3000 * numberOftries++);

                        }

                    } catch (InterruptedException e) {

                        log.error("Error on waiting report", e);

                    }

                }

                if (reports == null) {

                    throw new RuntimeException("Unable to get report for WorklowId " + wfId);

                }

                reports.forEach(it -> it.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;

    }

}