D-Net

package eu.dnetlib.msro.workflows.nodes;

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

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

import eu.dnetlib.msro.workflows.procs.Env;

import eu.dnetlib.msro.workflows.procs.ProcessAware;

import eu.dnetlib.msro.workflows.procs.Token;

import eu.dnetlib.msro.workflows.procs.WorkflowProcess;

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

import eu.dnetlib.rmi.data.MDStoreService;

import eu.dnetlib.rmi.enabling.ISLookUpService;

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

import java.util.List;

public class MergeDLIRecord extends BlackboardJobNode implements ProcessAware, ProgressProvider {

    private final static String queryTemplate = "for $x in collection(' /db/DRIVER/RepositoryServiceResources/RepositoryServiceResourceType') where $x//RESOURCE_IDENTIFIER/@value/string()='%s' return $x//FIELD[./key='NamespacePrefix']/value/text()";

    private String mdStoreId;

    private String dsId;

    private String mongoHost;

    private String mongoDBName;

    private String sparkJobPath;

    private String sparkPath;

    private String numExecutor;

    private boolean skipJob = false;

    private String sparkApplicationName;

    private WorkflowProcess process;

    @Autowired

    private UniqueServiceLocator serviceLocator;

    private String getDatasourcePrefix() throws Exception {

        final String query = String.format(queryTemplate, dsId);

        List<String> result  = serviceLocator.getService(ISLookUpService.class).quickSearchProfile(query);

        if (result!= null && result.size() ==1 )

            return result.get(0);

        throw new Exception("Unexpected result on query "+query);

    }

    @Override

    protected String obtainServiceId(Env env) {

        return getServiceLocator().getServiceId(MDStoreService.class, getMdStoreId());

    }

    @Override

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

        if (!skipJob) {

            job.setAction("RUN_PLUGIN");

            job.getParameters().put("plugin.name", "dliMergeRecord");

            job.getParameters().put("mdStoreId", getMdStoreId());

            job.getParameters().put("mongoHost", getMongoHost());

            job.getParameters().put("nsPrefix", getDatasourcePrefix());

            job.getParameters().put("sparkPath", getSparkPath());

            job.getParameters().put("sparkJobPath", getSparkJobPath());

            job.getParameters().put("mongoDBName", getMongoDBName());

            job.getParameters().put("numExecutor", getNumExecutor());

            job.getParameters().put("sparkApplicationName", process.getId());

        }

    }

    @Override

    public void setProcess(WorkflowProcess process) {

        this.process = process;

    }

    public String getSparkJobPath() {

        return sparkJobPath;

    }

    public void setSparkJobPath(String sparkJobPath) {

        this.sparkJobPath = sparkJobPath;

    }

    public String getSparkPath() {

        return sparkPath;

    }

    public void setSparkPath(String sparkPath) {

        this.sparkPath = sparkPath;

    }

    public String getSparkApplicationName() {

        return sparkApplicationName;

    }

    public void setSparkApplicationName(String sparkApplicationName) {

        this.sparkApplicationName = sparkApplicationName;

    }

    public String getMongoHost() {

        return mongoHost;

    }

    public void setMongoHost(String mongoHost) {

        this.mongoHost = mongoHost;

    }

    public String getMdStoreId() {

        return mdStoreId;

    }

    public void setMdStoreId(String mdStoreId) {

        this.mdStoreId = mdStoreId;

    }

    public String getMongoDBName() {

        return mongoDBName;

    }

    public void setMongoDBName(String mongoDBName) {

        this.mongoDBName = mongoDBName;

    }

    public boolean isSkipJob() {

        return skipJob;

    }

    public void setSkipJob(boolean skipJob) {

        this.skipJob = skipJob;

    }

    public String getDsId() {

        return dsId;

    }

    public void setDsId(String dsId) {

        this.dsId = dsId;

    }

    public String getNumExecutor() {

        return numExecutor;

    }

    public void setNumExecutor(String numExecutor) {

        this.numExecutor = numExecutor;

    }

    @Override

    public String getProgressDescription() {

        return null;

    }

}