D-Net

	public void testParseDatasetNeuroVault() throws Exception {

		doTest(loadFromTransformationProfile("odf2hbase.xml"), load("recordNeuroVault.xml"));

	}

	@Test

<?xml version="1.0" encoding="UTF-8"?>

<record xmlns:dr="http://www.driver-repository.eu/namespace/dr"

        xmlns:oaf="http://namespace.openaire.eu/oaf" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">

    <oai:header xmlns="http://namespace.openaire.eu/"

                xmlns:dc="http://purl.org/dc/elements/1.1/"

                xmlns:dri="http://www.driver-repository.eu/namespace/dri"

                xmlns:oai="http://www.openarchives.org/OAI/2.0/" xmlns:prov="http://www.openarchives.org/OAI/2.0/provenance">

        <dri:objIdentifier>r36ad727a71a::01934defe9d8c0ada9dbf6f785f7a802</dri:objIdentifier>

        <dri:recordIdentifier>oai:oai.datacite.org:12102337</dri:recordIdentifier>

        <dri:dateOfCollection>2017-12-12T17:06:28.545+01:00</dri:dateOfCollection>

        <oaf:datasourceprefix>r36ad727a71a</oaf:datasourceprefix>

        <identifier xmlns="http://www.openarchives.org/OAI/2.0/">oai:oai.datacite.org:12102337</identifier>

        <datestamp xmlns="http://www.openarchives.org/OAI/2.0/">2017-09-29T09:14:53Z</datestamp>

        <setSpec xmlns="http://www.openarchives.org/OAI/2.0/">BL</setSpec>

        <setSpec xmlns="http://www.openarchives.org/OAI/2.0/">BL.BATH</setSpec>

        <dr:dateOfTransformation>2017-12-12T17:06:45.312+01:00</dr:dateOfTransformation>

    </oai:header>

    <metadata>

        <resource xmlns="http://datacite.org/schema/kernel-4"

                  xmlns:dc="http://purl.org/dc/elements/1.1/"

                  xmlns:dri="http://www.driver-repository.eu/namespace/dri"

                  xmlns:oai="http://www.openarchives.org/OAI/2.0/"

                  xmlns:prov="http://www.openarchives.org/OAI/2.0/provenance" xsi:schemaLocation="http://datacite.org/schema/kernel-4 http://schema.datacite.org/meta/kernel-4/metadata.xsd">

            <identifier identifierType="DOI">10.15125/BATH-00381</identifier>

            <creators>

                <creator>

                    <creatorName>James, Tony D.</creatorName>

                    <givenName>Tony D.</givenName>

                    <familyName>James</familyName>

                    <nameIdentifier nameIdentifierScheme="ORCID" schemeURI="http://orcid.org/">0000-0002-4095-2191</nameIdentifier>

                    <affiliation>University of Bath</affiliation>

                </creator>

                <creator>

                    <creatorName>Sedgwick, Adam C.</creatorName>

                    <givenName>Adam C.</givenName>

                    <familyName>Sedgwick</familyName>

                    <affiliation>University of Bath</affiliation>

                </creator>

            </creators>

            <titles>

                <title>Data from "A bodipy based hydroxylamine sensor"</title>

            </titles>

            <publisher>University of Bath</publisher>

            <publicationYear>2017</publicationYear>

            <subjects>

                <subject

                        schemeURI="http://www.rcuk.ac.uk/research/efficiency/researchadmin/harmonisation/" subjectScheme="RCUK Research Classifications">Biological and Medicinal Chemistry</subject>

            </subjects>

            <resourceType resourceTypeGeneral="Dataset">Dataset</resourceType>

            <fundingReferences>

                <fundingReference>

                    <funderName>Engineering and Physical Sciences Research Council (EPSRC)</funderName>

                    <funderIdentifier funderIdentifierType="Crossref Funder ID">https://doi.org/10.13039/501100000266</funderIdentifier>

                    <awardNumber>EP/G03768X/1</awardNumber>

                    <awardTitle>Doctoral Training Centre in Sustainable Chemical Technologies</awardTitle>

                </fundingReference>

            </fundingReferences>

            <contributors>

                <contributor contributorType="DataCollector">

                    <contributorName>Chapman, Robert S. L.</contributorName>

                    <givenName>Robert S. L.</givenName>

                    <familyName>Chapman</familyName>

                    <affiliation>University of Bath</affiliation>

                </contributor>

                <contributor contributorType="DataCollector">

                    <contributorName>Gardiner, Jordan E.</contributorName>

                    <givenName>Jordan E.</givenName>

                    <familyName>Gardiner</familyName>

                    <affiliation>University of Bath</affiliation>

                </contributor>

                <contributor contributorType="DataCollector">

                    <contributorName>Peacock, Lucy R.</contributorName>

                    <givenName>Lucy R.</givenName>

                    <familyName>Peacock</familyName>

                    <affiliation>University of Bath</affiliation>

                </contributor>

                <contributor contributorType="DataCollector">

                    <contributorName>Kim, Gyungmi</contributorName>

                    <givenName>Gyungmi</givenName>

                    <familyName>Kim</familyName>

                    <affiliation>Ewha Womans University</affiliation>

                </contributor>

                <contributor contributorType="Supervisor">

                    <contributorName>Yoon, Juyoung</contributorName>

                    <givenName>Juyoung</givenName>

                    <familyName>Yoon</familyName>

                    <affiliation>Ewha Womans University</affiliation>

                </contributor>

                <contributor contributorType="Supervisor">

                    <contributorName>Bull, Steven D.</contributorName>

                    <givenName>Steven D.</givenName>

                    <familyName>Bull</familyName>

                    <affiliation>University of Bath</affiliation>

                </contributor>

                <contributor contributorType="RightsHolder">

                    <contributorName>University of Bath</contributorName>

                </contributor>

            </contributors>

            <language>en</language>

            <alternateIdentifiers xmlns="http://www.openarchives.org/OAI/2.0/">

                <alternateIdentifier alternateIdentifierType="URL" xmlns="http://datacite.org/schema/kernel-4">http://researchdata.bath.ac.uk/381/</alternateIdentifier>

                <alternateIdentifier alternateIdentifierType="URL">http://dx.doi.org/10.15125/BATH-00381</alternateIdentifier>

            </alternateIdentifiers>

            <relatedIdentifiers>

                <relatedIdentifier relatedIdentifierType="URL" relationType="IsDocumentedBy">http://researchdata.bath.ac.uk/381/2/README.rtf</relatedIdentifier>

                <relatedIdentifier relatedIdentifierType="URL" relationType="HasPart">http://researchdata.bath.ac.uk/381/3/A_Bodipy_based_Hydroxylamine_sensor.zip</relatedIdentifier>

                <relatedIdentifier relatedIdentifierType="DOI" relationType="IsReferencedBy">10.1039/C7CC05872A</relatedIdentifier>

            </relatedIdentifiers>

            <sizes>

                <size>README.rtf - 94kB</size>

                <size>A_Bodipy_based_Hydroxylamine_sensor.zip - 12MB</size>

            </sizes>

            <formats>

                <format>README.rtf - text/rtf</format>

                <format>A_Bodipy_based_Hydroxylamine_sensor.zip - application/zip</format>

            </formats>

            <version>1</version>

            <rightsList>

                <rights>README.rtf has no associated licence.  Please contact the archive for advice.</rights>

                <rights rightsURI="http://creativecommons.org/licenses/by/4.0/">A_Bodipy_based_Hydroxylamine_sensor.zip is licensed under the Creative Commons Attribution 4.0 License</rights>

            </rightsList>

            <descriptions>

                <description descriptionType="Abstract">This repository contains the experimental data discussed in the manuscript. Including, 1H NMR, 13C NMR (FID and PDF) and Mass Spectra for all the intermediates (A-H); probe 1; the hydroxylamine nitrone cyclic product; N-Methylhydroxylamine cyclic product; N,O,-BocHydroxylamine alkyne and N-Hydroxylamine alkyne. Fluorescence analysis data of probe 1 including the hydroxylamine titration curve; selectivity data against other hydroxylamines; amines and amino acids, Fluorescence intensity changes for probe 1 as a function of time with increasing concentrations of hydroxylamine (time drive) and cell images.</description>

            </descriptions>

        </resource>

        <dr:CobjCategory>0021</dr:CobjCategory>

        <oaf:dateAccepted>2017-01-01</oaf:dateAccepted>

        <oaf:accessrights>OPEN</oaf:accessrights>

        <oaf:language>eng</oaf:language>

        <oaf:hostedBy id="re3data_____::r3d100011947" name="University of Bath Research Data Archive"/>

        <oaf:collectedFrom id="re3data_____::r3d100011947" name="University of Bath Research Data Archive"/>

    </metadata>

    <about xmlns:dc="http://purl.org/dc/elements/1.1/"

           xmlns:dri="http://www.driver-repository.eu/namespace/dri"

           xmlns:oai="http://www.openarchives.org/OAI/2.0/" xmlns:prov="http://www.openarchives.org/OAI/2.0/provenance">

        <provenance xmlns="http://www.openarchives.org/OAI/2.0/provenance" xsi:schemaLocation="http://www.openarchives.org/OAI/2.0/provenance http://www.openarchives.org/OAI/2.0/provenance.xsd">

            <originDescription altered="true" harvestDate="2017-12-12T17:06:28.545+01:00">

                <baseURL>https://oai.datacite.org/oai</baseURL>

                <identifier>oai:oai.datacite.org:12102337</identifier>

                <datestamp>2017-09-29T09:14:53Z</datestamp>

                <metadataNamespace>http://datacite.org/schema/nonexistant</metadataNamespace>

            </originDescription>

        </provenance>

        <oaf:datainfo>

            <oaf:inferred>false</oaf:inferred>

            <oaf:deletedbyinference>false</oaf:deletedbyinference>

            <oaf:trust>0.9</oaf:trust>

            <oaf:inferenceprovenance/>

            <oaf:provenanceaction classid="sysimport:crosswalk:datasetarchive"

                                  classname="sysimport:crosswalk:datasetarchive"

                                  schemeid="dnet:provenanceActions" schemename="dnet:provenanceActions"/>

        </oaf:datainfo>

    </about>

