D-Net

{

	"type_source": "SVN", 

	"goal": "package -U source:jar", 

	"url": "http://svn-public.driver.research-infrastructures.eu/driver/dnet40/modules/dnet-deduplication/trunk", 

	"deploy_repository": "dnet4-snapshots", 

	"version": "4", 

	"mail": "alessia.bardi@isti.cnr.it", 

	"deploy_repository_url": "http://maven.research-infrastructures.eu/nexus/content/repositories/dnet4-snapshots", 

	"name": "dnet-deduplication"

}

package eu.dnetlib.msro.workflows.dedup.conf;

import static org.junit.Assert.assertNotNull;

import static org.junit.Assert.assertTrue;

import java.io.IOException;

import java.util.Queue;

import org.junit.Before;

import org.junit.Test;

import com.google.common.collect.Lists;

import eu.dnetlib.pace.config.DedupConfig;

public class DedupConfigurationOrchestrationTest {

	public DedupConfigurationOrchestration dco;

	@Before

	public void setUp() throws IOException {

		final Entity e = new Entity("result", "50", "Publication");

		final String actionSetId = "001";

		final Queue<DedupConfig> configurations = Lists.newLinkedList();

		configurations.add(DedupConfig.loadDefault());

		dco = new DedupConfigurationOrchestration(e, actionSetId, configurations);

		assertNotNull(dco);

		assertNotNull(dco.getActionSetId());

		assertNotNull(dco.getEntity());

		assertNotNull(dco.getConfigurations());

	}

	@Test

	public void testSerialization() {

		final String json = dco.toString();

		final DedupConfigurationOrchestration anotherDco = DedupConfigurationOrchestration.fromJSON(json);

		assertNotNull(anotherDco);

		assertTrue(json.equals(anotherDco.toString()));

	}

}

package eu.dnetlib.msro.workflows.dedup;

import java.util.List;

import org.junit.Before;

import org.junit.Test;

import com.google.common.collect.Lists;

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

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

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

public class SimilarityMeshBuilderTest {

	private List<String> list;

	@Before

	public void setUp() throws Exception {

		list = Lists.newArrayList();

		for (int i = 0; i < 10; i++) {

			list.add(i + "");

		}

	}

	@Test

	public void test() {

		final List<Similarity> combinations = SimilarityMeshBuilder.build(Type.result, list);

		System.out.println(combinations);

		System.out.println(combinations.size());

	}

}

package eu.dnetlib.msro.workflows.actions;

import java.util.List;

import java.util.Map;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import com.google.common.collect.Lists;

import com.google.common.collect.Maps;

import com.google.gson.Gson;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.actionmanager.set.RawSet;

import eu.dnetlib.miscutils.datetime.DateUtils;

import eu.dnetlib.msro.workflows.dedup.conf.DedupConfigurationOrchestration;

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

/**

 * The Class PrepareConfiguredActionSetJobNode.

 */

public class PrepareConfiguredActionSetJobNode extends SimpleJobNode {

	/**

	 * logger.

	 */

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

	/** The dedup config sequence param. */

	private String dedupConfigSequenceParam;

	/** The job property. */

	private String jobProperty;

	/*

	 * (non-Javadoc)

	 *

	 * @see eu.dnetlib.msro.workflows.nodes.SimpleJobNode#execute(com.googlecode.sarasvati.NodeToken)

	 */

	@Override

	protected String execute(final NodeToken token) throws Exception {

		final List<Map<String, String>> setList = Lists.newArrayList();

		final Map<String, String> set = Maps.newHashMap();

		set.put("rawset", RawSet.newInstance().getId());

		set.put("creationDate", DateUtils.now_ISO8601());

		set.put("set", getActionSetId(token));

		set.put("enabled", "true");

		set.put("jobProperty", getJobProperty());

		token.getEnv().setAttribute(set.get("jobProperty"), set.get("rawset"));

		setList.add(set);

		final String sets = new Gson().toJson(setList);

		log.debug("built set: " + sets);

		token.getEnv().setAttribute("sets", sets);

		return Arc.DEFAULT_ARC;

	}

	/**

	 * Gets the action set id.

	 *

	 * @param token

	 *            the token

	 * @return the action set id

	 */

	private String getActionSetId(final NodeToken token) {

		final String json = token.getEnv().getAttribute(getDedupConfigSequenceParam());

		final DedupConfigurationOrchestration dco = DedupConfigurationOrchestration.fromJSON(json);

		final String actionSetId = dco.getActionSetId();

		log.info("found actionSetId in workflow env: " + actionSetId);

		return actionSetId;

	}

