package es.tid.cosmos.mobility.mapreduce;

import java.io.IOException;

import com.twitter.elephantbird.mapreduce.io.ProtobufWritable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.mapreduce.Reducer;

import es.tid.cosmos.mobility.data.MobProtocol.NodeBtsDay;
import es.tid.cosmos.mobility.data.NodeBtsDayUtil;

/**
 * Count NodeBtsDay
 *
 * @author sortega, dmicol
 */
public class RepbtsAggbybtsReducer extends
        Reducer<ProtobufWritable<NodeBtsDay>, IntWritable, LongWritable,
        ProtobufWritable<NodeBtsDay>> {
    private LongWritable userId;

    @Override
    public void setup(Context context) throws IOException,
                                              InterruptedException {
        this.userId = new LongWritable();
    }

    @Override
    public void reduce(ProtobufWritable<NodeBtsDay> inKey,
            Iterable<IntWritable> callCounts, Context context)
            throws IOException, InterruptedException {
        inKey.setConverter(NodeBtsDay.class);
        final NodeBtsDay byDay = inKey.get();
        this.userId.set(byDay.getUserId());

	int totalCallCount = 0;
        for (IntWritable callCount : callCounts) {
            totalCallCount += callCount.get();
        }

        ProtobufWritable<NodeBtsDay> nodeBtsDay = NodeBtsDayUtil.createAndWrap(
                byDay.getUserId(),
                byDay.getPlaceId(),
                byDay.getWorkday(),
                totalCallCount);
        context.write(this.userId, nodeBtsDay);
    }
}

@@ -409,23 +409,23 @@
 hor 
-sortega, dmicol
+dmicol, sortega
 %0A */
@@ -467,24 +467,16 @@
  extends
-%0A       
  Reducer
@@ -505,16 +505,24 @@
 BtsDay%3E,
+%0A       
  IntWrit
@@ -540,24 +540,16 @@
 ritable,
-%0A       
  Protobu
@@ -864,19 +864,17 @@
 BtsDay%3E 
-inK
+k
 ey,%0A    
@@ -903,25 +903,21 @@
 itable%3E 
-callCount
+value
 s, Conte
@@ -987,54 +987,8 @@
 n %7B%0A
-        inKey.setConverter(NodeBtsDay.class);%0A
     
@@ -1020,11 +1020,9 @@
 y = 
-inK
+k
 ey.g
@@ -1134,25 +1134,21 @@
 Count : 
-callCount
+value
 s) %7B%0A   

Improve RepbtsAggbybtsReducer.