</record>

package eu.dnetlib.data.mapreduce.hbase.propagation;

import eu.dnetlib.data.mapreduce.hbase.propagation.country.institutionalrepositories.InstOrgKey;

import org.apache.hadoop.hbase.io.ImmutableBytesWritable;

import org.apache.hadoop.mapreduce.Partitioner;

/**

 * Created by miriam on 17/08/2018.

 */

public class NaturalKeyPartitioner extends Partitioner<InstOrgKey, ImmutableBytesWritable > {

    @Override

    public int getPartition(InstOrgKey key, ImmutableBytesWritable  val, int numPartitions) {

        final int res = Math.abs(key.getId().hashCode() % numPartitions);

        //System.out.println(String.format("%s goes to partition %s", key.toString(), res));

        return res;

    }

}

package eu.dnetlib.data.mapreduce.hbase.propagation;

import eu.dnetlib.data.proto.TypeProtos;

public class PropagationConstants {

    public static final String ZERO = "0";

    public static final String ONE = "1";

    public static final String COUNTER_PROPAGATION = "Propagation";

    public final static int PROJECT = TypeProtos.Type.project.getNumber();

    public final static int DATASOURCE = TypeProtos.Type.datasource.getNumber();

    public final static int ORGANIZATION = TypeProtos.Type.organization.getNumber();

