D-Net

package eu.dnetlib.msro.workflows.nodes.validation;

import javax.annotation.Resource;

import eu.dnetlib.enabling.datasources.common.Api;

import eu.dnetlib.enabling.datasources.common.Datasource;

import eu.dnetlib.enabling.datasources.common.DsmException;

import eu.dnetlib.enabling.datasources.common.LocalDatasourceManager;

import org.apache.commons.lang.StringUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import com.googlecode.sarasvati.Engine;

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.data.mdstore.MDStoreService;

import eu.dnetlib.data.mdstore.MDStoreServiceException;

import eu.dnetlib.enabling.is.lookup.rmi.ISLookUpService;

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

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

import eu.dnetlib.miscutils.datetime.DateUtils;

import eu.dnetlib.msro.workflows.nodes.BlackboardJobNode;

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.springframework.beans.factory.annotation.Autowired;

/**

 * This node send the BB message to the ValidatorService to validate records from an mdstore.

 * <p>

 * See ticket https://issue.openaire.research-infrastructures.eu/issues/538 for specs.

 * </p>

 *

 * @author andrea.mannocci, alessia

 *

 */

public class ValidatorServiceBlackboardJobNode extends BlackboardJobNode implements ProgressJobNode, ProgressProvider {

	public static final Log log = LogFactory.getLog(ValidatorServiceBlackboardJobNode.class); // NOPMD by marko on 11/24/08 5:02 PM

	public static final String VALIDATE_ACTION = "VALIDATE";

	public static final String VALIDATION_TYPE = "DNET";

	public static final String VALIDATOR_EPR_PARAM_NAME = "outputResultSetEpr";

	private String xQueryForValidatorServiceId;

	private String outputEprName = VALIDATOR_EPR_PARAM_NAME;

	// VALIDATOR JOB PARAM VALUES

	private String mdstoreId;

	private String guidelines;

	private String blacklistRuleName;

	private String recordsSampleSize;

	/** Json key-value map of other parameters that will be stored in the validator back-end as they are. **/

	private String extraParams;

	/** True to ask the validator to generate the output EPR. **/

	private boolean shouldOutputRecords = false;

	/** True to ask the validator to update the blacklist. **/

	private boolean shouldUpdateBlacklist = true;

	@Resource

	private UniqueServiceLocator serviceLocator;

	@Autowired

	private LocalDatasourceManager<Datasource<?, ?>, Api<?>> dsManager;

	/** Number of records after which the Validator must update its progress status in currentlyValidatedParam **/

	private int jobStatusUpdateInterval = 1000;

	/** Parameter where the validator tells how many records have been processed sofar **/

	private String currentlyValidatedParam = "recordsTested";

	/** Number of records validated so far. **/

	private int currentlyValidated = 0;

	private int total = 0;

	private boolean accurate;

	/**

	 * <p>

	 * Parameters to set for the VALIDATE BB job:

	 *

	 * <pre>

	 *     <PARAMETER name="type" value="DNET"/>

	 *     <PARAMETER name="guidelines" value="openaire3.0"/> //values: openaire3.0, openaire2.0, openaire2.0_data, driver

	 *     <PARAMETER name="records" value="#records"/> //OPTIONAL

	 *     <PARAMETER name="blacklistedRecords" value="true"/> //enables the feature to keep blacklisted records. Default is true.

	 *     <PARAMETER name="blacklistGuidelines" value="blacklist_ruleset"/> //optional. values: name of the ruleset (as created from the validator admin panel) to use to generate the blacklist. Default value is the value in "guidelines".

	 *     <PARAMETER name="datasourceID" value=""/>

	 *     <PARAMETER name="datasourceName" value=""/> //OPTIONAL

	 *     <PARAMETER name="datasourceNamespacePrefix" value=""/> //OPTIONAL

	 *     <PARAMETER name="interfaceId" value=" "/>

	 *     <PARAMETER name="baseUrl" value=""/> //OPTIONAL

	 *     <PARAMETER name="mdstoreId" value=" "/>

	 *     <PARAMETER name="outputEpr" value="boolean"/> //TRUE to get the EPR of validated records. Default is false.

	 *     <PARAMETER name="extraParams" value="json key-value map"/> //OPTIONAL

	 *     <PARAMETER name="submissionDate" value="YYYY-MM-DD HH24:MI:SS"/>

	 * </pre>

	 *

	 * </p>

	 * {@inheritDoc}

	 *

	 * @see eu.dnetlib.msro.workflows.nodes.BlackboardJobNode#prepareJob(eu.dnetlib.enabling.tools.blackboard.BlackboardJob,

	 *      com.googlecode.sarasvati.NodeToken)

	 */

