D-Net

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

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.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)

                    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;

    }

}