	/**

	 * Gets the dedup config sequence param.

	 *

	 * @return the dedup config sequence param

	 */

	public String getDedupConfigSequenceParam() {

		return dedupConfigSequenceParam;

	}

	/**

	 * Sets the dedup config sequence param.

	 *

	 * @param dedupConfigSequenceParam

	 *            the new dedup config sequence param

	 */

	public void setDedupConfigSequenceParam(final String dedupConfigSequenceParam) {

		this.dedupConfigSequenceParam = dedupConfigSequenceParam;

	}

	/**

	 * Gets the job property.

	 *

	 * @return the job property

	 */

	public String getJobProperty() {

		return jobProperty;

	}

	/**

	 * Sets the job property.

	 *

	 * @param jobProperty

	 *            the new job property

	 */

	public void setJobProperty(final String jobProperty) {

		this.jobProperty = jobProperty;

	}

}

package eu.dnetlib.msro.workflows.actions;

import javax.annotation.Resource;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.enabling.is.registry.rmi.ISRegistryService;

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

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

public class CleanActionSetsProfileJobNode extends SimpleJobNode {

	/**

	 * logger.

	 */

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

	@Resource

	private UniqueServiceLocator serviceLocator;

	private String xupdate;

	@Override

	protected String execute(final NodeToken token) throws Exception {

		log.info("updating Action Sets profiles: " + getXupdate());

		serviceLocator.getService(ISRegistryService.class).executeXUpdate(getXupdate());

		return Arc.DEFAULT_ARC;

	}

	public String getXupdate() {

		return xupdate;

	}

	@Required

	public void setXupdate(final String xupdate) {

		this.xupdate = xupdate;

	}

}

package eu.dnetlib.msro.workflows.actions;

import java.io.StringReader;

import java.util.List;

import java.util.Map;

import javax.annotation.Resource;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.dom4j.Attribute;

import org.dom4j.Document;

import org.dom4j.Element;

import org.dom4j.io.SAXReader;

import com.google.gson.Gson;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

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

import eu.dnetlib.enabling.is.registry.rmi.ISRegistryService;

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

import eu.dnetlib.miscutils.datetime.DateUtils;

import eu.dnetlib.msro.rmi.MSROException;

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

public class UpdateSetsJobNode extends SimpleJobNode {

	/**

	 * logger.

	 */

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

	@Resource

	private UniqueServiceLocator serviceLocator;

	@Override

	protected String execute(final NodeToken token) throws Exception {

		@SuppressWarnings("unchecked")

		final List<Map<String, String>> sets = new Gson().fromJson(token.getEnv().getAttribute("sets"), List.class);

		final String lastUpdate = DateUtils.now_ISO8601();

		for (Map<String, String> set : sets) {

			// update only the enabled sets.

			if (isEnabled(set)) {

				log.info("updating set: " + set.toString());

				addLatestRawSet(set, lastUpdate);

			} else {

				log.info("skip set update: " + set.toString());

			}

		}

		return Arc.DEFAULT_ARC;

	}

	private boolean isEnabled(final Map<String, String> set) {

		return set.containsKey("enabled") && set.get("enabled").equals("true");

	}

	public void addLatestRawSet(final Map<String, String> set, final String lastUpdate) throws MSROException {

		final String q = "for $x in collection('/db/DRIVER/ActionManagerSetDSResources/ActionManagerSetDSResourceType') where $x//SET/@id = '" + set.get("set")

				+ "' return $x";

		try {

			final String profile = serviceLocator.getService(ISLookUpService.class).getResourceProfileByQuery(q);

			final Document doc = new SAXReader().read(new StringReader(profile));

			final String profId = doc.valueOf("//RESOURCE_IDENTIFIER/@value");

			final Element latest = (Element) doc.selectSingleNode("//RAW_SETS/LATEST");

			final Element expired = ((Element) doc.selectSingleNode("//RAW_SETS")).addElement("EXPIRED");

			for (Object o : latest.attributes()) {

				final Attribute a = (Attribute) o;

				expired.addAttribute(a.getName(), a.getValue());

			}

			latest.addAttribute("id", set.get("rawset"));

			latest.addAttribute("creationDate", set.get("creationDate"));

			latest.addAttribute("lastUpdate", lastUpdate);

			serviceLocator.getService(ISRegistryService.class).updateProfile(profId, doc.asXML(), "ActionManagerSetDSResourceType");

		} catch (Exception e) {

			String msg = "Error updating profile of set: " + set;

			log.error(msg);

			throw new MSROException(msg, e);

		}

	}

}

