D-Net

package eu.dnetlib.msro.puma.nodes;

import java.io.BufferedReader;

import java.io.FileReader;

import javax.annotation.Resource;

import javax.xml.ws.wsaddressing.W3CEndpointReference;

import com.google.common.collect.BiMap;

import com.google.common.collect.HashBiMap;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.enabling.resultset.IterableResultSetFactory;

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

import eu.dnetlib.enabling.resultset.client.utils.EPRUtils;

import eu.dnetlib.msro.puma.objectstore.PUMAObjectStore;

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

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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

/**

 * Created by sandro on 2/3/16.

 */

public class FeedEnrichedRecordsJobNode extends SimpleJobNode {

	private static final Log log = LogFactory.getLog(FeedEnrichedRecordsJobNode.class);

	@Autowired

	private PUMAObjectStore pumaObjectStore;

	private String inputEprParam;

	private String deduplicatedRecordCSV;

	/**

	 * The result set client factory.

	 */

	@Autowired

	private ResultSetClientFactory resultSetClientFactory;

	/**

	 * The result set factory.

	 */

	@Resource(name = "iterableResultSetFactory")

	private IterableResultSetFactory resultSetFactory;

	@Override

	protected String execute(NodeToken nodeToken) throws Exception {

		BiMap<String, String> deduplicatedRecord = createDeduplicatedMap();

		final W3CEndpointReference inputEpr = new EPRUtils().getEpr(nodeToken.getEnv().getAttribute(getInputEprParam()));

		final Iterable<String> inputRecord = resultSetClientFactory.getClient(inputEpr);

		pumaObjectStore.initialize();

		pumaObjectStore.feedStore(inputRecord, deduplicatedRecord);

		return Arc.DEFAULT_ARC;

	}

	public String getInputEprParam() {

		return inputEprParam;

	}

	public void setInputEprParam(String inputEprParam) {

		this.inputEprParam = inputEprParam;

	}

	public String getDeduplicatedRecordCSV() {

		return deduplicatedRecordCSV;

	}

	public void setDeduplicatedRecordCSV(String deduplicatedRecordCSV) {

		this.deduplicatedRecordCSV = deduplicatedRecordCSV;

	}

	private BiMap<String, String> createDeduplicatedMap() throws Exception {

		BufferedReader br = null;

		String sCurrentLine = null;

		BiMap<String, String> map = HashBiMap.create();

		br = new BufferedReader(new FileReader(getDeduplicatedRecordCSV()));

		while ((sCurrentLine = br.readLine()) != null) {

			String[] data = sCurrentLine.trim().split(",");

			if (data != null && data.length == 2 && !map.containsValue(data[1]))

				map.put(data[0], data[1]);

			else

				log.error("Unexpected values on creating map");

		}

		return map;

	}

}