D-Net

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

import com.google.gson.Gson;

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

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

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

import eu.dnetlib.data.mapreduce.hbase.propagation.organizationtoresult.Emit;

import eu.dnetlib.data.proto.*;

import org.apache.hadoop.io.Text;

import java.util.*;

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

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

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

public class ResultOrganizationIterator extends ResultIterator {

    private Set<String> toBeAffiliated;

    private String organizationToPropagate ;

    private Iterator<String> roi;

    public ResultOrganizationIterator(final Iterable<Text> values, final int key) throws NotValidResultSequenceException {

        super(values,key);

        init();

    }

private void init(){

        roi = toBeAffiliated.iterator();

        getNext();

}

@Override

protected void checkSequence() throws NotValidResultSequenceException {

    if (!it.hasNext()) {

        throw new NotValidResultSequenceException("Empty information for key");

    }

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

    trust = first.getTrust();

    if ( key != TypeProtos.Type.datasource.getNumber()){

        throw new NotValidResultSequenceException("First Element in reducer has not datasource as type, means no association with datasource was found for the organization ");

    }

    //ensure we are dealing with an institutional repository

    if (first.getValue().equals(ONE)) {

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

        if (!it.hasNext()) {

            throw new NotValidResultSequenceException("No information apart of type of datasource");

        }

        organizationToPropagate = Value.fromJson(it.next().toString()).getValue();

        if (!organizationToPropagate.startsWith("20|")) {

            throw new NotValidResultSequenceException("Second element in reducer is not the organization");

        }

        Emit e;

        toBeAffiliated = new HashSet<>();

        while (it.hasNext()) {

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

            if (!v.getType().equals(Type.fromresult)) {

                throw new NotValidResultSequenceException("result in reducer mixed up with other types ");

            }

            e = new Gson().fromJson(v.getValue(), Emit.class);

            if (!e.getResult_list().contains(organizationToPropagate))

                toBeAffiliated.add(e.getId());

        }

    }else

        throw new NotValidResultSequenceException("Not allowed dsType institutional datasource");

}

    private void getNext(){

        if (roi.hasNext())

            resultId = roi.next();

        else

            resultId = TERMINATOR;

    }

    @Override

    public List<OafProtos.Oaf> next() {

        ArrayList<OafProtos.Oaf> ret = new ArrayList<OafProtos.Oaf>();

             ret.add(getOafRel(resultId, organizationToPropagate, REL_RESULT_ORGANIZATION));

             ret.add(getOafRel(organizationToPropagate, resultId, REL_ORGANIZATIO_RESULT));

        getNext();

        return ret;

    }

    private OafProtos.Oaf getOafRel(String source, String target, String semantics){

        final ResultOrganizationProtos.ResultOrganization.Builder rob = ResultOrganizationProtos.ResultOrganization.newBuilder()

                .setAffiliation(ResultOrganizationProtos.ResultOrganization.Affiliation.newBuilder()

                        .setRelMetadata(RelMetadataProtos.RelMetadata.newBuilder()

                                .setSemantics(getQualifier(semantics,DNET_RELATION_SCHEMA_ORGANIZATION, semantics, DNET_RELATION_SCHEMA_ORGANIZATION)

                                )

                        )

                );

        final OafProtos.OafRel.Builder relation = OafProtos.OafRel.newBuilder()

                .setChild(false)

                .setSubRelType(SUBREL_TYPE_ORGANIZATION)

                .setRelType(REL_TYPE_ORGANIZATION)

                .setRelClass(semantics)

                .setTarget(target)

                .setSource(source)

                .setResultOrganization(rob);

        return OafProtos.Oaf.newBuilder().setKind(KindProtos.Kind.relation)

                .setRel(relation)

                .setDataInfo(getDataInfo(trust, CLASS_ORGANIZATION_RESULT_ID, SCHEMA_ID, SCHEMA_NAME, DATA_INFO_TYPE, CLASS_ORGANIZATION_RESULT_NAME))

                .build();

    }

}