package eu.dnetlib.msro.workflows.actions;

import org.apache.commons.lang.StringUtils;

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.actionmanager.rmi.ActionManagerService;

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

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

public class PromoteActionsJobNode extends BlackboardJobNode {

	public static final String ALL_SETS = "ALL SETS";

	private String set;

	@Override

	protected String obtainServiceId(final NodeToken token) {

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

	}

	@Override

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

		job.setAction("PROMOTE");

		if (!StringUtils.isBlank(getSet()) && !getSet().equals(ALL_SETS)) {

			job.getParameters().put("set", getSet());

			token.getEnv().setAttribute("set", getSet());

		}

	}

	public String getSet() {

		return set;

	}

	public void setSet(final String set) {

		this.set = set;

	}

}

package eu.dnetlib.msro.workflows.actions;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.actionmanager.rmi.ActionManagerService;

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

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

public class GarbageSetsJobNode extends BlackboardJobNode {

	/**

	 * logger.

	 */

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

	@Override

	protected String obtainServiceId(final NodeToken token) {

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

	}

	@Override

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

		log.info("preparing garbage collection for Action sets Job");

		job.setAction("GARBAGE");

	}

}

package eu.dnetlib.msro.workflows.actions;

import java.util.List;

import java.util.Map;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import com.google.common.base.Predicate;

import com.google.common.collect.Maps;

import com.google.gson.Gson;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.actionmanager.set.RawSet;

import eu.dnetlib.miscutils.datetime.DateUtils;

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

public class PrepareActionSetsJobNode extends SimpleJobNode {

	/**

	 * logger.

	 */

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

	private String sets;

	@Override

	protected String execute(final NodeToken token) throws Exception {

		final List<Map<String, String>> setList = getSetList();

		final String now = DateUtils.now_ISO8601();

		for (Map<String, String> set : setList) {

			set.put("rawset", RawSet.newInstance().getId());

			set.put("creationDate", now);

			if (set.get("enabled").equals("true")) {

				log.info("preparing set: " + simplifySetInfo(set));

			}

			// setting the job properties needed to name the rawsets

			token.getEnv().setAttribute(set.get("jobProperty"), set.get("rawset"));

		}

		token.getEnv().setAttribute("sets", new Gson().toJson(setList));

		return Arc.DEFAULT_ARC;

	}

	private Map<String, String> simplifySetInfo(final Map<String, String> set) {

		return Maps.filterKeys(set, new Predicate<String>() {

			@Override

			public boolean apply(final String k) {

				return k.equals("set") || k.equals("rawset");

			}

		});

	}

	@SuppressWarnings("unchecked")

	protected List<Map<String, String>> getSetList() {

		return new Gson().fromJson(getSets(), List.class);

	}

	public String getSets() {

		return sets;

	}

	public void setSets(final String sets) {

		this.sets = sets;

	}

}

package eu.dnetlib.msro.workflows.dedup;

import org.apache.commons.lang.StringUtils;

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

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.msro.workflows.dedup.conf.DedupConfigurationOrchestration;

import eu.dnetlib.msro.workflows.dedup.conf.DedupConfigurationOrchestrationLoader;

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

public class DedupConfigurationSetterJobNode extends AsyncJobNode {

	private String dedupConfigSequence;

	private String dedupConfigSequenceParam;

	@Autowired

	private DedupConfigurationOrchestrationLoader dedupOrchestrationLoader;

	@Override

	protected String execute(final NodeToken token) throws Exception {

		if (StringUtils.isBlank(getDedupConfigSequence())) throw new IllegalArgumentException("missing configuration sequence");

		final DedupConfigurationOrchestration dedupOrchestration = dedupOrchestrationLoader.loadByActionSetId(getDedupConfigSequence());

		token.getEnv().setAttribute("entityType", dedupOrchestration.getEntity().getName());

		token.getEnv().setAttribute("entityTypeId", dedupOrchestration.getEntity().getCode());

		token.getEnv().setAttribute(getDedupConfigSequenceParam(), dedupOrchestration.toString());

		return Arc.DEFAULT_ARC;

	}

	public String getDedupConfigSequence() {

		return dedupConfigSequence;

	}

