Index: modules/dnet-mapreduce-jobs/branches/master/src/main/java/eu/dnetlib/data/mapreduce/hbase/propagation/country/institutionalrepositories/PropagationCountryFromDsOrgResultFileReducer.java
===================================================================
--- modules/dnet-mapreduce-jobs/branches/master/src/main/java/eu/dnetlib/data/mapreduce/hbase/propagation/country/institutionalrepositories/PropagationCountryFromDsOrgResultFileReducer.java	(revision 54036)
+++ modules/dnet-mapreduce-jobs/branches/master/src/main/java/eu/dnetlib/data/mapreduce/hbase/propagation/country/institutionalrepositories/PropagationCountryFromDsOrgResultFileReducer.java	(revision 54037)
@@ -1,21 +1,18 @@
 package eu.dnetlib.data.mapreduce.hbase.propagation.country.institutionalrepositories;
 
 import com.googlecode.protobuf.format.JsonFormat;
-
-import eu.dnetlib.data.mapreduce.hbase.propagation.Value;
-import eu.dnetlib.data.proto.*;
+import eu.dnetlib.data.mapreduce.hbase.propagation.NotValidResultSequenceException;
+import eu.dnetlib.data.mapreduce.hbase.propagation.ResultIterator;
+import eu.dnetlib.data.proto.OafProtos;
 import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.io.Text;
 import org.apache.hadoop.mapreduce.Reducer;
 
 import java.io.IOException;
-import java.util.Iterator;
 
-import org.apache.commons.logging.LogFactory;
+import static eu.dnetlib.data.mapreduce.hbase.propagation.PropagationConstants.COUNTER_PROPAGATION;
 
-import static eu.dnetlib.data.mapreduce.hbase.propagation.PropagationConstants.*;
-import static eu.dnetlib.data.mapreduce.hbase.propagation.Utils.getCountry;
-
 public class PropagationCountryFromDsOrgResultFileReducer extends Reducer<InstOrgKey, Text, Text,Text> {
 
     private static final Log log = LogFactory.getLog(PropagationCountryFromDsOrgResultFileReducer.class); // NOPMD by marko on 11/24/08 5:02 PM
@@ -24,10 +21,6 @@
 
     private Text outValue;
 
-    private final static String DNETSCHEMA = "dnet:countries";
-    private final static String CLASS_ID = "propagation::country::instrepos";
-
-
     @Override
     protected void setup(final Context context) throws IOException, InterruptedException {
         super.setup(context);
@@ -50,84 +43,29 @@
     @Override
     protected void reduce(InstOrgKey key, Iterable<Text> values, Context context)  {
 
-        final Iterator<Text> it = values.iterator();
-        if(!it.hasNext()){
-            context.getCounter(COUNTER_PROPAGATION,"empty information for key").increment(1);
+        ResultIterator rh = null;
+        try {
+            rh = new ResultCountryIterator(values,key.getKeyType().get());
+        } catch (NotValidResultSequenceException e) {
+            context.getCounter(COUNTER_PROPAGATION,e.getMessage()).increment(1);
             return;
         }
+        context.getCounter(COUNTER_PROPAGATION,"institutional datasource").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,String.format("WARNINGS: First element value is  %s " , first));
-                while (it.hasNext()) {
+        while(rh.hasNext()){
+            OafProtos.Oaf oafUpdate = rh.next();
+            emit(context, rh.getResultId(), JsonFormat.printToString(oafUpdate));
+            context.getCounter(COUNTER_PROPAGATION, " added country to product ").increment(1);
+        }
 
-                    String resultId = Value.fromJson(it.next().toString()).getValue();
-                    if (!resultId.startsWith("50|"))
-                        context.getCounter(COUNTER_PROPAGATION,"ERROR ORDERING CHECK").increment(1);
-                }
-            }
+        if (!rh.getPropagate()){
+            context.getCounter(COUNTER_PROPAGATION, "resultId expected in ordering was not found" ).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){
-                context.getCounter(COUNTER_PROPAGATION,"second element in reducer is not country").increment(1);
-                return;
-            }
-
-            boolean propagate = true;
-            while(it.hasNext()) {
-                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);
-                        propagate = false;
-                    }
-
-                }else{
-                    if (propagate){
-                        OafProtos.Oaf oafUpdate = getOafCountry(resultId, country,first.getTrust());
-                        emit(context, resultId, JsonFormat.printToString(oafUpdate));
-
-                        context.getCounter(COUNTER_PROPAGATION, " added country to product ").increment(1);
-                    }
-
-                }
-
-            }
-        } else {
-            context.getCounter(COUNTER_PROPAGATION,"not allowed dsType institutional datasource").increment(1);
-        }
     }
 
 
 
