D-Net

import java.util.Iterator;

import com.google.gson.Gson;

import eu.dnetlib.data.proto.*;

import org.apache.commons.lang3.StringUtils;

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

    final static String DNETSCHEMA = "dnet:countries";

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

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

        if(!it.hasNext()){

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

        final Value first = Value.fromJson(it.next().toString());

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

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

                context.getCounter(COUNTER_PROPAGATION,"First Element in reducer is not type of datasource,  but the organization exists").increment(1);

		    else {

		        context.getCounter(COUNTER_PROPAGATION, "WARNING: unexpected first element").increment(1);

                while (it.hasNext()) {

                    String resultId = Value.fromJson(it.next().toString()).getValue();

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

                        context.getCounter(COUNTER_PROPAGATION,"ERROR ORDERING CHECK").increment(1);

                    }

                }

            }

		}

        //ensure we are dealing with an institutional repository

        if (first.getValue().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 = Value.fromJson(it.next().toString()).getValue(); // 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) { }

                context.getCounter(COUNTER_PROPAGATION,"second element in reducer is not country").increment(1);

                return;

            }

            boolean propagate = true;

			while(it.hasNext()) {

				final String resultId = Value.fromJson(it.next().toString()).getValue();

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

				    if(!resultId.equalsIgnoreCase(country)) {

                        context.getCounter(COUNTER_PROPAGATION, "resultId expected in ordering was not found").increment(1);

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

                        propagate = false;

                    }

                } else {

				    if (propagate) {

                        byte[] oafUpdate = getOafCountry(resultId, country, first.getTrust());

                        byte[] targetRowKey = Bytes.toBytes(resultId);

                        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, "added country to product").increment(1);

                    }

                }

            }

        } else {

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

        }

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

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

		        getCountry(countryValue,trust,DNETSCHEMA,CLASS_ID,SCHEMA_ID,SCHEMA_NAME,DATA_INFO_TYPE,"Propagation of country information from datasources belonging to institutional repositories"));

		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();

	}