D-Net

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

import java.io.IOException;

import java.util.Map;

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

import eu.dnetlib.enabling.resultset.client.ResultSetClient;

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

import eu.dnetlib.miscutils.functional.xml.DnetXsltFunctions;

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

import eu.dnetlib.msro.workflows.nodes.blackboard.BlackboardWorkflowJobListener;

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

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

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

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

import eu.dnetlib.rmi.common.ResultSet;

import eu.dnetlib.rmi.data.hadoop.HadoopBlackboardActions;

import eu.dnetlib.rmi.data.hadoop.HadoopService;

import eu.dnetlib.rmi.enabling.ISLookUpException;

import eu.dnetlib.rmi.enabling.ISLookUpService;

import eu.dnetlib.rmi.manager.MSROException;

import org.apache.commons.lang3.StringUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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

/**

 * Created by claudio on 08/04/16.

 */

public abstract class AbstractHBaseJobNode extends BlackboardJobNode {

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

	private final String INPUT_HBASE_TABLE_PARAM = "hbaseTable";

	private final String INPUT_EPR_PARAM = "input_epr";

	private final String INPUT_CLUSTER_PARAM = "cluster";

	private final String XSLT_PARAM = "xslt";

	private final String OUTPUT_HBASE_TABLE_PARAM = "table";

	private final String OUTPUT_CLUSTER_PARAM = "cluster";

	private final String SIMULATION_PARAM = "simulation";

	@Autowired

	protected UniqueServiceLocator serviceLocator;

	@Autowired

	protected ResultSetClient resultSetClient;

	protected String inputEprParam;

	protected String hbaseTable;

	protected String cluster;

	protected String mapping;

	protected boolean simulation = false;

	protected abstract HadoopBlackboardActions getAction();

	@Override

	protected String obtainServiceId(final Env env) {

		return getServiceLocator().getServiceId(HadoopService.class);

	}

	@Override

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

		log.info("Invoking blackboard method: " + getAction().toString());

		final ResultSet<?> rs = token.getEnv().getAttribute(getInputEprParam(), ResultSet.class);

		token.setProgressProvider(new ResultsetProgressProvider(rs, resultSetClient));

		job.setAction(getAction().toString());

		job.getParameters().put(INPUT_EPR_PARAM, rs.toJson());

		job.getParameters().put(XSLT_PARAM, DnetXsltFunctions.encodeBase64(readXslt(getMapping())));

		job.getParameters().put(OUTPUT_HBASE_TABLE_PARAM, getHbaseTable());

		job.getParameters().put(OUTPUT_CLUSTER_PARAM, getCluster());

		job.getParameters().put(SIMULATION_PARAM, String.valueOf(isSimulation()));

	}

	@Override

	protected BlackboardWorkflowJobListener generateBlackboardListener(final Token token) {

		return new BlackboardWorkflowJobListener(token) {

			@Override

			protected void responseToEnv(final Env env, final Map<String, String> responseParams) {

				final String count = responseParams.get("count");

				log.info(String.format("%s %s objects to HBase table %s, cluster %s", getAction().toString(), count, getHbaseTable(), getCluster()));

				env.setAttribute(WorkflowsConstants.MAIN_LOG_PREFIX + getNodeName() + ":count", count);

			}

		};

	}

	private String readXslt(final String profileId) throws IOException, MSROException, ISLookUpException {

		if (StringUtils.isBlank(profileId)) throw new MSROException("missing profile id");

		log.info("loading mapping from profile id: " + profileId);

		final String xquery =

				String.format("/RESOURCE_PROFILE[.//RESOURCE_IDENTIFIER/@value='%s']/BODY/CONFIGURATION/SCRIPT/CODE/*[local-name()='stylesheet']", profileId);

		return serviceLocator.getService(ISLookUpService.class).getResourceProfileByQuery(xquery);

	}

	public String getInputEprParam() {

		return inputEprParam;

	}

	public void setInputEprParam(final String inputEprParam) {

		this.inputEprParam = inputEprParam;

	}

	public String getMapping() {

		return mapping;

	}

	public void setMapping(final String mapping) {

		this.mapping = mapping;

	}

	public String getCluster() {

		return cluster;

	}

	public void setCluster(final String cluster) {

		this.cluster = cluster;

	}

	public boolean isSimulation() {

		return simulation;

	}

	public void setSimulation(final boolean simulation) {

		this.simulation = simulation;

	}

	public String getHbaseTable() {

		return hbaseTable;

	}

	public void setHbaseTable(final String hbaseTable) {

		this.hbaseTable = hbaseTable;

	}

}