Merge branch 'master' into provision_indexing

dhp-workflows/dhp-broker-events/src/main/java/eu/dnetlib/dhp/broker/model/Topic.java

@@ -20,7 +20,7 @@ public enum Topic {
 	// ENRICHMENT MORE
 	ENRICH_MORE_PID("ENRICH/MORE/PID"), ENRICH_MORE_OA_VERSION("ENRICH/MORE/OPENACCESS_VERSION"), ENRICH_MORE_ABSTRACT(
 		"ENRICH/MORE/ABSTRACT"), ENRICH_MORE_PUBLICATION_DATE("ENRICH/MORE/PUBLICATION_DATE"), ENRICH_MORE_PROJECT(
-			"ENRICH/MORE/PROJECT"), ENRICH_MORE_SUBJECT_MESHEUROPMC(
+			"ENRICH/MORE/PROJECT"), ENRICH_MORE_SOFTWARE("ENRICH/MORE/SOFTWARE"), ENRICH_MORE_SUBJECT_MESHEUROPMC(
 				"ENRICH/MORE/SUBJECT/MESHEUROPMC"), ENRICH_MORE_SUBJECT_ARXIV(
 					"ENRICH/MORE/SUBJECT/ARXIV"), ENRICH_MORE_SUBJECT_JEL(
 						"ENRICH/MORE/SUBJECT/JEL"), ENRICH_MORE_SUBJECT_DDC(
@@ -28,7 +28,21 @@ public enum Topic {
 								"ENRICH/MORE/SUBJECT/ACM"), ENRICH_MORE_SUBJECT_RVK("ENRICH/MORE/SUBJECT/RVK"),
 
 	// ADDITION
-	ADD_BY_PROJECT("ADD/BY_PROJECT");
+	ADD_BY_PROJECT("ADD/BY_PROJECT"),
+
+	// OTHER RELS
+	ENRICH_MISSING_PUBLICATION_IS_RELATED_TO(
+		"ENRICH/MISSING/PUBLICATION/IS_RELATED_TO"), ENRICH_MISSING_PUBLICATION_REFERENCES(
+			"ENRICH/MISSING/PUBLICATION/REFERENCES"), ENRICH_MISSING_PUBLICATION_IS_REFERENCED_BY(
+				"ENRICH/MISSING/PUBLICATION/IS_REFERENCED_BY"), ENRICH_MISSING_PUBLICATION_IS_SUPPLEMENTED_TO(
+					"ENRICH/MISSING/PUBLICATION/IS_SUPPLEMENTED_TO"), ENRICH_MISSING_PUBLICATION_IS_SUPPLEMENTED_BY(
+						"ENRICH/MISSING/PUBLICATION/IS_SUPPLEMENTED_BY"),
+
+	ENRICH_MISSING_DATASET_IS_RELATED_TO("ENRICH/MISSING/DATASET/IS_RELATED_TO"), ENRICH_MISSING_DATASET_REFERENCES(
+		"ENRICH/MISSING/DATASET/REFERENCES"), ENRICH_MISSING_DATASET_IS_REFERENCED_BY(
+			"ENRICH/MISSING/DATASET/IS_REFERENCED_BY"), ENRICH_MISSING_DATASET_IS_SUPPLEMENTED_TO(
+				"ENRICH/MISSING/DATASET/IS_SUPPLEMENTED_TO"), ENRICH_MISSING_DATASET_IS_SUPPLEMENTED_BY(
+					"ENRICH/MISSING/DATASET/IS_SUPPLEMENTED_BY"),;
 
 	Topic(final String path) {
 		this.path = path;

dhp-workflows/dhp-broker-events/src/main/java/eu/dnetlib/dhp/broker/oa/GenerateEventsApplication.java

@@ -4,48 +4,101 @@ package eu.dnetlib.dhp.broker.oa;
 import static eu.dnetlib.dhp.common.SparkSessionSupport.runWithSparkSession;
 
 import java.util.ArrayList;
+import java.util.Collection;
 import java.util.List;
+import java.util.Map;
 import java.util.Optional;
 import java.util.stream.Collectors;
 
 import org.apache.commons.io.IOUtils;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.hadoop.io.compress.GzipCodec;
 import org.apache.spark.SparkConf;
+import org.apache.spark.api.java.JavaRDD;
+import org.apache.spark.api.java.JavaSparkContext;
+import org.apache.spark.api.java.function.MapFunction;
+import org.apache.spark.sql.Column;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Encoders;
+import org.apache.spark.sql.Row;
 import org.apache.spark.sql.SparkSession;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import com.fasterxml.jackson.databind.ObjectMapper;
+
 import eu.dnetlib.dhp.application.ArgumentApplicationParser;
 import eu.dnetlib.dhp.broker.model.Event;
 import eu.dnetlib.dhp.broker.model.EventFactory;
 import eu.dnetlib.dhp.broker.oa.matchers.EnrichMissingAbstract;
 import eu.dnetlib.dhp.broker.oa.matchers.EnrichMissingAuthorOrcid;
+import eu.dnetlib.dhp.broker.oa.matchers.EnrichMissingDatasetIsReferencedBy;
+import eu.dnetlib.dhp.broker.oa.matchers.EnrichMissingDatasetIsRelatedTo;
+import eu.dnetlib.dhp.broker.oa.matchers.EnrichMissingDatasetIsSupplementedBy;
+import eu.dnetlib.dhp.broker.oa.matchers.EnrichMissingDatasetIsSupplementedTo;
+import eu.dnetlib.dhp.broker.oa.matchers.EnrichMissingDatasetReferences;
 import eu.dnetlib.dhp.broker.oa.matchers.EnrichMissingOpenAccess;
 import eu.dnetlib.dhp.broker.oa.matchers.EnrichMissingPid;
 import eu.dnetlib.dhp.broker.oa.matchers.EnrichMissingProject;
 import eu.dnetlib.dhp.broker.oa.matchers.EnrichMissingPublicationDate;
+import eu.dnetlib.dhp.broker.oa.matchers.EnrichMissingPublicationIsReferencedBy;
+import eu.dnetlib.dhp.broker.oa.matchers.EnrichMissingPublicationIsRelatedTo;
+import eu.dnetlib.dhp.broker.oa.matchers.EnrichMissingPublicationIsSupplementedBy;
+import eu.dnetlib.dhp.broker.oa.matchers.EnrichMissingPublicationIsSupplementedTo;
+import eu.dnetlib.dhp.broker.oa.matchers.EnrichMissingPublicationReferences;
+import eu.dnetlib.dhp.broker.oa.matchers.EnrichMissingSoftware;
 import eu.dnetlib.dhp.broker.oa.matchers.EnrichMissingSubject;
 import eu.dnetlib.dhp.broker.oa.matchers.EnrichMoreOpenAccess;
 import eu.dnetlib.dhp.broker.oa.matchers.EnrichMorePid;
+import eu.dnetlib.dhp.broker.oa.matchers.EnrichMoreProject;
+import eu.dnetlib.dhp.broker.oa.matchers.EnrichMoreSoftware;
 import eu.dnetlib.dhp.broker.oa.matchers.EnrichMoreSubject;
 import eu.dnetlib.dhp.broker.oa.matchers.UpdateMatcher;
+import eu.dnetlib.dhp.broker.oa.util.BrokerConstants;
 import eu.dnetlib.dhp.broker.oa.util.UpdateInfo;
 import eu.dnetlib.dhp.common.HdfsSupport;
+import eu.dnetlib.dhp.schema.oaf.OtherResearchProduct;
+import eu.dnetlib.dhp.schema.oaf.Project;
+import eu.dnetlib.dhp.schema.oaf.Publication;
+import eu.dnetlib.dhp.schema.oaf.Relation;
 import eu.dnetlib.dhp.schema.oaf.Result;
+import eu.dnetlib.dhp.schema.oaf.Software;
 
 public class GenerateEventsApplication {
 
 	private static final Logger log = LoggerFactory.getLogger(GenerateEventsApplication.class);
 
-	private static final UpdateMatcher<?> enrichMissingAbstract = new EnrichMissingAbstract();
+	// Simple Matchers
-	private static final UpdateMatcher<?> enrichMissingAuthorOrcid = new EnrichMissingAuthorOrcid();
+	private static final UpdateMatcher<Result, ?> enrichMissingAbstract = new EnrichMissingAbstract();
-	private static final UpdateMatcher<?> enrichMissingOpenAccess = new EnrichMissingOpenAccess();
+	private static final UpdateMatcher<Result, ?> enrichMissingAuthorOrcid = new EnrichMissingAuthorOrcid();
-	private static final UpdateMatcher<?> enrichMissingPid = new EnrichMissingPid();
+	private static final UpdateMatcher<Result, ?> enrichMissingOpenAccess = new EnrichMissingOpenAccess();
-	private static final UpdateMatcher<?> enrichMissingProject = new EnrichMissingProject();
+	private static final UpdateMatcher<Result, ?> enrichMissingPid = new EnrichMissingPid();
-	private static final UpdateMatcher<?> enrichMissingPublicationDate = new EnrichMissingPublicationDate();
+	private static final UpdateMatcher<Result, ?> enrichMissingPublicationDate = new EnrichMissingPublicationDate();
-	private static final UpdateMatcher<?> enrichMissingSubject = new EnrichMissingSubject();
+	private static final UpdateMatcher<Result, ?> enrichMissingSubject = new EnrichMissingSubject();
-	private static final UpdateMatcher<?> enrichMoreOpenAccess = new EnrichMoreOpenAccess();
+	private static final UpdateMatcher<Result, ?> enrichMoreOpenAccess = new EnrichMoreOpenAccess();
-	private static final UpdateMatcher<?> enrichMorePid = new EnrichMorePid();
+	private static final UpdateMatcher<Result, ?> enrichMorePid = new EnrichMorePid();
-	private static final UpdateMatcher<?> enrichMoreSubject = new EnrichMoreSubject();
+	private static final UpdateMatcher<Result, ?> enrichMoreSubject = new EnrichMoreSubject();
+
+	// Advanced matchers
+	private static final UpdateMatcher<Pair<Result, List<Project>>, ?> enrichMissingProject = new EnrichMissingProject();
+	private static final UpdateMatcher<Pair<Result, List<Project>>, ?> enrichMoreProject = new EnrichMoreProject();
+
+	private static final UpdateMatcher<Pair<Result, List<Software>>, ?> enrichMissingSoftware = new EnrichMissingSoftware();
+	private static final UpdateMatcher<Pair<Result, List<Software>>, ?> enrichMoreSoftware = new EnrichMoreSoftware();
+
+	private static final UpdateMatcher<Pair<Result, List<Publication>>, ?> enrichMisissingPublicationIsRelatedTo = new EnrichMissingPublicationIsRelatedTo();
+	private static final UpdateMatcher<Pair<Result, List<Publication>>, ?> enrichMissingPublicationIsReferencedBy = new EnrichMissingPublicationIsReferencedBy();
+	private static final UpdateMatcher<Pair<Result, List<Publication>>, ?> enrichMissingPublicationReferences = new EnrichMissingPublicationReferences();
+	private static final UpdateMatcher<Pair<Result, List<Publication>>, ?> enrichMissingPublicationIsSupplementedTo = new EnrichMissingPublicationIsSupplementedTo();
+	private static final UpdateMatcher<Pair<Result, List<Publication>>, ?> enrichMissingPublicationIsSupplementedBy = new EnrichMissingPublicationIsSupplementedBy();
+
+	private static final UpdateMatcher<Pair<Result, List<eu.dnetlib.dhp.schema.oaf.Dataset>>, ?> enrichMisissingDatasetIsRelatedTo = new EnrichMissingDatasetIsRelatedTo();
+	private static final UpdateMatcher<Pair<Result, List<eu.dnetlib.dhp.schema.oaf.Dataset>>, ?> enrichMissingDatasetIsReferencedBy = new EnrichMissingDatasetIsReferencedBy();
+	private static final UpdateMatcher<Pair<Result, List<eu.dnetlib.dhp.schema.oaf.Dataset>>, ?> enrichMissingDatasetReferences = new EnrichMissingDatasetReferences();
+	private static final UpdateMatcher<Pair<Result, List<eu.dnetlib.dhp.schema.oaf.Dataset>>, ?> enrichMissingDatasetIsSupplementedTo = new EnrichMissingDatasetIsSupplementedTo();
+	private static final UpdateMatcher<Pair<Result, List<eu.dnetlib.dhp.schema.oaf.Dataset>>, ?> enrichMissingDatasetIsSupplementedBy = new EnrichMissingDatasetIsSupplementedBy();
+
+	public static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
 
 	public static void main(final String[] args) throws Exception {
 		final ArgumentApplicationParser parser = new ArgumentApplicationParser(
@@ -68,9 +121,19 @@ public class GenerateEventsApplication {
 		log.info("eventsPath: {}", eventsPath);
 
 		final SparkConf conf = new SparkConf();
+
 		runWithSparkSession(conf, isSparkSessionManaged, spark -> {
+			final JavaSparkContext sc = JavaSparkContext.fromSparkContext(spark.sparkContext());
 			removeOutputDir(spark, eventsPath);
-			generateEvents(spark, graphPath, eventsPath);
+
+			final JavaRDD<Event> eventsRdd = sc.emptyRDD();
+
+			eventsRdd.union(generateSimpleEvents(spark, graphPath, Publication.class));
+			eventsRdd.union(generateSimpleEvents(spark, graphPath, eu.dnetlib.dhp.schema.oaf.Dataset.class));
+			eventsRdd.union(generateSimpleEvents(spark, graphPath, Software.class));
+			eventsRdd.union(generateSimpleEvents(spark, graphPath, OtherResearchProduct.class));
+
+			eventsRdd.saveAsTextFile(eventsPath, GzipCodec.class);
 		});
 
 	}
@@ -79,11 +142,34 @@ public class GenerateEventsApplication {
 		HdfsSupport.remove(path, spark.sparkContext().hadoopConfiguration());
 	}
 
-	private static void generateEvents(final SparkSession spark, final String graphPath, final String eventsPath) {
+	private static <R extends Result> JavaRDD<Event> generateSimpleEvents(final SparkSession spark,
-		// TODO
+		final String graphPath,
+		final Class<R> resultClazz) {
+
+		final Dataset<R> results = readPath(
+			spark, graphPath + "/" + resultClazz.getSimpleName().toLowerCase(), resultClazz)
+				.filter(r -> r.getDataInfo().getDeletedbyinference());
+
+		final Dataset<Relation> rels = readPath(spark, graphPath + "/relation", Relation.class)
+			.filter(r -> r.getRelClass().equals(BrokerConstants.IS_MERGED_IN_CLASS));
+
+		final Column c = null; // TODO
+
+		final Dataset<Row> aa = results
+			.joinWith(rels, results.col("id").equalTo(rels.col("source")), "inner")
+			.groupBy(rels.col("target"))
+			.agg(c)
+			.filter(x -> x.size() > 1)
+		// generateSimpleEvents(...)
+		// flatMap()
+		// toRdd()
+		;
+
+		return null;
+
 	}
 
-	private List<Event> generateEvents(final Result... children) {
+	private List<Event> generateSimpleEvents(final Collection<Result> children) {
 		final List<UpdateInfo<?>> list = new ArrayList<>();
 
 		for (final Result target : children) {
@@ -91,7 +177,6 @@ public class GenerateEventsApplication {
 			list.addAll(enrichMissingAuthorOrcid.searchUpdatesForRecord(target, children));
 			list.addAll(enrichMissingOpenAccess.searchUpdatesForRecord(target, children));
 			list.addAll(enrichMissingPid.searchUpdatesForRecord(target, children));
-			list.addAll(enrichMissingProject.searchUpdatesForRecord(target, children));
 			list.addAll(enrichMissingPublicationDate.searchUpdatesForRecord(target, children));
 			list.addAll(enrichMissingSubject.searchUpdatesForRecord(target, children));
 			list.addAll(enrichMoreOpenAccess.searchUpdatesForRecord(target, children));
@@ -102,4 +187,94 @@ public class GenerateEventsApplication {
 		return list.stream().map(EventFactory::newBrokerEvent).collect(Collectors.toList());
 	}
 
+	private List<Event> generateProjectsEvents(final Collection<Pair<Result, List<Project>>> childrenWithProjects) {
+		final List<UpdateInfo<?>> list = new ArrayList<>();
+
+		for (final Pair<Result, List<Project>> target : childrenWithProjects) {
+			list.addAll(enrichMissingProject.searchUpdatesForRecord(target, childrenWithProjects));
+			list.addAll(enrichMoreProject.searchUpdatesForRecord(target, childrenWithProjects));
+		}
+
+		return list.stream().map(EventFactory::newBrokerEvent).collect(Collectors.toList());
+	}
+
+	private List<Event> generateSoftwareEvents(final Collection<Pair<Result, List<Software>>> childrenWithSoftwares) {
+		final List<UpdateInfo<?>> list = new ArrayList<>();
+
+		for (final Pair<Result, List<Software>> target : childrenWithSoftwares) {
+			list.addAll(enrichMissingSoftware.searchUpdatesForRecord(target, childrenWithSoftwares));
+			list.addAll(enrichMoreSoftware.searchUpdatesForRecord(target, childrenWithSoftwares));
+		}
+		return list.stream().map(EventFactory::newBrokerEvent).collect(Collectors.toList());
+	}
+
+	private List<Event> generatePublicationRelatedEvents(final String relType,
+		final Collection<Pair<Result, Map<String, List<Publication>>>> childrenWithRels) {
+
+		final List<UpdateInfo<?>> list = new ArrayList<>();
+
+		final List<Pair<Result, List<Publication>>> cleanedChildrens = childrenWithRels
+			.stream()
+			.filter(p -> p.getRight().containsKey(relType))
+			.map(p -> Pair.of(p.getLeft(), p.getRight().get(relType)))
+			.filter(p -> p.getRight().size() > 0)
+			.collect(Collectors.toList());
+
+		for (final Pair<Result, List<Publication>> target : cleanedChildrens) {
+			if (relType.equals("isRelatedTo")) {
+				list.addAll(enrichMisissingPublicationIsRelatedTo.searchUpdatesForRecord(target, cleanedChildrens));
+			} else if (relType.equals("references")) {
+				list.addAll(enrichMissingPublicationReferences.searchUpdatesForRecord(target, cleanedChildrens));
+			} else if (relType.equals("isReferencedBy")) {
+				list.addAll(enrichMissingPublicationIsReferencedBy.searchUpdatesForRecord(target, cleanedChildrens));
+			} else if (relType.equals("isSupplementedTo")) {
+				list.addAll(enrichMissingPublicationIsSupplementedTo.searchUpdatesForRecord(target, cleanedChildrens));
+			} else if (relType.equals("isSupplementedBy")) {
+				list.addAll(enrichMissingPublicationIsSupplementedBy.searchUpdatesForRecord(target, cleanedChildrens));
+			}
+		}
+
+		return list.stream().map(EventFactory::newBrokerEvent).collect(Collectors.toList());
+
+	}
+
+	private List<Event> generateDatasetRelatedEvents(final String relType,
+		final Collection<Pair<Result, Map<String, List<eu.dnetlib.dhp.schema.oaf.Dataset>>>> childrenWithRels) {
+
+		final List<UpdateInfo<?>> list = new ArrayList<>();
+
+		final List<Pair<Result, List<eu.dnetlib.dhp.schema.oaf.Dataset>>> cleanedChildrens = childrenWithRels
+			.stream()
+			.filter(p -> p.getRight().containsKey(relType))
+			.map(p -> Pair.of(p.getLeft(), p.getRight().get(relType)))
+			.filter(p -> p.getRight().size() > 0)
+			.collect(Collectors.toList());
+
+		for (final Pair<Result, List<eu.dnetlib.dhp.schema.oaf.Dataset>> target : cleanedChildrens) {
+			if (relType.equals("isRelatedTo")) {
+				list.addAll(enrichMisissingDatasetIsRelatedTo.searchUpdatesForRecord(target, cleanedChildrens));
+			} else if (relType.equals("references")) {
+				list.addAll(enrichMissingDatasetReferences.searchUpdatesForRecord(target, cleanedChildrens));
+			} else if (relType.equals("isReferencedBy")) {
+				list.addAll(enrichMissingDatasetIsReferencedBy.searchUpdatesForRecord(target, cleanedChildrens));
+			} else if (relType.equals("isSupplementedTo")) {
+				list.addAll(enrichMissingDatasetIsSupplementedTo.searchUpdatesForRecord(target, cleanedChildrens));
+			} else if (relType.equals("isSupplementedBy")) {
+				list.addAll(enrichMissingDatasetIsSupplementedBy.searchUpdatesForRecord(target, cleanedChildrens));
+			}
+		}
+
+		return list.stream().map(EventFactory::newBrokerEvent).collect(Collectors.toList());
+
+	}
+
+	public static <R> Dataset<R> readPath(
+		final SparkSession spark,
+		final String inputPath,
+		final Class<R> clazz) {
+		return spark
+			.read()
+			.textFile(inputPath)
+			.map((MapFunction<String, R>) value -> OBJECT_MAPPER.readValue(value, clazz), Encoders.bean(clazz));
+	}
 }

dhp-workflows/dhp-broker-events/src/main/java/eu/dnetlib/dhp/broker/oa/matchers/EnrichMissingAbstract.java

@@ -9,7 +9,7 @@ import eu.dnetlib.dhp.broker.model.Topic;
 import eu.dnetlib.dhp.broker.oa.util.UpdateInfo;
 import eu.dnetlib.dhp.schema.oaf.Result;
 
-public class EnrichMissingAbstract extends UpdateMatcher<String> {
+public class EnrichMissingAbstract extends UpdateMatcher<Result, String> {
 
 	public EnrichMissingAbstract() {
 		super(false);
@@ -24,7 +24,8 @@ public class EnrichMissingAbstract extends UpdateMatcher<String> {
 	}
 
 	@Override
-	public UpdateInfo<String> generateUpdateInfo(final String highlightValue, final Result source,
+	public UpdateInfo<String> generateUpdateInfo(final String highlightValue,
+		final Result source,
 		final Result target) {
 		return new UpdateInfo<>(
 			Topic.ENRICH_MISSING_ABSTRACT,

dhp-workflows/dhp-broker-events/src/main/java/eu/dnetlib/dhp/broker/oa/matchers/EnrichMissingAuthorOrcid.java

@@ -10,7 +10,7 @@ import eu.dnetlib.dhp.broker.model.Topic;
 import eu.dnetlib.dhp.broker.oa.util.UpdateInfo;
 import eu.dnetlib.dhp.schema.oaf.Result;
 
-public class EnrichMissingAuthorOrcid extends UpdateMatcher<Pair<String, String>> {
+public class EnrichMissingAuthorOrcid extends UpdateMatcher<Result, Pair<String, String>> {
 
 	public EnrichMissingAuthorOrcid() {
 		super(true);
@@ -24,7 +24,8 @@ public class EnrichMissingAuthorOrcid extends UpdateMatcher<Pair<String, String>
 
 	@Override
 	public UpdateInfo<Pair<String, String>> generateUpdateInfo(final Pair<String, String> highlightValue,
-		final Result source, final Result target) {
+		final Result source,
+		final Result target) {
 		return new UpdateInfo<>(
 			Topic.ENRICH_MISSING_AUTHOR_ORCID,
 			highlightValue, source, target,

dhp-workflows/dhp-broker-events/src/main/java/eu/dnetlib/dhp/broker/oa/matchers/EnrichMissingDatasetIsReferencedBy.java

@@ -0,0 +1,38 @@
+
+package eu.dnetlib.dhp.broker.oa.matchers;
+
+import java.util.List;
+
+import org.apache.commons.lang3.tuple.Pair;
+
+import eu.dnetlib.dhp.broker.model.Topic;
+import eu.dnetlib.dhp.broker.oa.util.UpdateInfo;
+import eu.dnetlib.dhp.schema.oaf.Dataset;
+import eu.dnetlib.dhp.schema.oaf.Result;
+
+public class EnrichMissingDatasetIsReferencedBy
+	extends UpdateMatcher<Pair<Result, List<Dataset>>, eu.dnetlib.broker.objects.Dataset> {
+
+	public EnrichMissingDatasetIsReferencedBy() {
+		super(true);
+	}
+
+	@Override
+	protected List<UpdateInfo<eu.dnetlib.broker.objects.Dataset>> findUpdates(final Pair<Result, List<Dataset>> source,
+		final Pair<Result, List<Dataset>> target) {
+		// TODO Auto-generated method stub
+		return null;
+	}
+
+	@Override
+	protected UpdateInfo<eu.dnetlib.broker.objects.Dataset> generateUpdateInfo(
+		final eu.dnetlib.broker.objects.Dataset highlightValue,
+		final Pair<Result, List<Dataset>> source,
+		final Pair<Result, List<Dataset>> target) {
+		return new UpdateInfo<>(
+			Topic.ENRICH_MISSING_DATASET_IS_REFERENCED_BY,
+			highlightValue, source.getLeft(), target.getLeft(),
+			(p, rel) -> p.getDatasets().add(rel),
+			rel -> rel.getInstances().get(0).getUrl());
+	}
+}

dhp-workflows/dhp-broker-events/src/main/java/eu/dnetlib/dhp/broker/oa/matchers/EnrichMissingDatasetIsRelatedTo.java

@@ -0,0 +1,38 @@
+
+package eu.dnetlib.dhp.broker.oa.matchers;
+
+import java.util.List;
+
+import org.apache.commons.lang3.tuple.Pair;
+
+import eu.dnetlib.dhp.broker.model.Topic;
+import eu.dnetlib.dhp.broker.oa.util.UpdateInfo;
+import eu.dnetlib.dhp.schema.oaf.Dataset;
+import eu.dnetlib.dhp.schema.oaf.Result;
+
+public class EnrichMissingDatasetIsRelatedTo
+	extends UpdateMatcher<Pair<Result, List<Dataset>>, eu.dnetlib.broker.objects.Dataset> {
+
+	public EnrichMissingDatasetIsRelatedTo() {
+		super(true);
+	}
+
+	@Override
+	protected List<UpdateInfo<eu.dnetlib.broker.objects.Dataset>> findUpdates(final Pair<Result, List<Dataset>> source,
+		final Pair<Result, List<Dataset>> target) {
+		// TODO Auto-generated method stub
+		return null;
+	}
+
+	@Override
+	protected UpdateInfo<eu.dnetlib.broker.objects.Dataset> generateUpdateInfo(
+		final eu.dnetlib.broker.objects.Dataset highlightValue,
+		final Pair<Result, List<Dataset>> source,
+		final Pair<Result, List<Dataset>> target) {
+		return new UpdateInfo<>(
+			Topic.ENRICH_MISSING_DATASET_IS_RELATED_TO,
+			highlightValue, source.getLeft(), target.getLeft(),
+			(p, rel) -> p.getDatasets().add(rel),
+			rel -> rel.getInstances().get(0).getUrl());
+	}
+}

dhp-workflows/dhp-broker-events/src/main/java/eu/dnetlib/dhp/broker/oa/matchers/EnrichMissingDatasetIsSupplementedBy.java

@@ -0,0 +1,38 @@
+
+package eu.dnetlib.dhp.broker.oa.matchers;
+
+import java.util.List;
+
+import org.apache.commons.lang3.tuple.Pair;
+
+import eu.dnetlib.dhp.broker.model.Topic;
+import eu.dnetlib.dhp.broker.oa.util.UpdateInfo;
+import eu.dnetlib.dhp.schema.oaf.Dataset;
+import eu.dnetlib.dhp.schema.oaf.Result;
+
+public class EnrichMissingDatasetIsSupplementedBy
+	extends UpdateMatcher<Pair<Result, List<Dataset>>, eu.dnetlib.broker.objects.Dataset> {
+
+	public EnrichMissingDatasetIsSupplementedBy() {
+		super(true);
+	}
+
+	@Override
+	protected List<UpdateInfo<eu.dnetlib.broker.objects.Dataset>> findUpdates(final Pair<Result, List<Dataset>> source,
+		final Pair<Result, List<Dataset>> target) {
+		// TODO Auto-generated method stub
+		return null;
+	}
+
+	@Override
+	protected UpdateInfo<eu.dnetlib.broker.objects.Dataset> generateUpdateInfo(
+		final eu.dnetlib.broker.objects.Dataset highlightValue,
+		final Pair<Result, List<Dataset>> source,
+		final Pair<Result, List<Dataset>> target) {
+		return new UpdateInfo<>(
+			Topic.ENRICH_MISSING_DATASET_IS_SUPPLEMENTED_BY,
+			highlightValue, source.getLeft(), target.getLeft(),
+			(p, rel) -> p.getDatasets().add(rel),
+			rel -> rel.getInstances().get(0).getUrl());
+	}
+}

dhp-workflows/dhp-broker-events/src/main/java/eu/dnetlib/dhp/broker/oa/matchers/EnrichMissingDatasetIsSupplementedTo.java

@@ -0,0 +1,38 @@
+
+package eu.dnetlib.dhp.broker.oa.matchers;
+
+import java.util.List;
+
+import org.apache.commons.lang3.tuple.Pair;
+
+import eu.dnetlib.dhp.broker.model.Topic;
+import eu.dnetlib.dhp.broker.oa.util.UpdateInfo;
+import eu.dnetlib.dhp.schema.oaf.Dataset;
+import eu.dnetlib.dhp.schema.oaf.Result;
+
+public class EnrichMissingDatasetIsSupplementedTo
+	extends UpdateMatcher<Pair<Result, List<Dataset>>, eu.dnetlib.broker.objects.Dataset> {
+
+	public EnrichMissingDatasetIsSupplementedTo() {
+		super(true);
+	}
+
+	@Override
+	protected List<UpdateInfo<eu.dnetlib.broker.objects.Dataset>> findUpdates(final Pair<Result, List<Dataset>> source,
+		final Pair<Result, List<Dataset>> target) {
+		// TODO Auto-generated method stub
+		return null;
+	}
+
+	@Override
+	protected UpdateInfo<eu.dnetlib.broker.objects.Dataset> generateUpdateInfo(
+		final eu.dnetlib.broker.objects.Dataset highlightValue,
+		final Pair<Result, List<Dataset>> source,
+		final Pair<Result, List<Dataset>> target) {
+		return new UpdateInfo<>(
+			Topic.ENRICH_MISSING_DATASET_IS_SUPPLEMENTED_TO,
+			highlightValue, source.getLeft(), target.getLeft(),
+			(p, rel) -> p.getDatasets().add(rel),
+			rel -> rel.getInstances().get(0).getUrl());
+	}
+}

dhp-workflows/dhp-broker-events/src/main/java/eu/dnetlib/dhp/broker/oa/matchers/EnrichMissingDatasetReferences.java

@@ -0,0 +1,38 @@
+
+package eu.dnetlib.dhp.broker.oa.matchers;
+
+import java.util.List;
+
+import org.apache.commons.lang3.tuple.Pair;
+
+import eu.dnetlib.dhp.broker.model.Topic;
+import eu.dnetlib.dhp.broker.oa.util.UpdateInfo;
+import eu.dnetlib.dhp.schema.oaf.Dataset;
+import eu.dnetlib.dhp.schema.oaf.Result;
+
+public class EnrichMissingDatasetReferences
+	extends UpdateMatcher<Pair<Result, List<Dataset>>, eu.dnetlib.broker.objects.Dataset> {
+
+	public EnrichMissingDatasetReferences() {
+		super(true);
+	}
+
+	@Override
+	protected List<UpdateInfo<eu.dnetlib.broker.objects.Dataset>> findUpdates(final Pair<Result, List<Dataset>> source,
+		final Pair<Result, List<Dataset>> target) {
+		// TODO Auto-generated method stub
+		return null;
+	}
+
+	@Override
+	protected UpdateInfo<eu.dnetlib.broker.objects.Dataset> generateUpdateInfo(
+		final eu.dnetlib.broker.objects.Dataset highlightValue,
+		final Pair<Result, List<Dataset>> source,
+		final Pair<Result, List<Dataset>> target) {
+		return new UpdateInfo<>(
+			Topic.ENRICH_MISSING_DATASET_REFERENCES,
+			highlightValue, source.getLeft(), target.getLeft(),
+			(p, rel) -> p.getDatasets().add(rel),
+			rel -> rel.getInstances().get(0).getUrl());
+	}
+}

dhp-workflows/dhp-broker-events/src/main/java/eu/dnetlib/dhp/broker/oa/matchers/EnrichMissingOpenAccess.java

@@ -12,7 +12,7 @@ import eu.dnetlib.dhp.broker.oa.util.ConversionUtils;
 import eu.dnetlib.dhp.broker.oa.util.UpdateInfo;
 import eu.dnetlib.dhp.schema.oaf.Result;
 
-public class EnrichMissingOpenAccess extends UpdateMatcher<Instance> {
+public class EnrichMissingOpenAccess extends UpdateMatcher<Result, Instance> {
 
 	public EnrichMissingOpenAccess() {
 		super(true);

dhp-workflows/dhp-broker-events/src/main/java/eu/dnetlib/dhp/broker/oa/matchers/EnrichMissingPid.java

@@ -11,7 +11,7 @@ import eu.dnetlib.dhp.broker.oa.util.ConversionUtils;
 import eu.dnetlib.dhp.broker.oa.util.UpdateInfo;
 import eu.dnetlib.dhp.schema.oaf.Result;
 
-public class EnrichMissingPid extends UpdateMatcher<Pid> {
+public class EnrichMissingPid extends UpdateMatcher<Result, Pid> {
 
 	public EnrichMissingPid() {
 		super(true);

dhp-workflows/dhp-broker-events/src/main/java/eu/dnetlib/dhp/broker/oa/matchers/EnrichMissingProject.java

@@ -4,30 +4,35 @@ package eu.dnetlib.dhp.broker.oa.matchers;
 import java.util.Arrays;
 import java.util.List;
 
-import eu.dnetlib.broker.objects.Project;
+import org.apache.commons.lang3.tuple.Pair;
+
 import eu.dnetlib.dhp.broker.model.Topic;
 import eu.dnetlib.dhp.broker.oa.util.UpdateInfo;
+import eu.dnetlib.dhp.schema.oaf.Project;
 import eu.dnetlib.dhp.schema.oaf.Result;
 
-public class EnrichMissingProject extends UpdateMatcher<Project> {
+public class EnrichMissingProject
+	extends UpdateMatcher<Pair<Result, List<Project>>, eu.dnetlib.broker.objects.Project> {
 
 	public EnrichMissingProject() {
 		super(true);
 	}
 
 	@Override
-	protected List<UpdateInfo<Project>> findUpdates(final Result source, final Result target) {
+	protected List<UpdateInfo<eu.dnetlib.broker.objects.Project>> findUpdates(final Pair<Result, List<Project>> source,
-		// return Arrays.asList(new EnrichMissingAbstract("xxxxxxx", 0.9f));
+		final Pair<Result, List<Project>> target) {
+		// TODO
 		return Arrays.asList();
 	}
 
 	@Override
-	public UpdateInfo<Project> generateUpdateInfo(final Project highlightValue,
+	public UpdateInfo<eu.dnetlib.broker.objects.Project> generateUpdateInfo(
-		final Result source,
+		final eu.dnetlib.broker.objects.Project highlightValue,
-		final Result target) {
+		final Pair<Result, List<Project>> source,
+		final Pair<Result, List<Project>> target) {
 		return new UpdateInfo<>(
 			Topic.ENRICH_MISSING_PROJECT,
-			highlightValue, source, target,
+			highlightValue, source.getLeft(), target.getLeft(),
 			(p, prj) -> p.getProjects().add(prj),
 			prj -> prj.getFunder() + "::" + prj.getFundingProgram() + prj.getCode());
 	}

dhp-workflows/dhp-broker-events/src/main/java/eu/dnetlib/dhp/broker/oa/matchers/EnrichMissingPublicationDate.java

@@ -1,6 +1,7 @@
 
 package eu.dnetlib.dhp.broker.oa.matchers;
 
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
 
@@ -8,7 +9,7 @@ import eu.dnetlib.dhp.broker.model.Topic;
 import eu.dnetlib.dhp.broker.oa.util.UpdateInfo;
 import eu.dnetlib.dhp.schema.oaf.Result;
 
-public class EnrichMissingPublicationDate extends UpdateMatcher<String> {
+public class EnrichMissingPublicationDate extends UpdateMatcher<Result, String> {
 
 	public EnrichMissingPublicationDate() {
 		super(false);
@@ -16,12 +17,15 @@ public class EnrichMissingPublicationDate extends UpdateMatcher<String> {
 
 	@Override
 	protected List<UpdateInfo<String>> findUpdates(final Result source, final Result target) {
-		// return Arrays.asList(new EnrichMissingAbstract("xxxxxxx", 0.9f));
+		if (isMissing(target.getDateofacceptance()) && !isMissing(source.getDateofacceptance())) {
-		return Arrays.asList();
+			return Arrays.asList(generateUpdateInfo(source.getDateofacceptance().getValue(), source, target));
+		}
+		return new ArrayList<>();
 	}
 
 	@Override
-	public UpdateInfo<String> generateUpdateInfo(final String highlightValue, final Result source,
+	public UpdateInfo<String> generateUpdateInfo(final String highlightValue,
+		final Result source,
 		final Result target) {
 		return new UpdateInfo<>(
 			Topic.ENRICH_MISSING_PUBLICATION_DATE,

dhp-workflows/dhp-broker-events/src/main/java/eu/dnetlib/dhp/broker/oa/matchers/EnrichMissingPublicationIsReferencedBy.java

@@ -0,0 +1,42 @@
+
+package eu.dnetlib.dhp.broker.oa.matchers;
+
+import java.util.Arrays;
+import java.util.List;
+
+import org.apache.commons.lang3.tuple.Pair;
+
+import eu.dnetlib.dhp.broker.model.Topic;
+import eu.dnetlib.dhp.broker.oa.util.UpdateInfo;
+import eu.dnetlib.dhp.schema.oaf.Publication;
+import eu.dnetlib.dhp.schema.oaf.Result;
+
+public class EnrichMissingPublicationIsReferencedBy
+	extends UpdateMatcher<Pair<Result, List<Publication>>, eu.dnetlib.broker.objects.Publication> {
+
+	public EnrichMissingPublicationIsReferencedBy() {
+		super(true);
+	}
+
+	@Override
+	protected List<UpdateInfo<eu.dnetlib.broker.objects.Publication>> findUpdates(
+		final Pair<Result, List<Publication>> source,
+		final Pair<Result, List<Publication>> target) {
+		// TODO Auto-generated method stub
+		return Arrays.asList();
+	}
+
+	@Override
+	protected UpdateInfo<eu.dnetlib.broker.objects.Publication> generateUpdateInfo(
+		final eu.dnetlib.broker.objects.Publication highlightValue,
+		final Pair<Result, List<Publication>> source,
+		final Pair<Result, List<Publication>> target) {
+		return new UpdateInfo<>(
+			Topic.ENRICH_MISSING_PUBLICATION_IS_REFERENCED_BY,
+			highlightValue, source.getLeft(), target.getLeft(),
+			(p, rel) -> {
+			}, // p.getPublications().add(rel), //TODO available in the future release of dnet-openaire-broker-common
+			rel -> rel.getOriginalId());
+	}
+
+}

dhp-workflows/dhp-broker-events/src/main/java/eu/dnetlib/dhp/broker/oa/matchers/EnrichMissingPublicationIsRelatedTo.java

@@ -0,0 +1,42 @@
+
+package eu.dnetlib.dhp.broker.oa.matchers;
+
+import java.util.Arrays;
+import java.util.List;
+
+import org.apache.commons.lang3.tuple.Pair;
+
+import eu.dnetlib.dhp.broker.model.Topic;
+import eu.dnetlib.dhp.broker.oa.util.UpdateInfo;
+import eu.dnetlib.dhp.schema.oaf.Publication;
+import eu.dnetlib.dhp.schema.oaf.Result;
+
+public class EnrichMissingPublicationIsRelatedTo
+	extends UpdateMatcher<Pair<Result, List<Publication>>, eu.dnetlib.broker.objects.Publication> {
+
+	public EnrichMissingPublicationIsRelatedTo() {
+		super(true);
+	}
+
+	@Override
+	protected List<UpdateInfo<eu.dnetlib.broker.objects.Publication>> findUpdates(
+		final Pair<Result, List<Publication>> source,
+		final Pair<Result, List<Publication>> target) {
+		// TODO Auto-generated method stub
+		return Arrays.asList();
+	}
+
+	@Override
+	protected UpdateInfo<eu.dnetlib.broker.objects.Publication> generateUpdateInfo(
+		final eu.dnetlib.broker.objects.Publication highlightValue,
+		final Pair<Result, List<Publication>> source,
+		final Pair<Result, List<Publication>> target) {
+		return new UpdateInfo<>(
+			Topic.ENRICH_MISSING_PUBLICATION_IS_RELATED_TO,
+			highlightValue, source.getLeft(), target.getLeft(),
+			(p, rel) -> {
+			}, // p.getPublications().add(rel), //TODO available in the future release of dnet-openaire-broker-common
+			rel -> rel.getOriginalId());
+	}
+
+}

dhp-workflows/dhp-broker-events/src/main/java/eu/dnetlib/dhp/broker/oa/matchers/EnrichMissingPublicationIsSupplementedBy.java

@@ -0,0 +1,42 @@
+
+package eu.dnetlib.dhp.broker.oa.matchers;
+
+import java.util.Arrays;
+import java.util.List;
+
+import org.apache.commons.lang3.tuple.Pair;
+
+import eu.dnetlib.dhp.broker.model.Topic;
+import eu.dnetlib.dhp.broker.oa.util.UpdateInfo;
+import eu.dnetlib.dhp.schema.oaf.Publication;
+import eu.dnetlib.dhp.schema.oaf.Result;
+
+public class EnrichMissingPublicationIsSupplementedBy
+	extends UpdateMatcher<Pair<Result, List<Publication>>, eu.dnetlib.broker.objects.Publication> {
+
+	public EnrichMissingPublicationIsSupplementedBy() {
+		super(true);
+	}
+
+	@Override
+	protected List<UpdateInfo<eu.dnetlib.broker.objects.Publication>> findUpdates(
+		final Pair<Result, List<Publication>> source,
+		final Pair<Result, List<Publication>> target) {
+		// TODO Auto-generated method stub
+		return Arrays.asList();
+	}
+
+	@Override
+	protected UpdateInfo<eu.dnetlib.broker.objects.Publication> generateUpdateInfo(
+		final eu.dnetlib.broker.objects.Publication highlightValue,
+		final Pair<Result, List<Publication>> source,
+		final Pair<Result, List<Publication>> target) {
+		return new UpdateInfo<>(
+			Topic.ENRICH_MISSING_PUBLICATION_IS_SUPPLEMENTED_BY,
+			highlightValue, source.getLeft(), target.getLeft(),
+			(p, rel) -> {
+			}, // p.getPublications().add(rel), //TODO available in the future release of dnet-openaire-broker-common
+			rel -> rel.getOriginalId());
+	}
+
+}

dhp-workflows/dhp-broker-events/src/main/java/eu/dnetlib/dhp/broker/oa/matchers/EnrichMissingPublicationIsSupplementedTo.java

@@ -0,0 +1,42 @@
+
+package eu.dnetlib.dhp.broker.oa.matchers;
+
+import java.util.Arrays;
+import java.util.List;
+
+import org.apache.commons.lang3.tuple.Pair;
+
+import eu.dnetlib.dhp.broker.model.Topic;
+import eu.dnetlib.dhp.broker.oa.util.UpdateInfo;
+import eu.dnetlib.dhp.schema.oaf.Publication;
+import eu.dnetlib.dhp.schema.oaf.Result;
+
+public class EnrichMissingPublicationIsSupplementedTo
+	extends UpdateMatcher<Pair<Result, List<Publication>>, eu.dnetlib.broker.objects.Publication> {
+
+	public EnrichMissingPublicationIsSupplementedTo() {
+		super(true);
+	}
+
+	@Override
+	protected List<UpdateInfo<eu.dnetlib.broker.objects.Publication>> findUpdates(
+		final Pair<Result, List<Publication>> source,
+		final Pair<Result, List<Publication>> target) {
+		// TODO Auto-generated method stub
+		return Arrays.asList();
+	}
+
+	@Override
+	protected UpdateInfo<eu.dnetlib.broker.objects.Publication> generateUpdateInfo(
+		final eu.dnetlib.broker.objects.Publication highlightValue,
+		final Pair<Result, List<Publication>> source,
+		final Pair<Result, List<Publication>> target) {
+		return new UpdateInfo<>(
+			Topic.ENRICH_MISSING_PUBLICATION_IS_SUPPLEMENTED_TO,
+			highlightValue, source.getLeft(), target.getLeft(),
+			(p, rel) -> {
+			}, // p.getPublications().add(rel), //TODO available in the future release of dnet-openaire-broker-common
+			rel -> rel.getOriginalId());
+	}
+
+}

dhp-workflows/dhp-broker-events/src/main/java/eu/dnetlib/dhp/broker/oa/matchers/EnrichMissingPublicationReferences.java

@@ -0,0 +1,42 @@
+
+package eu.dnetlib.dhp.broker.oa.matchers;
+
+import java.util.Arrays;
+import java.util.List;
+
+import org.apache.commons.lang3.tuple.Pair;
+
+import eu.dnetlib.dhp.broker.model.Topic;
+import eu.dnetlib.dhp.broker.oa.util.UpdateInfo;
+import eu.dnetlib.dhp.schema.oaf.Publication;
+import eu.dnetlib.dhp.schema.oaf.Result;
+
+public class EnrichMissingPublicationReferences
+	extends UpdateMatcher<Pair<Result, List<Publication>>, eu.dnetlib.broker.objects.Publication> {
+
+	public EnrichMissingPublicationReferences() {
+		super(true);
+	}
+
+	@Override
+	protected List<UpdateInfo<eu.dnetlib.broker.objects.Publication>> findUpdates(
+		final Pair<Result, List<Publication>> source,
+		final Pair<Result, List<Publication>> target) {
+		// TODO Auto-generated method stub
+		return Arrays.asList();
+	}
+
+	@Override
+	protected UpdateInfo<eu.dnetlib.broker.objects.Publication> generateUpdateInfo(
+		final eu.dnetlib.broker.objects.Publication highlightValue,
+		final Pair<Result, List<Publication>> source,
+		final Pair<Result, List<Publication>> target) {
+		return new UpdateInfo<>(
+			Topic.ENRICH_MISSING_PUBLICATION_REFERENCES,
+			highlightValue, source.getLeft(), target.getLeft(),
+			(p, rel) -> {
+			}, // p.getPublications().add(rel), //TODO available in the future release of dnet-openaire-broker-common
+			rel -> rel.getOriginalId());
+	}
+
+}

dhp-workflows/dhp-broker-events/src/main/java/eu/dnetlib/dhp/broker/oa/matchers/EnrichMissingSoftware.java

@@ -0,0 +1,41 @@
+
+package eu.dnetlib.dhp.broker.oa.matchers;
+
+import java.util.Arrays;
+import java.util.List;
+
+import org.apache.commons.lang3.tuple.Pair;
+
+import eu.dnetlib.dhp.broker.model.Topic;
+import eu.dnetlib.dhp.broker.oa.util.UpdateInfo;
+import eu.dnetlib.dhp.schema.oaf.Result;
+import eu.dnetlib.dhp.schema.oaf.Software;
+
+public class EnrichMissingSoftware
+	extends UpdateMatcher<Pair<Result, List<Software>>, eu.dnetlib.broker.objects.Software> {
+
+	public EnrichMissingSoftware() {
+		super(true);
+	}
+
+	@Override
+	protected List<UpdateInfo<eu.dnetlib.broker.objects.Software>> findUpdates(
+		final Pair<Result, List<Software>> source,
+		final Pair<Result, List<Software>> target) {
+		// TODO
+		return Arrays.asList();
+	}
+
+	@Override
+	public UpdateInfo<eu.dnetlib.broker.objects.Software> generateUpdateInfo(
+		final eu.dnetlib.broker.objects.Software highlightValue,
+		final Pair<Result, List<Software>> source,
+		final Pair<Result, List<Software>> target) {
+		return new UpdateInfo<>(
+			Topic.ENRICH_MISSING_SOFTWARE,
+			highlightValue, source.getLeft(), target.getLeft(),
+			(p, s) -> p.getSoftwares().add(s),
+			s -> s.getName());
+	}
+
+}

dhp-workflows/dhp-broker-events/src/main/java/eu/dnetlib/dhp/broker/oa/matchers/EnrichMissingSubject.java

@@ -14,7 +14,7 @@ import eu.dnetlib.dhp.schema.oaf.Qualifier;
 import eu.dnetlib.dhp.schema.oaf.Result;
 import eu.dnetlib.dhp.schema.oaf.StructuredProperty;
 
-public class EnrichMissingSubject extends UpdateMatcher<Pair<String, String>> {
+public class EnrichMissingSubject extends UpdateMatcher<Result, Pair<String, String>> {
 
 	public EnrichMissingSubject() {
 		super(true);

dhp-workflows/dhp-broker-events/src/main/java/eu/dnetlib/dhp/broker/oa/matchers/EnrichMoreOpenAccess.java

@@ -12,7 +12,7 @@ import eu.dnetlib.dhp.broker.oa.util.ConversionUtils;
 import eu.dnetlib.dhp.broker.oa.util.UpdateInfo;
 import eu.dnetlib.dhp.schema.oaf.Result;
 
-public class EnrichMoreOpenAccess extends UpdateMatcher<Instance> {
+public class EnrichMoreOpenAccess extends UpdateMatcher<Result, Instance> {
 
 	public EnrichMoreOpenAccess() {
 		super(true);

dhp-workflows/dhp-broker-events/src/main/java/eu/dnetlib/dhp/broker/oa/matchers/EnrichMorePid.java

@@ -11,7 +11,7 @@ import eu.dnetlib.dhp.broker.oa.util.ConversionUtils;
 import eu.dnetlib.dhp.broker.oa.util.UpdateInfo;
 import eu.dnetlib.dhp.schema.oaf.Result;
 
-public class EnrichMorePid extends UpdateMatcher<Pid> {
+public class EnrichMorePid extends UpdateMatcher<Result, Pid> {
 
 	public EnrichMorePid() {
 		super(true);

dhp-workflows/dhp-broker-events/src/main/java/eu/dnetlib/dhp/broker/oa/matchers/EnrichMoreProject.java

@@ -0,0 +1,39 @@
+
+package eu.dnetlib.dhp.broker.oa.matchers;
+
+import java.util.Arrays;
+import java.util.List;
+
+import org.apache.commons.lang3.tuple.Pair;
+
+import eu.dnetlib.dhp.broker.model.Topic;
+import eu.dnetlib.dhp.broker.oa.util.UpdateInfo;
+import eu.dnetlib.dhp.schema.oaf.Project;
+import eu.dnetlib.dhp.schema.oaf.Result;
+
+public class EnrichMoreProject extends UpdateMatcher<Pair<Result, List<Project>>, eu.dnetlib.broker.objects.Project> {
+
+	public EnrichMoreProject() {
+		super(true);
+	}
+
+	@Override
+	protected List<UpdateInfo<eu.dnetlib.broker.objects.Project>> findUpdates(final Pair<Result, List<Project>> source,
+		final Pair<Result, List<Project>> target) {
+		// TODO
+		return Arrays.asList();
+	}
+
+	@Override
+	public UpdateInfo<eu.dnetlib.broker.objects.Project> generateUpdateInfo(
+		final eu.dnetlib.broker.objects.Project highlightValue,
+		final Pair<Result, List<Project>> source,
+		final Pair<Result, List<Project>> target) {
+		return new UpdateInfo<>(
+			Topic.ENRICH_MORE_PROJECT,
+			highlightValue, source.getLeft(), target.getLeft(),
+			(p, prj) -> p.getProjects().add(prj),
+			prj -> prj.getFunder() + "::" + prj.getFundingProgram() + prj.getCode());
+	}
+
+}

dhp-workflows/dhp-broker-events/src/main/java/eu/dnetlib/dhp/broker/oa/matchers/EnrichMoreSoftware.java

@@ -0,0 +1,41 @@
+
+package eu.dnetlib.dhp.broker.oa.matchers;
+
+import java.util.Arrays;
+import java.util.List;
+
+import org.apache.commons.lang3.tuple.Pair;
+
+import eu.dnetlib.dhp.broker.model.Topic;
+import eu.dnetlib.dhp.broker.oa.util.UpdateInfo;
+import eu.dnetlib.dhp.schema.oaf.Result;
+import eu.dnetlib.dhp.schema.oaf.Software;
+
+public class EnrichMoreSoftware
+	extends UpdateMatcher<Pair<Result, List<Software>>, eu.dnetlib.broker.objects.Software> {
+
+	public EnrichMoreSoftware() {
+		super(true);
+	}
+
+	@Override
+	protected List<UpdateInfo<eu.dnetlib.broker.objects.Software>> findUpdates(
+		final Pair<Result, List<Software>> source,
+		final Pair<Result, List<Software>> target) {
+		// TODO
+		return Arrays.asList();
+	}
+
+	@Override
+	public UpdateInfo<eu.dnetlib.broker.objects.Software> generateUpdateInfo(
+		final eu.dnetlib.broker.objects.Software highlightValue,
+		final Pair<Result, List<Software>> source,
+		final Pair<Result, List<Software>> target) {
+		return new UpdateInfo<>(
+			Topic.ENRICH_MORE_SOFTWARE,
+			highlightValue, source.getLeft(), target.getLeft(),
+			(p, s) -> p.getSoftwares().add(s),
+			s -> s.getName());
+	}
+
+}

dhp-workflows/dhp-broker-events/src/main/java/eu/dnetlib/dhp/broker/oa/matchers/EnrichMoreSubject.java

@@ -12,7 +12,7 @@ import eu.dnetlib.dhp.broker.oa.util.ConversionUtils;
 import eu.dnetlib.dhp.broker.oa.util.UpdateInfo;
 import eu.dnetlib.dhp.schema.oaf.Result;
 
-public class EnrichMoreSubject extends UpdateMatcher<Pair<String, String>> {
+public class EnrichMoreSubject extends UpdateMatcher<Result, Pair<String, String>> {
 
 	public EnrichMoreSubject() {
 		super(true);

dhp-workflows/dhp-broker-events/src/main/java/eu/dnetlib/dhp/broker/oa/matchers/UpdateMatcher.java

@@ -12,9 +12,8 @@ import org.apache.commons.lang3.StringUtils;
 
 import eu.dnetlib.dhp.broker.oa.util.UpdateInfo;
 import eu.dnetlib.dhp.schema.oaf.Field;
-import eu.dnetlib.dhp.schema.oaf.Result;
 
-public abstract class UpdateMatcher<T> {
+public abstract class UpdateMatcher<K, T> {
 
 	private final boolean multipleUpdate;
 
@@ -22,11 +21,11 @@ public abstract class UpdateMatcher<T> {
 		this.multipleUpdate = multipleUpdate;
 	}
 
-	public Collection<UpdateInfo<T>> searchUpdatesForRecord(final Result res, final Result... others) {
+	public Collection<UpdateInfo<T>> searchUpdatesForRecord(final K res, final Collection<K> others) {
 
 		final Map<String, UpdateInfo<T>> infoMap = new HashMap<>();
 
-		for (final Result source : others) {
+		for (final K source : others) {
 			if (source != res) {
 				for (final UpdateInfo<T> info : findUpdates(source, res)) {
 					final String s = DigestUtils.md5Hex(info.getHighlightValueAsString());
@@ -52,13 +51,18 @@ public abstract class UpdateMatcher<T> {
 		}
 	}
 
-	protected abstract List<UpdateInfo<T>> findUpdates(Result source, Result target);
+	protected abstract List<UpdateInfo<T>> findUpdates(K source, K target);
 
-	protected abstract UpdateInfo<T> generateUpdateInfo(final T highlightValue, final Result source,
+	protected abstract UpdateInfo<T> generateUpdateInfo(final T highlightValue,
-		final Result target);
+		final K source,
+		final K target);
 
 	protected static boolean isMissing(final List<Field<String>> list) {
 		return list == null || list.isEmpty() || StringUtils.isBlank(list.get(0).getValue());
 	}
 
+	protected boolean isMissing(final Field<String> field) {
+		return field == null || StringUtils.isBlank(field.getValue());
+	}
+
 }

dhp-workflows/dhp-broker-events/src/main/java/eu/dnetlib/dhp/broker/oa/util/BrokerConstants.java

@@ -4,4 +4,6 @@ package eu.dnetlib.dhp.broker.oa.util;
 public class BrokerConstants {
 
 	public final static String OPEN_ACCESS = "OPEN";
+	public final static String IS_MERGED_IN_CLASS = "isMergedIn";
+
 }

dhp-workflows/dhp-dedup-openaire/src/main/java/eu/dnetlib/dhp/oa/dedup/AuthorMerger.java

@@ -0,0 +1,159 @@
+package eu.dnetlib.dhp.oa.dedup;
+
+import com.wcohen.ss.JaroWinkler;
+import eu.dnetlib.dhp.schema.oaf.Author;
+import eu.dnetlib.dhp.schema.oaf.StructuredProperty;
+import eu.dnetlib.pace.model.Person;
+import org.apache.commons.lang3.StringUtils;
+import scala.Tuple2;
+
+import java.text.Normalizer;
+import java.util.*;
+import java.util.stream.Collectors;
+
+public class AuthorMerger {
+
+    private static final Double THRESHOLD = 0.95;
+
+    public static List<Author> merge(List<List<Author>> authors){
+
+        authors.sort(new Comparator<List<Author>>() {
+            @Override
+            public int compare(List<Author> o1, List<Author> o2) {
+                return -Integer.compare(countAuthorsPids(o1), countAuthorsPids(o2));
+            }
+        });
+
+        List<Author> author = new ArrayList<>();
+
+        for(List<Author> a : authors){
+            author = mergeAuthor(author, a);
+        }
+
+        return author;
+
+    }
+
+    public static List<Author> mergeAuthor(final List<Author> a, final List<Author> b) {
+        int pa = countAuthorsPids(a);
+        int pb = countAuthorsPids(b);
+        List<Author> base, enrich;
+        int sa = authorsSize(a);
+        int sb = authorsSize(b);
+
+        if (pa == pb) {
+            base = sa > sb ? a : b;
+            enrich = sa > sb ? b : a;
+        } else {
+            base = pa > pb ? a : b;
+            enrich = pa > pb ? b : a;
+        }
+        enrichPidFromList(base, enrich);
+        return base;
+    }
+
+    private static void enrichPidFromList(List<Author> base, List<Author> enrich) {
+        if (base == null || enrich == null)
+            return;
+        final Map<String, Author> basePidAuthorMap = base
+                .stream()
+                .filter(a -> a.getPid() != null && a.getPid().size() > 0)
+                .flatMap(
+                        a -> a
+                                .getPid()
+                                .stream()
+                                .map(p -> new Tuple2<>(pidToComparableString(p), a)))
+                .collect(Collectors.toMap(Tuple2::_1, Tuple2::_2, (x1, x2) -> x1));
+
+        final List<Tuple2<StructuredProperty, Author>> pidToEnrich = enrich
+                .stream()
+                .filter(a -> a.getPid() != null && a.getPid().size() > 0)
+                .flatMap(
+                        a -> a
+                                .getPid()
+                                .stream()
+                                .filter(p -> !basePidAuthorMap.containsKey(pidToComparableString(p)))
+                                .map(p -> new Tuple2<>(p, a)))
+                .collect(Collectors.toList());
+
+        pidToEnrich
+                .forEach(
+                        a -> {
+                            Optional<Tuple2<Double, Author>> simAuthor = base
+                                    .stream()
+                                    .map(ba -> new Tuple2<>(sim(ba, a._2()), ba))
+                                    .max(Comparator.comparing(Tuple2::_1));
+                            if (simAuthor.isPresent() && simAuthor.get()._1() > THRESHOLD) {
+                                Author r = simAuthor.get()._2();
+                                if (r.getPid() == null) {
+                                    r.setPid(new ArrayList<>());
+                                }
+                                r.getPid().add(a._1());
+                            }
+                        });
+    }
+
+    public static String pidToComparableString(StructuredProperty pid){
+        return (pid.getQualifier()!=null? pid.getQualifier().getClassid()!=null?pid.getQualifier().getClassid().toLowerCase():"":"") + (pid.getValue()!=null? pid.getValue().toLowerCase():"");
+    }
+
+    public static int countAuthorsPids(List<Author> authors) {
+        if (authors == null)
+            return 0;
+
+        return (int) authors.stream().filter(AuthorMerger::hasPid).count();
+    }
+
+    private static int authorsSize(List<Author> authors) {
+        if (authors == null)
+            return 0;
+        return authors.size();
+    }
+
+    private static Double sim(Author a, Author b) {
+
+        final Person pa = parse(a);
+        final Person pb = parse(b);
+
+        if (pa.isAccurate() & pb.isAccurate()) {
+            return new JaroWinkler()
+                    .score(normalize(pa.getSurnameString()), normalize(pb.getSurnameString()));
+        } else {
+            return new JaroWinkler()
+                    .score(normalize(pa.getNormalisedFullname()), normalize(pb.getNormalisedFullname()));
+        }
+    }
+
+    private static boolean hasPid(Author a) {
+        if (a == null || a.getPid() == null || a.getPid().size() == 0)
+            return false;
+        return a.getPid().stream().anyMatch(p -> p != null && StringUtils.isNotBlank(p.getValue()));
+    }
+
+    private static Person parse(Author author) {
+        if (StringUtils.isNotBlank(author.getSurname())) {
+            return new Person(author.getSurname() + ", " + author.getName(), false);
+        } else {
+            return new Person(author.getFullname(), false);
+        }
+    }
+
+    private static String normalize(final String s) {
+        return nfd(s)
+                .toLowerCase()
+                // do not compact the regexes in a single expression, would cause StackOverflowError
+                // in case
+                // of large input strings
+                .replaceAll("(\\W)+", " ")
+                .replaceAll("(\\p{InCombiningDiacriticalMarks})+", " ")
+                .replaceAll("(\\p{Punct})+", " ")
+                .replaceAll("(\\d)+", " ")
+                .replaceAll("(\\n)+", " ")
+                .trim();
+    }
+
+    private static String nfd(final String s) {
+        return Normalizer.normalize(s, Normalizer.Form.NFD);
+    }
+
+}

dhp-workflows/dhp-dedup-openaire/src/main/java/eu/dnetlib/dhp/oa/dedup/DedupRecordFactory.java

@@ -1,8 +1,9 @@
-
 package eu.dnetlib.dhp.oa.dedup;
 
+import java.io.Serializable;
 import java.util.Collection;
 import java.util.Iterator;
+import java.util.List;
 
 import org.apache.spark.api.java.function.MapFunction;
 import org.apache.spark.api.java.function.MapGroupsFunction;
@@ -67,16 +68,18 @@ public class DedupRecordFactory {
 				(MapFunction<Tuple2<String, T>, String>) entity -> entity._1(), Encoders.STRING())
 			.mapGroups(
 				(MapGroupsFunction<String, Tuple2<String, T>, T>) (key,
-					values) -> entityMerger(key, values, ts, dataInfo),
+					values) -> entityMerger(key, values, ts, dataInfo, clazz),
 				Encoders.bean(clazz));
 	}
 
-	private static <T extends OafEntity> T entityMerger(
+	public static <T extends OafEntity> T entityMerger(
-		String id, Iterator<Tuple2<String, T>> entities, long ts, DataInfo dataInfo) {
+		String id, Iterator<Tuple2<String, T>> entities, long ts, DataInfo dataInfo, Class<T> clazz) throws IllegalAccessException, InstantiationException {
 
-		T entity = entities.next()._2();
+		T entity = clazz.newInstance();
 
 		final Collection<String> dates = Lists.newArrayList();
+		final List<List<Author>> authors = Lists.newArrayList();
+
 		entities
 			.forEachRemaining(
 				t -> {
@@ -84,17 +87,17 @@ public class DedupRecordFactory {
 					entity.mergeFrom(duplicate);
 					if (ModelSupport.isSubClass(duplicate, Result.class)) {
 						Result r1 = (Result) duplicate;
-						Result er = (Result) entity;
+						if (r1.getAuthor() != null && r1.getAuthor().size()>0)
-						er.setAuthor(DedupUtility.mergeAuthor(er.getAuthor(), r1.getAuthor()));
+							authors.add(r1.getAuthor());
-
+						if (r1.getDateofacceptance() != null)
-						if (r1.getDateofacceptance() != null) {
 							dates.add(r1.getDateofacceptance().getValue());
 					}
-					}
 				});
 
+		//set authors and date
 		if (ModelSupport.isSubClass(entity, Result.class)) {
 			((Result) entity).setDateofacceptance(DatePicker.pick(dates));
+			((Result) entity).setAuthor(AuthorMerger.merge(authors));
 		}
 
 		entity.setId(id);

dhp-workflows/dhp-dedup-openaire/src/main/java/eu/dnetlib/dhp/oa/dedup/DedupUtility.java

@@ -32,7 +32,6 @@ import eu.dnetlib.pace.model.Person;
 import scala.Tuple2;
 
 public class DedupUtility {
-	private static final Double THRESHOLD = 0.95;
 
 	public static Map<String, LongAccumulator> constructAccumulator(
 		final DedupConfig dedupConf, final SparkContext context) {
@@ -82,61 +81,6 @@ public class DedupUtility {
 		}
 	}
 
-	public static List<Author> mergeAuthor(final List<Author> a, final List<Author> b) {
-		int pa = countAuthorsPids(a);
-		int pb = countAuthorsPids(b);
-		List<Author> base, enrich;
-		int sa = authorsSize(a);
-		int sb = authorsSize(b);
-
-		if (pa == pb) {
-			base = sa > sb ? a : b;
-			enrich = sa > sb ? b : a;
-		} else {
-			base = pa > pb ? a : b;
-			enrich = pa > pb ? b : a;
-		}
-		enrichPidFromList(base, enrich);
-		return base;
-	}
-
-	private static void enrichPidFromList(List<Author> base, List<Author> enrich) {
-		if (base == null || enrich == null)
-			return;
-		final Map<String, Author> basePidAuthorMap = base
-			.stream()
-			.filter(a -> a.getPid() != null && a.getPid().size() > 0)
-			.flatMap(a -> a.getPid().stream().map(p -> new Tuple2<>(p.toComparableString(), a)))
-			.collect(Collectors.toMap(Tuple2::_1, Tuple2::_2, (x1, x2) -> x1));
-
-		final List<Tuple2<StructuredProperty, Author>> pidToEnrich = enrich
-			.stream()
-			.filter(a -> a.getPid() != null && a.getPid().size() > 0)
-			.flatMap(
-				a -> a
-					.getPid()
-					.stream()
-					.filter(p -> !basePidAuthorMap.containsKey(p.toComparableString()))
-					.map(p -> new Tuple2<>(p, a)))
-			.collect(Collectors.toList());
-
-		pidToEnrich
-			.forEach(
-				a -> {
-					Optional<Tuple2<Double, Author>> simAuhtor = base
-						.stream()
-						.map(ba -> new Tuple2<>(sim(ba, a._2()), ba))
-						.max(Comparator.comparing(Tuple2::_1));
-					if (simAuhtor.isPresent() && simAuhtor.get()._1() > THRESHOLD) {
-						Author r = simAuhtor.get()._2();
-						if (r.getPid() == null) {
-							r.setPid(new ArrayList<>());
-						}
-						r.getPid().add(a._1());
-					}
-				});
-	}
-
 	public static String createDedupRecordPath(
 		final String basePath, final String actionSetId, final String entityType) {
 		return String.format("%s/%s/%s_deduprecord", basePath, actionSetId, entityType);
@@ -156,65 +100,6 @@ public class DedupUtility {
 		return String.format("%s/%s/%s_mergerel", basePath, actionSetId, entityType);
 	}
 
-	private static Double sim(Author a, Author b) {
-
-		final Person pa = parse(a);
-		final Person pb = parse(b);
-
-		if (pa.isAccurate() & pb.isAccurate()) {
-			return new JaroWinkler()
-				.score(normalize(pa.getSurnameString()), normalize(pb.getSurnameString()));
-		} else {
-			return new JaroWinkler()
-				.score(normalize(pa.getNormalisedFullname()), normalize(pb.getNormalisedFullname()));
-		}
-	}
-
-	private static String normalize(final String s) {
-		return nfd(s)
-			.toLowerCase()
-			// do not compact the regexes in a single expression, would cause StackOverflowError
-			// in case
-			// of large input strings
-			.replaceAll("(\\W)+", " ")
-			.replaceAll("(\\p{InCombiningDiacriticalMarks})+", " ")
-			.replaceAll("(\\p{Punct})+", " ")
-			.replaceAll("(\\d)+", " ")
-			.replaceAll("(\\n)+", " ")
-			.trim();
-	}
-
-	private static String nfd(final String s) {
-		return Normalizer.normalize(s, Normalizer.Form.NFD);
-	}
-
-	private static Person parse(Author author) {
-		if (StringUtils.isNotBlank(author.getSurname())) {
-			return new Person(author.getSurname() + ", " + author.getName(), false);
-		} else {
-			return new Person(author.getFullname(), false);
-		}
-	}
-
-	private static int countAuthorsPids(List<Author> authors) {
-		if (authors == null)
-			return 0;
-
-		return (int) authors.stream().filter(DedupUtility::hasPid).count();
-	}
-
-	private static int authorsSize(List<Author> authors) {
-		if (authors == null)
-			return 0;
-		return authors.size();
-	}
-
-	private static boolean hasPid(Author a) {
-		if (a == null || a.getPid() == null || a.getPid().size() == 0)
-			return false;
-		return a.getPid().stream().anyMatch(p -> p != null && StringUtils.isNotBlank(p.getValue()));
-	}
-
 	public static List<DedupConfig> getConfigurations(String isLookUpUrl, String orchestrator)
 		throws ISLookUpException, DocumentException {
 		final ISLookUpService isLookUpService = ISLookupClientFactory.getLookUpService(isLookUpUrl);

dhp-workflows/dhp-dedup-openaire/src/test/java/eu/dnetlib/dhp/oa/dedup/EntityMergerTest.java

@@ -0,0 +1,138 @@
+package eu.dnetlib.dhp.oa.dedup;
+
+import java.io.BufferedReader;
+import java.io.FileReader;
+import java.io.IOException;
+import java.io.Serializable;
+import java.nio.file.Paths;
+import java.util.*;
+
+import eu.dnetlib.dhp.schema.oaf.*;
+import eu.dnetlib.pace.util.MapDocumentUtil;
+import org.codehaus.jackson.map.ObjectMapper;
+import org.junit.jupiter.api.BeforeEach;
+
+import org.junit.jupiter.api.Test;
+import scala.Tuple2;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+public class EntityMergerTest implements Serializable {
+
+	List<Tuple2<String, Publication>> publications;
+
+	String testEntityBasePath;
+	DataInfo dataInfo;
+	String dedupId = "dedup_id";
+	Publication pub_top;
+
+	@BeforeEach
+	public void setUp() throws Exception {
+
+		testEntityBasePath = Paths
+				.get(SparkDedupTest.class.getResource("/eu/dnetlib/dhp/dedup/json").toURI())
+				.toFile()
+				.getAbsolutePath();
+
+		publications = readSample(testEntityBasePath + "/publication_merge.json", Publication.class);
+
+		pub_top = getTopPub(publications);
+
+		dataInfo = setDI();
+
+	}
+
+	@Test
+	public void publicationMergerTest() throws InstantiationException, IllegalAccessException {
+
+		Publication pub_merged = DedupRecordFactory.entityMerger(dedupId, publications.iterator(), 0, dataInfo, Publication.class);
+
+		assertEquals(dedupId, pub_merged.getId());
+
+		assertEquals(pub_merged.getJournal(), pub_top.getJournal());
+		assertEquals(pub_merged.getBestaccessright(), pub_top.getBestaccessright());
+		assertEquals(pub_merged.getResulttype(), pub_top.getResulttype());
+		assertEquals(pub_merged.getLanguage(), pub_merged.getLanguage());
+		assertEquals(pub_merged.getPublisher(), pub_top.getPublisher());
+		assertEquals(pub_merged.getEmbargoenddate(), pub_top.getEmbargoenddate());
+		assertEquals(pub_merged.getResourcetype().getClassid(), "0004");
+		assertEquals(pub_merged.getDateoftransformation(), pub_top.getDateoftransformation());
+		assertEquals(pub_merged.getOaiprovenance(), pub_top.getOaiprovenance());
+		assertEquals(pub_merged.getDateofcollection(), pub_top.getDateofcollection());
+		assertEquals(pub_merged.getInstance().size(),3);
+		assertEquals(pub_merged.getCountry().size(), 2);
+		assertEquals(pub_merged.getSubject().size(), 0);
+		assertEquals(pub_merged.getTitle().size(), 2);
+		assertEquals(pub_merged.getRelevantdate().size(),0);
+		assertEquals(pub_merged.getDescription().size(),0);
+		assertEquals(pub_merged.getSource().size(),0);
+		assertEquals(pub_merged.getFulltext().size(),0);
+		assertEquals(pub_merged.getFormat().size(),0);
+		assertEquals(pub_merged.getContributor().size(),0);
+		assertEquals(pub_merged.getCoverage().size(),0);
+		assertEquals(pub_merged.getContext().size(),0);
+		assertEquals(pub_merged.getExternalReference().size(),0);
+		assertEquals(pub_merged.getOriginalId().size(),3);
+		assertEquals(pub_merged.getCollectedfrom().size(),3);
+		assertEquals(pub_merged.getPid().size(),1);
+		assertEquals(pub_merged.getExtraInfo().size(),0);
+
+		//verify datainfo
+		assertEquals(pub_merged.getDataInfo(), dataInfo);
+
+		//verify datepicker
+		assertEquals(pub_merged.getDateofacceptance().getValue(), "2018-09-30");
+
+		//verify authors
+		assertEquals(pub_merged.getAuthor().size(), 9);
+		assertEquals(AuthorMerger.countAuthorsPids(pub_merged.getAuthor()), 4);
+	}
+
+	public DataInfo setDI(){
+		DataInfo dataInfo = new DataInfo();
+		dataInfo.setTrust("0.9");
+		dataInfo.setDeletedbyinference(false);
+		dataInfo.setInferenceprovenance("testing");
+		dataInfo.setInferred(true);
+		return dataInfo;
+	}
+
+	public Publication getTopPub(List<Tuple2<String, Publication>> publications){
+
+		Double maxTrust = 0.0;
+		Publication maxPub = new Publication();
+		for (Tuple2<String, Publication> publication : publications) {
+			Double pubTrust = Double.parseDouble(publication._2().getDataInfo().getTrust());
+			if(pubTrust > maxTrust){
+				maxTrust = pubTrust;
+				maxPub = publication._2();
+			}
+		}
+		return maxPub;
+	}
+
+	public <T> List<Tuple2<String, T>> readSample(String path, Class<T> clazz) {
+		List<Tuple2<String, T>> res = new ArrayList<>();
+		BufferedReader reader;
+		try {
+			reader = new BufferedReader(new FileReader(path));
+			String line = reader.readLine();
+			while (line != null) {
+				res.add(
+						new Tuple2<>(
+								MapDocumentUtil.getJPathString("$.id", line),
+								new ObjectMapper().readValue(line, clazz))
+				);
+				// read next line
+				line = reader.readLine();
+			}
+			reader.close();
+		} catch (IOException e) {
+			e.printStackTrace();
+		}
+
+		return res;
+	}
+
+
+}

dhp-workflows/dhp-dedup-openaire/src/test/java/eu/dnetlib/dhp/oa/dedup/MergeAuthorTest.java

@@ -1,54 +0,0 @@
-
-package eu.dnetlib.dhp.oa.dedup;
-
-import java.io.IOException;
-import java.util.Arrays;
-import java.util.List;
-import java.util.stream.Collectors;
-
-import org.apache.commons.io.IOUtils;
-import org.codehaus.jackson.map.ObjectMapper;
-import org.junit.jupiter.api.BeforeEach;
-
-import eu.dnetlib.dhp.schema.oaf.Publication;
-
-public class MergeAuthorTest {
-
-	private List<Publication> publicationsToMerge;
-	private final ObjectMapper mapper = new ObjectMapper();
-
-	@BeforeEach
-	public void setUp() throws Exception {
-		final String json = IOUtils
-			.toString(
-				this.getClass().getResourceAsStream("/eu/dnetlib/dhp/dedup/json/authors_merge.json"));
-
-		publicationsToMerge = Arrays
-			.asList(json.split("\n"))
-			.stream()
-			.map(
-				s -> {
-					try {
-						return mapper.readValue(s, Publication.class);
-					} catch (IOException e) {
-						throw new RuntimeException(e);
-					}
-				})
-			.collect(Collectors.toList());
-	}
-
-	// FIX ME Michele DB this tests doesn't work
-	// @Test
-	public void test() throws Exception {
-		Publication dedup = new Publication();
-
-		publicationsToMerge
-			.forEach(
-				p -> {
-					dedup.mergeFrom(p);
-					dedup.setAuthor(DedupUtility.mergeAuthor(dedup.getAuthor(), p.getAuthor()));
-				});
-
-		System.out.println(mapper.writeValueAsString(dedup));
-	}
-}

dhp-workflows/dhp-graph-mapper/src/main/resources/eu/dnetlib/dhp/oa/graph/sql/queryOrganizations.sql

@@ -22,12 +22,13 @@ SELECT
 		''                                                        AS inferenceprovenance,
 		d.id                                                      AS collectedfromid,
 		d.officialname                                            AS collectedfromname,
-		o.country || '@@@' || o.country || '@@@dnet:countries@@@dnet:countries' AS country,
+		o.country || '@@@' || COALESCE(cntr.name,o.country) || '@@@dnet:countries@@@dnet:countries' AS country,
 		'sysimport:crosswalk:entityregistry@@@sysimport:crosswalk:entityregistry@@@dnet:provenance_actions@@@dnet:provenance_actions' AS provenanceaction,
-
 		ARRAY[]::text[]                                           AS pid
+
 FROM dsm_organizations o
 	LEFT OUTER JOIN dsm_datasources d ON (d.id = o.collectedfrom)
+	LEFT OUTER JOIN class cntr ON (cntr.code = o.country)