    public final static int PUBLICATION = TypeProtos.Type.result.getNumber();

}

package eu.dnetlib.data.mapreduce.hbase.propagation;

import eu.dnetlib.data.mapreduce.hbase.propagation.country.institutionalrepositories.InstOrgKey;

import org.apache.hadoop.io.WritableComparable;

import org.apache.hadoop.io.WritableComparator;

/**

 * Created by miriam on 17/08/2018.

 */

public class NaturalKeyGroupingComparator  extends WritableComparator {

    protected NaturalKeyGroupingComparator() {

        super(InstOrgKey.class, true);

    }

    @Override

    public int compare(WritableComparable w1, WritableComparable w2) {

        final InstOrgKey k1 = (InstOrgKey) w1;

        final InstOrgKey k2 = (InstOrgKey) w2;

        return k1.getId().compareTo(k2.getId());

    }

}

package eu.dnetlib.data.mapreduce.hbase.propagation.country.institutionalrepositories;

import java.io.IOException;

import java.util.Iterator;

import com.google.gson.Gson;

import eu.dnetlib.data.mapreduce.hbase.propagation.Value;

import eu.dnetlib.data.proto.*;

import org.apache.commons.lang3.StringUtils;

import org.apache.hadoop.hbase.io.ImmutableBytesWritable;