	public void setDedupConfigSequence(final String dedupConfigSequence) {

		this.dedupConfigSequence = dedupConfigSequence;

	}

	public String getDedupConfigSequenceParam() {

		return dedupConfigSequenceParam;

	}

	public void setDedupConfigSequenceParam(final String dedupConfigSequenceParam) {

		this.dedupConfigSequenceParam = dedupConfigSequenceParam;

	}

}

package eu.dnetlib.msro.workflows.dedup;

import org.apache.commons.collections.CollectionUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.Engine;

import com.googlecode.sarasvati.NodeToken;

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

import eu.dnetlib.msro.workflows.dedup.conf.DedupConfigurationOrchestration;

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

public class DedupDuplicateScanJobNode extends DedupConfigurationAwareJobNode {

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

	private class DedupBlackboardWorkflowJobListener extends BlackboardWorkflowJobListener {

		public DedupBlackboardWorkflowJobListener(final Engine engine, final NodeToken token) {

			super(engine, token);

		}

		@Override

		protected void onDone(final BlackboardJob job) {

			final DedupConfigurationOrchestration confs = dedupConfigurations(getToken());

			confs.getConfigurations().poll();

			log.info("checking dedup configs queue, size: " + confs.getConfigurations().size());

			if (CollectionUtils.isEmpty(confs.getConfigurations())) {

				log.info("dedup similarity scan done");

				super.complete(job, "done");

			} else {

				log.debug("remaining confs: " + confs);

				getToken().getEnv().setAttribute(getDedupConfigSequenceParam(), confs.toString());

				super.complete(job, Arc.DEFAULT_ARC);

			}

		}

	}

	@Override

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

		return new DedupBlackboardWorkflowJobListener(engine, token);

	}

}

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;

	}

}

package eu.dnetlib.msro.workflows.dedup.conf;

import java.io.StringReader;

import java.util.List;

import java.util.Queue;

import javax.annotation.Resource;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.dom4j.Document;

import org.dom4j.DocumentException;

import org.dom4j.Element;

import org.dom4j.io.SAXReader;

import com.google.common.collect.Lists;

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

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

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

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

import eu.dnetlib.pace.config.DedupConfig;

/**

 * The Class DedupConfigurationOrchestrationLoader.

 */

public class DedupConfigurationOrchestrationLoader {

	/** The Constant log. */

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

	/** The service locator. */

	@Resource

	private UniqueServiceLocator serviceLocator;

	/**

	 * Load the dedup orchestration profile from the IS.

	 *

	 * @param id

	 *            the id

	 * @return the dedup configuration orchestration

	 * @throws ISLookUpDocumentNotFoundException

	 *             the IS look up document not found exception

	 * @throws ISLookUpException

	 *             the IS look up exception

	 * @throws DocumentException

	 *             the document exception

	 */

	public DedupConfigurationOrchestration loadByActionSetId(final String id) throws Exception {

		final ISLookUpService isLookUpService = serviceLocator.getService(ISLookUpService.class);

		final String xquery = String.format("/RESOURCE_PROFILE[.//DEDUPLICATION/ACTION_SET/@id = '%s']", id);

		log.info("loading dedup orchestration: " + xquery);

		return parseOrchestrationProfile(isLookUpService, isLookUpService.getResourceProfileByQuery(xquery));

	}

	public List<DedupConfigurationOrchestration> loadByEntityName(final String entityName) throws Exception {

		final ISLookUpService isLookUpService = serviceLocator.getService(ISLookUpService.class);

		final String xquery = String.format("/RESOURCE_PROFILE[.//DEDUPLICATION/ENTITY/@name = '%s']", entityName);

		log.info("loading dedup orchestration: " + xquery);

		final List<DedupConfigurationOrchestration> res = Lists.newArrayList();

		for (final String profile : isLookUpService.quickSearchProfile(xquery)) {

			res.add(parseOrchestrationProfile(isLookUpService, profile));

		}

		return res;

	}

	private DedupConfigurationOrchestration parseOrchestrationProfile(final ISLookUpService isLookUpService, final String dedupOrchestation)

			throws DocumentException,