	@Override

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

		job.setAction(VALIDATE_ACTION);

		job.getParameters().put("type", VALIDATION_TYPE);

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

		job.getParameters().put("records", getRecordsSampleSize());

		// params for blacklist

		job.getParameters().put("blacklistedRecords", isShouldUpdateBlacklist() + "");

		if (isShouldUpdateBlacklist()) {

			job.getParameters().put("blacklistGuidelines", blacklistRuleName);

		}

		// datasource params

		final String datasourceId = token.getFullEnv().getAttribute(WorkflowsConstants.DATAPROVIDER_ORIGINALID);

		final String api = token.getFullEnv().getAttribute(WorkflowsConstants.DATAPROVIDER_INTERFACE);

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

		job.getParameters().put("datasourceName", token.getFullEnv().getAttribute(WorkflowsConstants.DATAPROVIDER_NAME));

		job.getParameters().put("datasourceNamespacePrefix", token.getFullEnv().getAttribute(WorkflowsConstants.DATAPROVIDER_NAMESPACE_PREFIX));

		job.getParameters().put("interfaceId", api);

		job.getParameters().put("baseUrl", token.getFullEnv().getAttribute(WorkflowsConstants.DATAPROVIDER_INTERFACE_BASEURL));

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

		// output params

		job.getParameters().put("outputEpr", isShouldOutputRecords() + "");

		// extra params

		if (StringUtils.isNotBlank(extraParams)) {

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

		}

		// Setting properties for the progress bar

		if (StringUtils.isBlank(getRecordsSampleSize()) || getRecordsSampleSize().equalsIgnoreCase("ALL")) {

			total = getTotal();

		} else {

			total = Integer.getInteger(getRecordsSampleSize());

		}

		if (jobStatusUpdateInterval > 0) {

			job.getParameters().put("jobStatusUpdateInterval", "" + jobStatusUpdateInterval);

		}

		log.debug("*****Launching validation with the following parameters:****");

		log.debug("Datasource id: " + datasourceId);

		log.debug("API: " + api);

		log.debug("Total records: " + total);

		log.debug("Number of records to validate: " + getRecordsSampleSize());

		log.debug("jobStatusUpdateInterval (set only if > 0, wf default is 1000, validator's default is 100): " + jobStatusUpdateInterval);

