D-Net

package eu.dnetlib.msro.workflows.nodes;

import java.util.function.Function;

import eu.dnetlib.enabling.resultset.factory.ResultSetFactory;

import eu.dnetlib.msro.workflows.graph.Arc;

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

import eu.dnetlib.dli.resolver.model.DLIResolvedObject;

import eu.dnetlib.pid.resolver.parser.AbstractResolverParser;

import eu.dnetlib.resolver.parser.DMFResolverParser;

import eu.dnetlib.resolver.parser.PMFResolverParser;

import eu.dnetlib.resolver.parser.ScholixResolverParser;

import eu.dnetlib.rmi.common.ResultSet;

import eu.dnetlib.rmi.manager.MSROException;

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

/**

 * Created by sandro on 10/10/16.

 */

public class FilterDLIRecordJobNode extends SimpleJobNode {

	private final DMFResolverParser dmfResolverParser = new DMFResolverParser();

	private final PMFResolverParser pmfResolverParser = new PMFResolverParser();

	private final ScholixResolverParser scholixResolverParser = new ScholixResolverParser();

	private String type;

	private String inputEprParam;

	private String outputEprParam;

	@Autowired

	private ResultSetFactory resultSetFactory;

	@Override

	protected String execute(final Env env) throws Exception {

		final ResultSet<?> rsIn = env.getAttribute(this.inputEprParam, ResultSet.class);

		if ((rsIn == null)) { throw new MSROException("InputEprParam (" + this.inputEprParam + ") not found in ENV"); }

		ResultSet<String> outputResultSet;

		switch (type) {

		case "publication":

			outputResultSet = this.resultSetFactory.map(rsIn, String.class, createMappingFuncion(pmfResolverParser));

			break;

		case "dataset":

			outputResultSet = this.resultSetFactory.map(rsIn, String.class, createMappingFuncion(dmfResolverParser));

			break;

		case "unknown":

			outputResultSet = this.resultSetFactory.map(rsIn, String.class, createMappingFuncion(scholixResolverParser));

			break;

		default:

			throw new MSROException("unexpectd value of type");

		}

		env.setAttribute(this.outputEprParam, outputResultSet);

		return Arc.DEFAULT_ARC;

	}

	private Function<String, String> createMappingFuncion(AbstractResolverParser parser) {

		return it -> {

			final DLIResolvedObject resolvedObject = (DLIResolvedObject) parser.parseObject(it);

			if (resolvedObject != null) return it;

			else return "<JUNK/>";

		};

	}

	public String getType() {

		return type;

	}

	public void setType(final String type) {

		this.type = type;

	}

	public String getInputEprParam() {

		return inputEprParam;

	}

	public void setInputEprParam(final String inputEprParam) {

		this.inputEprParam = inputEprParam;

	}

	public String getOutputEprParam() {

		return outputEprParam;

	}

	public void setOutputEprParam(final String outputEprParam) {

		this.outputEprParam = outputEprParam;

	}

}