D-Net

package eu.dnetlib.msro.workflows.dedup;

import java.io.StringReader;

import java.util.Iterator;

import java.util.List;

import java.util.Queue;

import javax.annotation.Resource;

import javax.xml.ws.wsaddressing.W3CEndpointReference;

import org.antlr.stringtemplate.StringTemplate;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.dom4j.Document;

import org.dom4j.DocumentException;

import org.dom4j.Node;

import org.dom4j.io.SAXReader;

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

import com.google.common.collect.Lists;

import com.google.common.collect.Queues;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.data.proto.TypeProtos.Type;

import eu.dnetlib.enabling.resultset.IterableResultSetFactory;

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

import eu.dnetlib.msro.workflows.hadoop.utils.Similarity;

import eu.dnetlib.msro.workflows.hadoop.utils.SimilarityMeshBuilder;

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

public class BuildSimilarityMeshJobNode extends AsyncJobNode {

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

	/** The result set factory. */

	@Resource(name = "iterableResultSetFactory")

	private IterableResultSetFactory resultSetFactory;

	/** The result set client factory. */

	@Resource(name = "resultSetClientFactory")

	private ResultSetClientFactory resultSetClientFactory;

	private StringTemplate similarity;

	private String inputEprParam;

	private String outputEprParam;

	@Override

	protected String execute(final NodeToken token) throws Exception {

		final String inputEpr = token.getEnv().getAttribute(getInputEprParam());

		final Iterator<String> rsClient = resultSetClientFactory.getClient(inputEpr).iterator();

		final Queue<Object> queue = Queues.newLinkedBlockingQueue();

		final SAXReader reader = new SAXReader();

		if (rsClient.hasNext()) {

			populateQueue(queue, reader, rsClient.next());

		}

		final W3CEndpointReference eprOut = resultSetFactory.createIterableResultSet(new Iterable<String>() {

			@Override

			public Iterator<String> iterator() {

				return new Iterator<String>() {

					@Override

					public boolean hasNext() {

						synchronized (queue) {

							return !queue.isEmpty();

						}

					}

					@Override

					public String next() {

						synchronized (queue) {

							final Object o = queue.poll();

							while (queue.isEmpty() && rsClient.hasNext()) {

								populateQueue(queue, reader, rsClient.next());

							}

							return buildSimilarity((Similarity) o);

						}

					}

					@Override

					public void remove() {

						throw new UnsupportedOperationException();

					}

				};

			}

		});

		token.getEnv().setAttribute(getOutputEprParam(), eprOut.toString());

		return Arc.DEFAULT_ARC;

	}

	private void populateQueue(final Queue<Object> q, final SAXReader r, final String xml) {

		try {

			final Document d = r.read(new StringReader(xml));

			final String groupid = d.valueOf("//FIELD[@name='id']");

			final List<?> items = d.selectNodes("//FIELD[@name='group']/ITEM");

			final String entitytype = d.valueOf("//FIELD[@name='entitytype']");

			final List<String> group = Lists.newArrayList();

			for (final Object id : items) {

				group.add(((Node) id).getText());

			}

			// compute the full mesh

			final Type type = Type.valueOf(entitytype);

			final List<Similarity> mesh = SimilarityMeshBuilder.build(type, group);

			// total += mesh.size();

			if (log.isDebugEnabled()) {

				log.debug(String.format("built mesh for group '%s', size %d", groupid, mesh.size()));

			}

			for (final Similarity s : mesh) {

				if (log.isDebugEnabled()) {

					log.debug(String.format("adding to queue: %s", s.toString()));

				}

				q.add(s);

			}

		} catch (final DocumentException e) {

			log.error("invalid document: " + xml);

		}

	}

	private String buildSimilarity(final Similarity s) {

		final StringTemplate template = new StringTemplate(getSimilarity().getTemplate());

		template.setAttribute("source", s.getPair().getKey());

		template.setAttribute("target", s.getPair().getValue());

		template.setAttribute("type", s.getType().toString());

		final String res = template.toString();

		return res;

	}

	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;

	}

	public StringTemplate getSimilarity() {

		return similarity;

	}

	@Required

	public void setSimilarity(final StringTemplate similarity) {

		this.similarity = similarity;

	}

}