package it.uniroma2.ember;

import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;

/**
 * This is a utility class for aggregate statistics in Project Ember
 */

public class EmberSensorsAggregation {

    /**
     * Implements a simple KeySelector to divide by key (address) the traffic data - window join
     */
    public static final class EmberTrafficMeanSelector implements KeySelector<Tuple2<String, Float>, String> {

        /**
         * @param trafficData the tuple <key, traffic_intensity>
         * @return address as a key
         * @throws Exception
         */
        @Override
        public String getKey(Tuple2<String, Float> trafficData) throws Exception {
            return trafficData.f0;
        }
    }

    /**
     * Implements a simple KeySelector to divide by key (address) the lumen data - window join
     */
    public static final class EmberLumenMeanSelector implements KeySelector<Tuple2<String, Float>, String> {

        /**
         * @param lumenData the tuple <key, light_level>
         * @return address as a key
         * @throws Exception
         */
        @Override
        public String getKey(Tuple2<String, Float> lumenData) throws Exception {
            return lumenData.f0;
        }
    }
}

@@ -21,16 +21,75 @@
 ember;%0A%0A
+import org.apache.flink.api.common.functions.JoinFunction;%0A
 import o
@@ -1348,11 +1348,455 @@
  %7D%0A    %7D
+%0A%0A    /**%0A     * Implements a join function to aggregate mean from traffic and lumen data streams%0A     */%0A    public static final class EmberAggregateSensors implements JoinFunction%3CTuple2%3CString, Float%3E, Tuple2%3CString, Float%3E, Float%3E %7B%0A%0A        @Override%0A        public Float join(Tuple2%3CString, Float%3E trafficData, Tuple2%3CString, Float%3E lumenData) throws Exception %7B%0A            return Math.abs(lumenData.f1 - trafficData.f1);%0A        %7D%0A    %7D
 %0A%7D%0A

Add a simple and wrong data aggregation.