		job.getParameters().put("submissionDate", DateUtils.now_ISO8601());

	}

	private int getTotal() {

		int size = -1;

		accurate = false;

		try {

			if (StringUtils.isBlank(getRecordsSampleSize()) || getRecordsSampleSize().equalsIgnoreCase("ALL")) {

				size = serviceLocator.getService(MDStoreService.class).size(mdstoreId);

				accurate = true;

			} else {

				total = Integer.parseInt(getRecordsSampleSize());

				accurate = true;

			}

		} catch (final MDStoreServiceException e) {

			log.warn("Can't get size of mdstore " + mdstoreId + " progress will be INACCURATE");

			e.printStackTrace();

		} catch (final NumberFormatException e) {

			log.warn("The parameter recordSampleSize is invalid: progress will be INACCURATE");

			e.printStackTrace();

		}

		return size;

	}

	/**

	 * <p>

	 * Parameters in the VALIDATE BB job response:

	 *

	 * <pre>

	 * <PARAMETER name="error" value=""/>

	 * <PARAMETER name="outputResultSetEpr" value=""/> //only if isShouldOutputRecords() == true

	 * <PARAMETER name="jobId" value=""/>

	 * </pre>

	 *

	 * </p>

	 *

	 *

	 * {@inheritDoc}

	 *

	 * @see eu.dnetlib.msro.workflows.nodes.BlackboardJobNode#generateBlackboardListener(com.googlecode.sarasvati.Engine,

	 *      com.googlecode.sarasvati.NodeToken)

	 */

	@Override

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

		return new BlackboardWorkflowJobListener(engine, token) {

			boolean done = false;

			@Override

			protected void onDone(final BlackboardJob job) {

				if (!isShouldOutputRecords()) {

					log.debug("If you see this log you are not asking to get the validated EPR...");

					final String jobId = job.getParameters().get("jobId");

					token.getEnv().setAttribute("validatorJobId", jobId);

					final String error = job.getParameters().get("error");

					if (StringUtils.isNotBlank(error)) {

						token.getEnv().setAttribute("validatorError", error);

					}

					try {

						final String dsId = token.getFullEnv().getAttribute(WorkflowsConstants.DATAPROVIDER_ORIGINALID);

						final String ifaceId = token.getFullEnv().getAttribute(WorkflowsConstants.DATAPROVIDER_INTERFACE);

						dsManager.setLastValidationJob(dsId, ifaceId, jobId);

					} catch (final DsmException e) {

						log.error("Error setting last_validation_job field");

						throw new RuntimeException("Error setting last_validation_job field", e);

					}

					super.onDone(job);

				}

			}

			@Override

			protected void onOngoing(final BlackboardJob job) {

				currentlyValidated = Integer.parseInt(job.getParameters().get(currentlyValidatedParam));

				if (isShouldOutputRecords()) {

					onGoingWithEPRResults(job);

				} else {

					onGoingWithoutEPRResults(job);

				}

			}

			private void onGoingWithEPRResults(final BlackboardJob job) {

				if (!done) {

					final String epr = job.getParameters().get(VALIDATOR_EPR_PARAM_NAME);

					log.fatal(VALIDATOR_EPR_PARAM_NAME + "= " + epr);

					token.getEnv().setAttribute(outputEprName, epr);

					done = true;

					//super.onDone(job);

				}

			}

			private void onGoingWithoutEPRResults(final BlackboardJob job) {

				super.onOngoing(job);

			}

		};

	}

	@Override

	protected String obtainServiceId(final NodeToken token) {

		try {

			return getServiceLocator().getService(ISLookUpService.class).getResourceProfileByQuery(xQueryForValidatorServiceId);

		} catch (final Exception e) {

			throw new RuntimeException(e);

		}

	}

	@Override

	public ProgressProvider getProgressProvider() {

		return this;

	}

	@Override

	public int getTotalValue() {

		return total;

	}

	@Override

	public int getCurrentValue() {

		return currentlyValidated;

	}

	@Override

	public boolean isInaccurate() {

		return !accurate;

	}

	public String getGuidelines() {

		return guidelines;

	}

	public void setGuidelines(final String guidelines) {

		this.guidelines = guidelines;

	}

	public String getRecordsSampleSize() {

		return recordsSampleSize;

	}

	public void setRecordsSampleSize(final String recordsSampleSize) {

		this.recordsSampleSize = recordsSampleSize;

	}

	public String getMdstoreId() {

		return mdstoreId;

	}

	public void setMdstoreId(final String mdstoreId) {

		this.mdstoreId = mdstoreId;

	}

	public String getExtra_param() {

		return extraParams;

	}

	public void setExtra_param(final String extra_param) {

		this.extraParams = extra_param;

	}

	public String getxQueryForValidatorServiceId() {

		return xQueryForValidatorServiceId;

	}

	public void setxQueryForValidatorServiceId(final String xQueryForValidatorServiceId) {

		this.xQueryForValidatorServiceId = xQueryForValidatorServiceId;

	}

	public String getOutputEprName() {

		return outputEprName;

	}

	public void setOutputEprName(final String outputEprName) {

		this.outputEprName = outputEprName;

	}

	public String getExtraParams() {

		return extraParams;

	}

	public void setExtraParams(final String extraParams) {

		this.extraParams = extraParams;

	}

	public String getCurrentlyValidatedParam() {

		return currentlyValidatedParam;

	}

	public void setCurrentlyValidatedParam(final String currentlyValidatedParam) {

		this.currentlyValidatedParam = currentlyValidatedParam;

	}

	public int getJobStatusUpdateInterval() {

		return jobStatusUpdateInterval;

	}

	public void setJobStatusUpdateInterval(final int jobStatusUpdateInterval) {

		this.jobStatusUpdateInterval = jobStatusUpdateInterval;

	}

	public int getCurrentlyValidated() {

		return currentlyValidated;

	}

	public void setCurrentlyValidated(final int currentlyValidated) {

		this.currentlyValidated = currentlyValidated;

	}

	public boolean isShouldOutputRecords() {

		return shouldOutputRecords;

	}

	public void setShouldOutputRecords(final boolean shouldOutputRecords) {

		this.shouldOutputRecords = shouldOutputRecords;

	}

	public boolean isShouldUpdateBlacklist() {

		return shouldUpdateBlacklist;

	}

	public void setShouldUpdateBlacklist(final boolean shouldUpdateBlacklist) {

		this.shouldUpdateBlacklist = shouldUpdateBlacklist;

	}

	public String getBlacklistRuleName() {

		return blacklistRuleName;

	}

	public void setBlacklistRuleName(final String blacklistRuleName) {

		this.blacklistRuleName = blacklistRuleName;

	}

}