			ISLookUpException, ISLookUpDocumentNotFoundException {

		final Document doc = new SAXReader().read(new StringReader(dedupOrchestation));

		final Element e = (Element) doc.selectSingleNode("//DEDUPLICATION/ENTITY");

		final Entity entity = new Entity(e.attributeValue("name"), e.attributeValue("code"), e.attributeValue("label"));

		final String actionSetId = doc.valueOf("//DEDUPLICATION/ACTION_SET/@id");

		final Queue<DedupConfig> configurations = Lists.newLinkedList();

		for (final Object o : doc.selectNodes("//SCAN_SEQUENCE/SCAN")) {

			configurations.add(loadConfig(isLookUpService, actionSetId, o));

		}

		final DedupConfigurationOrchestration dco = new DedupConfigurationOrchestration(entity, actionSetId, configurations);

		log.debug("loaded dedup configuration orchestration: " + dco.toString());

		log.info("loaded dedup configuration orchestration, size: " + dco.getConfigurations().size());

		return dco;

	}

	private DedupConfig loadConfig(final ISLookUpService isLookUpService, final String actionSetId, final Object o) throws ISLookUpException,

			ISLookUpDocumentNotFoundException {

		final Element s = (Element) o;

		final String configProfileId = s.attributeValue("id");

		final String conf =

				isLookUpService.getResourceProfileByQuery(String.format(

						"for $x in /RESOURCE_PROFILE[.//RESOURCE_IDENTIFIER/@value = '%s'] return $x//DEDUPLICATION/text()",

						configProfileId));

		final DedupConfig dedupConfig = DedupConfig.load(conf);

		dedupConfig.getWf().setConfigurationId(actionSetId);

		return dedupConfig;

	}

}

package eu.dnetlib.msro.workflows.dedup.conf;

import com.google.gson.GsonBuilder;

/**

 * The Class Entity.

 */

public class Entity {

	/** The name. */

	private String name;

	/** The code. */

	private String code;

	/** The label. */

	private String label;

	public Entity() {}

	/**

	 * Instantiates a new entity.

	 *

	 * @param name

	 *            the name

	 * @param code

	 *            the code

	 * @param label

	 *            the label

	 */

	public Entity(final String name, final String code, final String label) {

		super();

		this.setName(name);

		this.setCode(code);

		this.setLabel(label);

	}

	/**

	 * Gets the name.

	 *

	 * @return the name

	 */

	public String getName() {

		return name;

	}

	/**

	 * Gets the code.

	 *

	 * @return the code

	 */

	public String getCode() {

		return code;

	}

	/**

	 * Gets the label.

	 *

	 * @return the label

	 */

	public String getLabel() {

		return label;

	}

	public void setName(final String name) {

		this.name = name;

	}

	public void setCode(final String code) {

		this.code = code;

	}

	public void setLabel(final String label) {

		this.label = label;

	}

	/*

	 * (non-Javadoc)

	 *

	 * @see java.lang.Object#toString()

	 */

	@Override

	public String toString() {

		return new GsonBuilder().setPrettyPrinting().create().toJson(this);

	}

}

package eu.dnetlib.msro.workflows.dedup.conf;

import java.util.Queue;

import com.google.gson.Gson;

import com.google.gson.GsonBuilder;

import eu.dnetlib.pace.config.DedupConfig;

/**

 * The Class DedupConfigurationOrchestration.

 */

public class DedupConfigurationOrchestration {

	/** The entity. */

	private Entity entity;

	/** The action set id. */

	private String actionSetId;

	/** The configurations. */

	private Queue<DedupConfig> configurations;

	public DedupConfigurationOrchestration() {}

	/**

	 * Instantiates a new dedup configuration orchestration.

	 *

	 * @param entity

	 *            the entity

	 * @param actionSetId

	 *            the action set id

	 * @param configurations

	 *            the configurations

	 */

	public DedupConfigurationOrchestration(final Entity entity, final String actionSetId, final Queue<DedupConfig> configurations) {

		super();

		this.setEntity(entity);

		this.setActionSetId(actionSetId);

		this.setConfigurations(configurations);

	}

	/**

	 * Gets the entity.

	 *

	 * @return the entity

	 */

	public Entity getEntity() {

		return entity;

	}

	/**

	 * Gets the action set id.

	 *

	 * @return the action set id

	 */

	public String getActionSetId() {

		return actionSetId;

	}

	/**

	 * Gets the configurations.

	 *

	 * @return the configurations

	 */

	public Queue<DedupConfig> getConfigurations() {

		return configurations;

	}

	public void setEntity(final Entity entity) {

		this.entity = entity;

	}

	public void setActionSetId(final String actionSetId) {

		this.actionSetId = actionSetId;

	}