-    private OafProtos.Oaf 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();
-    }
-
-
 }
Index: modules/dnet-mapreduce-jobs/branches/master/src/main/java/eu/dnetlib/data/mapreduce/hbase/propagation/country/institutionalrepositories/PropagationCountryFromDsOrgResultReducer.java
===================================================================
--- modules/dnet-mapreduce-jobs/branches/master/src/main/java/eu/dnetlib/data/mapreduce/hbase/propagation/country/institutionalrepositories/PropagationCountryFromDsOrgResultReducer.java	(revision 54036)
+++ modules/dnet-mapreduce-jobs/branches/master/src/main/java/eu/dnetlib/data/mapreduce/hbase/propagation/country/institutionalrepositories/PropagationCountryFromDsOrgResultReducer.java	(revision 54037)
@@ -1,13 +1,11 @@
 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.NotValidResultSequenceException;
+import eu.dnetlib.data.mapreduce.hbase.propagation.ResultIterator;
 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;
@@ -18,18 +16,12 @@
 import org.apache.hadoop.io.Text;
 
 import static eu.dnetlib.data.mapreduce.hbase.propagation.PropagationConstants.*;
-import static eu.dnetlib.data.mapreduce.hbase.propagation.Utils.*;
 
 public class PropagationCountryFromDsOrgResultReducer extends TableReducer<InstOrgKey, Text, ImmutableBytesWritable>{
 
     private static final Log log = LogFactory.getLog(PropagationCountryFromDsOrgResultReducer.class);
 
-    final static String DNETSCHEMA = "dnet:countries";
 
-
-    private final static String CLASS_ID = "propagation::country::instrepos";
-
-
     private ImmutableBytesWritable keyOut;
 
     @Override
@@ -41,91 +33,35 @@
     @Override
     protected void reduce(final InstOrgKey key, final Iterable<Text> values, final Context context) throws IOException, InterruptedException {
 
-        final Iterator<Text> it = values.iterator();
-        if(!it.hasNext()){
-            context.getCounter(COUNTER_PROPAGATION,"empty information for key").increment(1);
+
+        ResultIterator rh = null;
+        try {
+            rh = new ResultCountryIterator(values,key.getKeyType().get());
+        } catch (NotValidResultSequenceException e) {
+            context.getCounter(COUNTER_PROPAGATION,e.getMessage()).increment(1);
             return;
         }
-        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()) {
+        context.getCounter(COUNTER_PROPAGATION,"institutional datasource").increment(1);
 
-                    String resultId = Value.fromJson(it.next().toString()).getValue();
-                    if (!resultId.startsWith("50|")) {
-                        context.getCounter(COUNTER_PROPAGATION,"ERROR ORDERING CHECK").increment(1);
-                    }
-                }
-            }
-		}
+        while(rh.hasNext()){
+            byte[] oafUpdate = rh.next().toByteArray();
 
-        //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;
-            }
+            byte[] targetRowKey = Bytes.toBytes(rh.getResultId());
+            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);
 
+        }
 
-            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;
-            }
+        if (!rh.getPropagate()){
+            context.getCounter(COUNTER_PROPAGATION, "resultId expected in ordering was not found" ).increment(1);
+        }
 
-            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();
-	}
-
 }
Index: modules/dnet-mapreduce-jobs/branches/master/src/main/java/eu/dnetlib/data/mapreduce/hbase/propagation/country/institutionalrepositories/PropagationCountryFromDsOrgResultMapper.java
===================================================================
--- modules/dnet-mapreduce-jobs/branches/master/src/main/java/eu/dnetlib/data/mapreduce/hbase/propagation/country/institutionalrepositories/PropagationCountryFromDsOrgResultMapper.java	(revision 54036)
+++ modules/dnet-mapreduce-jobs/branches/master/src/main/java/eu/dnetlib/data/mapreduce/hbase/propagation/country/institutionalrepositories/PropagationCountryFromDsOrgResultMapper.java	(revision 54037)
@@ -136,6 +136,7 @@
     }
 
 
+
     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"));