import org.apache.hadoop.hbase.mapreduce.TableReducer;

import org.apache.hadoop.hbase.util.Bytes;

import org.apache.hadoop.hbase.client.Put;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import static eu.dnetlib.data.mapreduce.hbase.propagation.PropagationConstants.*;

public class PropagationCountryFromDsOrgResultReducer extends TableReducer<InstOrgKey, ImmutableBytesWritable, ImmutableBytesWritable>{

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

    final static String DNETCOUNTRYSCHEMA = "dnet:countries";

    private final static String DATA_INFO_TYPE = "propagation";

    private final static String SCHEMA_NAME = "dnet:provenanceActions";

    private final static String CLASS_ID = "propagation::country::instrepos";

    private final static String SCHEMA_ID = "dnet:provenanceActions";

   // private final static String TRUST = "0.9";

    private ImmutableBytesWritable keyOut;

    //private Text keyOut;

    @Override

    protected void setup(final Context context) throws IOException, InterruptedException {

        super.setup(context);

       keyOut = new ImmutableBytesWritable();

        //keyOut = new Text();

    }

    @Override

    protected void reduce(final InstOrgKey dsId, final Iterable<ImmutableBytesWritable> values, final Context context) throws IOException, InterruptedException {

        final Iterator<ImmutableBytesWritable> it = values.iterator();

        if(!it.hasNext()){

            context.getCounter(COUNTER_PROPAGATION,"empty information for key").increment(1);

            return;

        }

        Gson gson = new Gson();

        Value v = gson.fromJson(Bytes.toString(it.next().get()),Value.class);

        final String first  = Bytes.toString(v.getValue().get());

        final String trust = v.getTrust();

		if (!(first.equals(ZERO) || first.equals(ONE))) {

		    if (dsId.getKeyType().get() == TypeProtos.Type.organization.getNumber())

			    throw new RuntimeException(String.format("First Element in reducer is not type of datasource, key: '%s', value: '%s' but the organization exists", dsId.toString(), first));

		    else {

		        context.getCounter(COUNTER_PROPAGATION,String.format("WARNINGS: First element value is  %s " , first));

                while (it.hasNext()) {

                    byte[] targetRowKey = gson.fromJson(Bytes.toString(it.next().get()), Value.class).getValue().get();

                    String resultId = Bytes.toString(targetRowKey);

                    if (!resultId.startsWith("50|"))

                        throw new RuntimeException("Wrong ordering. CHECK!!!");

                }

            }

		}

        //ensure we are dealing with an institutional repository

        if (first.equals(ONE)) {

            context.getCounter(COUNTER_PROPAGATION,"institutional datasource").increment(1);

            if(!it.hasNext()){

                context.getCounter(COUNTER_PROPAGATION,"no information apart of type of datasource").increment(1);

                return;

            }

            final String country = Bytes.toString(gson.fromJson(Bytes.toString(it.next().get()),Value.class).getValue().get()); // need to update the information for the country to each element in the iterator

            if(country.trim().length() != 2){

                try{

                    Integer.parseInt(country.trim());

                }catch(Exception e){

                    throw new RuntimeException("Second Element in reducer is not country " + country);

                }

                throw new RuntimeException("Second Element in reducer is not country " + country);

            }

            boolean propagate = true;

			while(it.hasNext()) {

                byte[] targetRowKey = gson.fromJson(Bytes.toString(it.next().get()),Value.class).getValue().get();

				String resultId = Bytes.toString(targetRowKey);

				if (!resultId.startsWith(("50|"))) {

				    if(!resultId.equalsIgnoreCase(country)) {

                        context.getCounter(COUNTER_PROPAGATION, String.format("resultId expected in ordering was not found. First country %s, second country %s", country, resultId)).increment(1);

                        //throw new RuntimeException("Problem in country values of institutional repositories" + targetRowKey);

                        propagate = false;

                    }

                }else{

				    if (propagate){

                        byte[] oafUpdate = getOafCountry(resultId, country,trust);

                        final Put put = new Put(targetRowKey).add(Bytes.toBytes("result"), Bytes.toBytes("update_"+System.nanoTime()), oafUpdate);

                        keyOut.set(targetRowKey);

                        context.write(keyOut, put);

                        context.getCounter(COUNTER_PROPAGATION, String.format(" added country %s to product from repo %s",country, StringUtils.substringBefore( resultId,"::"))).increment(1);

                    }

                }

            }

        } else {

            context.getCounter(COUNTER_PROPAGATION,"not allowed dsType institutional datasource").increment(1);

        }

    }