	public void setConfigurations(final Queue<DedupConfig> configurations) {

		this.configurations = configurations;

	}

	/**

	 * From json.

	 *

	 * @param json

	 *            the json

	 * @return the dedup configuration orchestration

	 */

	public static DedupConfigurationOrchestration fromJSON(final String json) {

		return new Gson().fromJson(json, DedupConfigurationOrchestration.class);

	}

	/*

	 * (non-Javadoc)

	 * 

	 * @see java.lang.Object#toString()

	 */

	@Override

	public String toString() {

		return new GsonBuilder().setPrettyPrinting().create().toJson(this);

	}

}

package eu.dnetlib.msro.workflows.dedup;

import java.util.Queue;

import org.apache.commons.lang.StringUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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

import com.google.common.base.Function;

import com.google.common.base.Splitter;

import com.google.common.collect.Iterables;

import com.google.common.collect.Lists;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.msro.rmi.MSROException;

import eu.dnetlib.msro.workflows.dedup.conf.DedupConfigurationOrchestration;

import eu.dnetlib.msro.workflows.dedup.conf.DedupConfigurationOrchestrationLoader;

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

public class DedupCheckEntitySequenceJobNode extends AsyncJobNode {

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

	@Autowired

	private DedupConfigurationOrchestrationLoader dedupOrchestrationLoader;

	private String dedupConfigSequenceParam;

	private String entitySequence;

	@Override

	protected String execute(final NodeToken token) throws Exception {

		if (StringUtils.isBlank(getEntitySequence())) throw new MSROException("missing entity sequence, e.g. a csv: organization,person,result");

		if (token.getFullEnv().hasAttribute(DedupGrouperJobNode.DEDUP_GROUPER_LOOPER)) {

			log.info("reset env variable: " + DedupGrouperJobNode.DEDUP_GROUPER_LOOPER + " to zero");

			token.getFullEnv().setAttribute(DedupGrouperJobNode.DEDUP_GROUPER_LOOPER, 0);

		}

		if (!token.getEnv().hasAttribute("entitySequence")) {

			log.info("parsing config sequence: " + getEntitySequence());

			token.getEnv().setAttribute("entitySequence", getEntitySequence());

			final Iterable<String> sequence = Splitter.on(",").omitEmptyStrings().split(getEntitySequence());

			final Queue<DedupConfigurationOrchestration> q =

					Lists.newLinkedList(Iterables.transform(sequence, new Function<String, DedupConfigurationOrchestration>() {

						@Override

						public DedupConfigurationOrchestration apply(final String entityName) {

							try {

								final DedupConfigurationOrchestration dco = Iterables.getFirst(dedupOrchestrationLoader.loadByEntityName(entityName), null);

								if (dco == null) throw new RuntimeException("unable to find DedupOrchestration profile for entity type: " + entityName);

								return dco;

							} catch (final Throwable e) {

								throw new RuntimeException("", e);

							}

						}

					}));

			log.info("built sequence of dedup orchestration profiles, size: " + q.size());

			final DedupConfigurationOrchestration dco = q.remove();

			log.info("closing mesh for entity: " + dco.getEntity().getName());

			setDedupConfParams(token, dco);

			token.getEnv().setTransientAttribute("entitySequenceQueue", q);

			return Arc.DEFAULT_ARC;

		}

		@SuppressWarnings("unchecked")

		final Queue<DedupConfigurationOrchestration> q = (Queue<DedupConfigurationOrchestration>) token.getEnv().getTransientAttribute("entitySequenceQueue");

		if (!q.isEmpty()) {

			log.info("remaining dedup orchestration profiles: " + q.size());

			final DedupConfigurationOrchestration dco = q.remove();

			log.info("closing mesh for entity: " + dco.getEntity().getName());

			setDedupConfParams(token, dco);

			return Arc.DEFAULT_ARC;

		}

		log.info("completed closing mesh for entities: " + getEntitySequence());

		return "done";

	}

	private void setDedupConfParams(final NodeToken token, final DedupConfigurationOrchestration dco) {

		token.getEnv().setAttribute("entityType", dco.getEntity().getName());

		token.getEnv().setAttribute("entityTypeId", dco.getEntity().getCode());

		token.getEnv().setAttribute(getDedupConfigSequenceParam(), dco.toString());

	}

	public String getEntitySequence() {

		return entitySequence;

	}

	public void setEntitySequence(final String entitySequence) {

		this.entitySequence = entitySequence;

	}

