D-Net

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

import java.io.IOException;

import java.util.Iterator;

import eu.dnetlib.data.proto.*;

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 PropagationCountryInstitutionalOrganizationReducer extends TableReducer<Key, ImmutableBytesWritable, ImmutableBytesWritable>{

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

    final static String DNETCOUNTRYSCHEMA = "dnet:countries";

    private ImmutableBytesWritable keyOut;

    @Override

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

        super.setup(context);

        keyOut = new ImmutableBytesWritable();

    }

    @Override

    protected void reduce(final Key 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;

        }

        final String first = Bytes.toString(it.next().get());

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

			throw new RuntimeException("First Element in reducer is not type of datasource");

		}

        //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(it.next().get()); // need to update the information for the country to each element in the iterator

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

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

            }

			while(it.hasNext()) {

                byte[] targetRowKey = it.next().get();

				String resultId = Bytes.toString(targetRowKey);

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

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

                    //throw new RuntimeException("Problem in ordering. Elements different from publication in list after second position " + targetRowKey);

                }

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

                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 institutional datasource").increment(1);

        }

    }

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

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

			.setClassid(countryValue)

			.setClassname(countryValue)

			.setSchemeid(DNETCOUNTRYSCHEMA)

			.setSchemename(DNETCOUNTRYSCHEMA);

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

	}

}