    private FieldTypeProtos.DataInfo.Builder getDataInfo(String trust) {

        FieldTypeProtos.DataInfo.Builder builder = FieldTypeProtos.DataInfo.newBuilder()

                .setInferred(true)

                .setProvenanceaction(

                        FieldTypeProtos.Qualifier.newBuilder()

                                .setClassid(CLASS_ID)

                                .setClassname("Propagation of country information from datasources belonging to institutional repositories")

                                .setSchemeid(SCHEMA_ID)

                                .setSchemename(SCHEMA_NAME))

                .setInferenceprovenance(DATA_INFO_TYPE)

                .setTrust(trust);

        return builder;

    }

	private byte[] getOafCountry(String resultId, String countryValue, String trust) {

		final FieldTypeProtos.Qualifier.Builder country = FieldTypeProtos.Qualifier.newBuilder()

			.setClassid(countryValue)

			.setClassname(countryValue)

			.setSchemeid(DNETCOUNTRYSCHEMA)

			.setSchemename(DNETCOUNTRYSCHEMA);

		country.setDataInfo(getDataInfo(trust));

		final ResultProtos.Result.Metadata.Builder metadata = ResultProtos.Result.Metadata.newBuilder().addCountry(country);

		final ResultProtos.Result.Builder result = ResultProtos.Result.newBuilder().setMetadata(metadata);

		final OafProtos.OafEntity.Builder entity = OafProtos.OafEntity.newBuilder()

				.setType(TypeProtos.Type.result)

				.setId(resultId)

				.setResult(result);

    	return OafProtos.Oaf.newBuilder()

				.setKind(KindProtos.Kind.entity)

				.setEntity(entity)

				.build()

				.toByteArray();

	}

}

package eu.dnetlib.data.mapreduce.hbase.propagation.country.institutionalrepositories;

import java.io.DataInput;

import java.io.DataOutput;

import java.io.IOException;

import com.google.common.collect.ComparisonChain;

import org.apache.hadoop.io.IntWritable;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.io.WritableComparable;

import static eu.dnetlib.data.mapreduce.hbase.propagation.PropagationConstants.*;

/**

 * Created by miriam on 17/08/2018.

 */

public class InstOrgKey implements WritableComparable<InstOrgKey> {

    private IntWritable keyType;

    private Text id;

    public InstOrgKey() {}

    public static InstOrgKey create(final int keyType, final String id) {

        return new InstOrgKey(keyType, id);

    }

    public static InstOrgKey datasource(final String id) {

        return new InstOrgKey(DATASOURCE, id);

    }

    public static InstOrgKey organization(final String id) {

        return new InstOrgKey(ORGANIZATION, id);

    }

    public static InstOrgKey publication(final String id){

        return new InstOrgKey(PUBLICATION, id);

    }

    public InstOrgKey(final int keyType, final String id) {

        this.id = new Text(id);

        this.keyType = new IntWritable(keyType);

    }

    public void setKeyType(final IntWritable keyType) {

        this.keyType = keyType;

    }

    public void setId(final Text id) {

        this.id = id;

    }

    public Text getId() {

        return id;

    }

    public IntWritable getKeyType() {

        return keyType;

    }

    @Override

    public int compareTo(final InstOrgKey o) {

        final int res = ComparisonChain.start()

                .compare(getId(), o.getId())

                .compare(getKeyType(), o.getKeyType())

                .result();

        //System.out.println(String.format("%s.compareTo(%s) = %s", toString(), o.toString(), res));

        return res;

    }

    @Override

    public void write(final DataOutput out) throws IOException {

        keyType.write(out);

        id.write(out);

    }

    @Override

    public void readFields(final DataInput in) throws IOException {

        keyType = new IntWritable();

        keyType.readFields(in);

        id = new Text();

        id.readFields(in);

    }