	public String getDedupConfigSequenceParam() {

		return dedupConfigSequenceParam;

	}

	public void setDedupConfigSequenceParam(final String dedupConfigSequenceParam) {

		this.dedupConfigSequenceParam = dedupConfigSequenceParam;

	}

}

package eu.dnetlib.msro.workflows.dedup;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.Engine;

import com.googlecode.sarasvati.NodeToken;

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

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

public class DedupGrouperJobNode extends DedupConfigurationAwareJobNode {

	// TODO factor out this constant, it should be a configuration parameter

	public static final int DEDUP_GROUPER_MAX_LOOPS = 10;

	public static final String DEDUP_GROUPER_LOOPER = "dedup.grouper.looper";

	public static final String DEDUP_GROUPER_CURR_WRITTEN_RELS = "dedup.grouper.written.rels";

	public static final String DEDUP_GROUPER_PREV_WRITTEN_RELS = "dedup.grouper.prev.written.rels";

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

	private class DedupBlackboardWorkflowJobListener extends BlackboardWorkflowJobListener {

		public DedupBlackboardWorkflowJobListener(final Engine engine, final NodeToken token) {

			super(engine, token);

		}

		@Override

		protected void onDone(final BlackboardJob job) {

			final int times = currentIteration(getToken());

			final String curr = job.getParameters().get(DEDUP_GROUPER_CURR_WRITTEN_RELS);

			if (times == 0) {

				getToken().getFullEnv().setAttribute(DEDUP_GROUPER_PREV_WRITTEN_RELS, -1);

			}

			if ((times >= DEDUP_GROUPER_MAX_LOOPS) || isStable(getToken(), curr)) {

				super.complete(job, "done");

			} else {

				log.info("incrementing dedup.grouper.looper to " + (times + 1));

				getToken().getFullEnv().setAttribute(DEDUP_GROUPER_LOOPER, times + 1);

				getToken().getFullEnv().setAttribute(DEDUP_GROUPER_PREV_WRITTEN_RELS, curr);

				super.complete(job, Arc.DEFAULT_ARC);

			}

		}

	}

	private int currentIteration(final NodeToken token) {

		try {

			final String sTimes = token.getFullEnv().getAttribute(DEDUP_GROUPER_LOOPER);

			log.info("read dedup.grouper.looper from fullEnv: '" + sTimes + "'");

			return Integer.parseInt(sTimes);

		} catch (final NumberFormatException e) {

			log.info("got empty dedup.grouper.looper, initializing to 0");

			return 0;

		}

	}

	private boolean isStable(final NodeToken token, final String sCurr) {

		final String sPrev = token.getFullEnv().getAttribute(DEDUP_GROUPER_PREV_WRITTEN_RELS);

		log.info("Comparing written rels, prev=" + sPrev + ", curr=" + sCurr);

		try {

			final boolean b = Integer.parseInt(sCurr) == Integer.parseInt(sPrev);

			if (b) {

				log.info("  --- The number of written rels is STABLE");

			}

			return b;

		} catch (final Exception e) {

			log.error("Invalid parsing of written rels counters - curr: " + sCurr + ", prev: " + sPrev);

			return false;

		}

	}

	@Override

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

		return new DedupBlackboardWorkflowJobListener(engine, token);

	}

}

package eu.dnetlib.msro.workflows.dedup;

import java.util.List;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

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

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

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

import eu.dnetlib.msro.rmi.MSROException;

import eu.dnetlib.msro.workflows.dedup.conf.DedupConfigurationOrchestration;

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

public class DedupCheckConfigurationJobNode extends AsyncJobNode {

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

	@Autowired

	private UniqueServiceLocator serviceLocator;

	private String dedupConfigSequenceParam;

	@Override

	protected String execute(final NodeToken token) throws Exception {

		final String dcoJson = token.getEnv().getAttribute(getDedupConfigSequenceParam());

		final DedupConfigurationOrchestration dco = DedupConfigurationOrchestration.fromJSON(dcoJson);

		if (!existActionSetProfile(dco)) throw new MSROException("missing action set profile: " + dco.getActionSetId());

		log.info("found action set profile: " + dco.getActionSetId());

		return Arc.DEFAULT_ARC;

	}