    @Override

    public String toString() {

        return (new StringBuilder())

                .append('{')

                .append(getKeyType().get())

                .append(',')

                .append(getId())

                .append('}')

                .toString();

    }

}

package eu.dnetlib.data.mapreduce.hbase.propagation.country.institutionalrepositories;

import com.google.common.base.Splitter;

import com.google.common.collect.Lists;

import com.google.common.collect.Sets;

import com.google.gson.Gson;

import com.google.protobuf.InvalidProtocolBufferException;

import eu.dnetlib.data.mapreduce.hbase.propagation.Value;

import eu.dnetlib.data.mapreduce.util.OafRowKeyDecoder;

import eu.dnetlib.data.proto.*;

import org.apache.commons.collections.MapUtils;

import org.apache.hadoop.hbase.client.Result;

import org.apache.hadoop.hbase.io.ImmutableBytesWritable;

import org.apache.hadoop.hbase.mapreduce.TableMapper;

import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;

import java.util.Map;

import java.util.Set;

import java.util.stream.Collectors;

import static eu.dnetlib.data.mapreduce.hbase.propagation.PropagationConstants.*;

/**

 * Created by miriam on 17/08/2018.

 */

public class PropagationCountryFromDsOrgResultMapper extends TableMapper<InstOrgKey, ImmutableBytesWritable> {

    private Value value = new Value();

    //private Text valueOut;

    private ImmutableBytesWritable valueOut;

    private Set<String> datasourceTypes = Sets.newHashSet("pubsrepository::institutional");

    private Set<String> whiteList = Sets.newHashSet("10|opendoar____::300891a62162b960cf02ce3827bb363c");

    private Set<String> blackList = Sets.newHashSet("");

    @Override

    protected void setup(final Context context) throws IOException, InterruptedException {

        super.setup(context);

        valueOut = new ImmutableBytesWritable();

        datasourceTypes.addAll(Lists.newArrayList(Splitter.on(",").omitEmptyStrings().split(context.getConfiguration().get("datasource.types", ""))));

        whiteList.addAll(Lists.newArrayList(Splitter.on(",").omitEmptyStrings().split(context.getConfiguration().get("datasource.whitelist", ""))));

    }

    @Override

    protected void map(final ImmutableBytesWritable keyIn, final Result value, final Context context) throws IOException, InterruptedException {

        final TypeProtos.Type type = OafRowKeyDecoder.decode(keyIn.copyBytes()).getType();

        final OafProtos.OafEntity entity = getEntity(value,type);

        if (entity != null) {

            switch (type) {

                case datasource:

                    final DatasourceProtos.Datasource datasource = entity.getDatasource();

                    final String id = entity.getId();

                    if (datasource == null) {

                        throw new RuntimeException("oaf type is datasource, but datasource proto is not found in oafproto");

                    }

                    String dsType = datasource.getMetadata().getDatasourcetype().getClassid();

                    if (datasourceTypes.contains(dsType)) {

                        // verify datasource is in blacklist

                            if (blackList.contains(id)){

                                context.getCounter(COUNTER_PROPAGATION,"blacklisted ").increment(1);

                                emitNotAllowedDatasource(context,entity.getId());

                            }else{

                                emitAllowedDatasource(value, context, entity.getId(), dsType);

                            }

                    } else {

                        // verify datasource is in whiteList

                            if (whiteList.contains(id)){

                                context.getCounter(COUNTER_PROPAGATION,"whitelisted " + id).increment(1);

                                emitAllowedDatasource(value,context,entity.getId(),dsType);

                        }else {

                                emitNotAllowedDatasource(context, entity.getId());

                            }

                    }

                    break;

                case organization:

                    OrganizationProtos.Organization organization = entity.getOrganization();

                    if (organization == null) {

                        throw new RuntimeException("oaf type is organization, but organization proto is not found in oafproto");

                    }

                    FieldTypeProtos.Qualifier country = organization.getMetadata().getCountry();

                    if (country == null) {

                        context.getCounter(COUNTER_PROPAGATION, "country elem does not exists").increment(1);

                    } else {

                        final Map<byte[], byte[]> ds_org = value.getFamilyMap("datasourceOrganization_provision_isProvidedBy".getBytes());

                        if (MapUtils.isNotEmpty(ds_org)) {

                            for (String dsId : ds_org.keySet().stream().map(String::new).collect(Collectors.toList())) {

                                this.value.getValue().set(country.getClassid().getBytes());

                                Gson gson = new Gson();

                                valueOut.set(gson.toJson(this.value).getBytes());

                                context.write(InstOrgKey.organization(dsId), valueOut);

                                //context.getCounter(COUNTER_PROPAGATION, "country " + country.getClassid() ).increment(1);

                                context.getCounter(COUNTER_PROPAGATION, "country ").increment(1);

                            }

                        }

                    }

                    break;

                case result:

                    ResultProtos.Result result = entity.getResult();

                    for (ResultProtos.Result.Instance instance : result.getInstanceList()) {

                        //todo add check if key is not empty and field is not null

                        String hostedBy = instance.getHostedby().getKey();

                        this.value.getValue().set(entity.getId().getBytes());

                        Gson gson = new Gson();

                        valueOut.set(gson.toJson(this.value).getBytes());

                        context.write(InstOrgKey.publication(hostedBy),valueOut);

                        context.getCounter(COUNTER_PROPAGATION, "emit publication ").increment(1);

                        String collectedFrom = instance.getCollectedfrom().getKey();

                        if (!hostedBy.equals(collectedFrom)) {

                            context.write(InstOrgKey.publication(collectedFrom), valueOut);

                            context.getCounter(COUNTER_PROPAGATION, "emit publication ").increment(1);

                        }

                    }

                    break;

            }

        }

    }