	private boolean existActionSetProfile(final DedupConfigurationOrchestration dco) throws ISLookUpException {

		final String xquery = "for $x in //RESOURCE_PROFILE[.//RESOURCE_TYPE/@value = 'DedupOrchestrationDSResourceType' and .//ACTION_SET/@id='%s'] return 1";

		log.info("looking for action set profile, id: " + dco.getActionSetId());

		final List<String> actionSets = serviceLocator.getService(ISLookUpService.class).quickSearchProfile(String.format(xquery, dco.getActionSetId()));

		return !actionSets.isEmpty();

	}

	public String getDedupConfigSequenceParam() {

		return dedupConfigSequenceParam;

	}

	public void setDedupConfigSequenceParam(final String dedupConfigSequenceParam) {

		this.dedupConfigSequenceParam = dedupConfigSequenceParam;

	}

}

package eu.dnetlib.msro.workflows.dedup;

import java.util.List;

import java.util.Map;

import javax.annotation.Resource;

import org.apache.commons.lang.StringUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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

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

import com.google.common.collect.Iterables;

import com.google.gson.Gson;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

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

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

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

import eu.dnetlib.miscutils.datetime.DateUtils;

import eu.dnetlib.msro.workflows.dedup.conf.DedupConfigurationOrchestrationLoader;

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

public class PrepareDedupIndexJobNode extends SimpleJobNode {

	/**

	 * logger.

	 */

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

	@Resource

	private UniqueServiceLocator serviceLocator;

	@Autowired

	private DedupConfigurationOrchestrationLoader dedupOrchestrationLoader;

	private String rottenRecordsPathParam;

	private String hbaseTable;

	private String dedupConfig;

	@Override

	protected String execute(final NodeToken token) throws Exception {

		log.info("start preparing job");

		final String fields = getFields(env("format", token), env("layout", token));

		token.getEnv().setAttribute("index.fields", fields);

		if (!StringUtils.isBlank(getRottenRecordsPathParam())) {

			token.getEnv().setAttribute(getRottenRecordsPathParam(), "/tmp" + getFileName(token, "rottenrecords"));

		}

		token.getEnv().setAttribute("index.solr.url", getIndexSolrUrlZk());

		token.getEnv().setAttribute("index.solr.collection", getCollectionName(token));

		token.getEnv().setAttribute("index.shutdown.wait.time", getIndexSolrShutdownWait());

		token.getEnv().setAttribute("index.buffer.flush.threshold", getIndexBufferFlushTreshold());

		token.getEnv().setAttribute("index.solr.sim.mode", isFeedingSimulationMode());

		token.getEnv().setAttribute("index.feed.timestamp", DateUtils.now_ISO8601());

		@SuppressWarnings("unchecked")

		final List<Map<String, String>> sets = new Gson().fromJson(token.getEnv().getAttribute("sets"), List.class);

		token.getEnv().setAttribute("actionset", Iterables.getOnlyElement(sets).get("set"));

		token.getEnv().setAttribute("entityType", token.getEnv().getAttribute("entityType"));

		token.getEnv().setAttribute("entityTypeId", token.getEnv().getAttribute("entityTypeId"));

		return Arc.DEFAULT_ARC;

	}

	private String getFields(final String format, final String layout) throws ISLookUpException {

		return isLookup(String

				.format("<FIELDS>{for $x in collection('')//RESOURCE_PROFILE[.//RESOURCE_TYPE/@value = 'MDFormatDSResourceType' and .//NAME='%s']//LAYOUT[@name='%s']/FIELDS/FIELD return $x[string(@path)]}</FIELDS>",

						format, layout));

	}

	public String getIndexSolrUrlZk() throws ISLookUpException {

		return isLookup("for $x in /RESOURCE_PROFILE[.//RESOURCE_TYPE/@value='IndexServiceResourceType'] return $x//PROTOCOL[./@name='solr']/@address/string()");

	}

	public String getIndexSolrShutdownWait() throws ISLookUpException {

		return queryForServiceProperty("solr:feedingShutdownTolerance");

	}

	public String getIndexBufferFlushTreshold() throws ISLookUpException {

		return queryForServiceProperty("solr:feedingBufferFlushThreshold");

	}

	public String isFeedingSimulationMode() throws ISLookUpException {

		return queryForServiceProperty("solr:feedingSimulationMode");

	}

	private String queryForServiceProperty(final String key) throws ISLookUpException {

		return isLookup("for $x in /RESOURCE_PROFILE[.//RESOURCE_TYPE/@value='IndexServiceResourceType'] return $x//SERVICE_PROPERTIES/PROPERTY[./@ key='"