    private void emitNotAllowedDatasource(Context context, String id) throws IOException, InterruptedException {

        final InstOrgKey datasource1 = InstOrgKey.datasource(id);

        context.getCounter(COUNTER_PROPAGATION, "ds Type not in propagation allowed list").increment(1);

        this.value.getValue().set(ZERO.getBytes());

        Gson gson = new Gson();

        valueOut.set(gson.toJson(this.value).getBytes());

        context.write(datasource1, valueOut);

    }

    private void emitAllowedDatasource(Result value, Context context, String id, String dsType) throws IOException, InterruptedException {

        final InstOrgKey datasource1 = InstOrgKey.datasource(id);

        context.getCounter(COUNTER_PROPAGATION, String.format("%s in propagation allowed list", dsType)).increment(1);

        //value.set(ONE.getBytes());

        this.value.getValue().set(ONE.getBytes());

        this.value.setTrust(getTrust(value));

        Gson gson = new Gson();

        valueOut.set(gson.toJson(this.value).getBytes());

        context.write(datasource1, valueOut);

    }

    private OafProtos.OafEntity getEntity(Result value, TypeProtos.Type type) throws InvalidProtocolBufferException {

        final Map<byte[],byte[]> map = value.getFamilyMap(Bytes.toBytes(type.toString()));

        final byte[] body = map.get(Bytes.toBytes("body"));

        if (body != null){

            OafProtos.Oaf oaf = OafProtos.Oaf.parseFrom(body);

            if(oaf.getDataInfo().getDeletedbyinference())

                return null;

            return oaf.getEntity();

        }

        return null;

    }

    private String getTrust(Result value) throws InvalidProtocolBufferException {

        final Map<byte[],byte[]> map = value.getFamilyMap(Bytes.toBytes("datasource"));

        final byte[] body = map.get(Bytes.toBytes("body"));

        if (body != null){

            OafProtos.Oaf oaf = OafProtos.Oaf.parseFrom(body);

            return oaf.getDataInfo().getTrust();

        }

        return null;

    }

}

package eu.dnetlib.data.mapreduce.hbase.propagation;

import org.apache.hadoop.hbase.io.ImmutableBytesWritable;

import java.io.Serializable;

public class Value implements Serializable {

    private ImmutableBytesWritable value;

    private String trust;

    public Value(){

        value = new ImmutableBytesWritable();

        trust = "0";

    }

    public ImmutableBytesWritable getValue() {

        return value;

    }

    public void setValue(ImmutableBytesWritable value) {

        this.value = value;

    }

    public String getTrust() {

        return trust;

    }

    public void setTrust(String trust) {

        this.trust = trust;

    }

}