content: package GeoFlink.spatialOperators.tStats;
  
  import GeoFlink.spatialObjects.Point;
  import GeoFlink.spatialOperators.QueryConfiguration;
  import GeoFlink.spatialOperators.QueryType;
  import org.apache.flink.api.common.functions.FilterFunction;
  import org.apache.flink.api.java.functions.KeySelector;
  import org.apache.flink.api.java.tuple.Tuple4;
  import org.apache.flink.api.java.tuple.Tuple5;
  import org.apache.flink.streaming.api.datastream.DataStream;
  import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
  import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
  import org.apache.flink.streaming.api.windowing.time.Time;
  
  import java.util.Set;
  
  public class PointTStatsQuery extends TStatsQuery<Point> {
      public PointTStatsQuery(QueryConfiguration conf) {
          super.initializeKNNQuery(conf);
      }
  
      public DataStream<?> run(DataStream<Point> pointStream, Set<String> trajIDSet) {
  
          //--------------- Real-time - POINT -----------------//
          if (this.getQueryConfiguration().getQueryType() == QueryType.RealTime) {
              return realTime(pointStream, trajIDSet);
          }
  
          //--------------- Window-based - POINT -----------------//
          else if (this.getQueryConfiguration().getQueryType() == QueryType.WindowBased) {
              int windowSize = this.getQueryConfiguration().getWindowSize();
              int windowSlideStep = this.getQueryConfiguration().getSlideStep();
              return windowBased(pointStream, trajIDSet, windowSize, windowSlideStep);
          }
  
          else {
              throw new IllegalArgumentException("Not yet support");
          }
      }
  
      // REAL-TIME
      private DataStream<Tuple5<String, Double, Long, Double, Long>> realTime(DataStream<Point> pointStream, Set<String> trajIDSet){
  
          DataStream<Point> filteredStream = pointStream.filter(new FilterFunction<Point>() {
              @Override
              public boolean filter(Point point) throws Exception {
  
                  if (trajIDSet.size() > 0)
                      return ((trajIDSet.contains(point.objID)));
                  else
                      return true;
              }
          });
  
          return filteredStream.keyBy(new KeySelector<Point, String>() {
              @Override
              public String getKey(Point p) throws Exception {
                  return p.objID;
              }
          }).flatMap(new TStatsQueryFlatmapFunction());
      }
  
      // WINDOW BASED
      private DataStream<Tuple4<String, Double, Long, Double>> windowBased(DataStream<Point> pointStream, Set<String> trajIDSet, int windowSize, int windowSlideStep){
  
          // Spatial stream with Timestamps and Watermarks
          // Max Allowed Lateness: windowSize
          DataStream<Point> pointStreamWithTsAndWm =
                  pointStream.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<Point>(Time.seconds(windowSize)) {
                      @Override
                      public long extractTimestamp(Point p) {
                          return p.timeStampMillisec;
                      }
                  });
  
          DataStream<Point> filteredStream = pointStreamWithTsAndWm.filter(new FilterFunction<Point>() {
              @Override
              public boolean filter(Point point) throws Exception {
                  if (trajIDSet.size() > 0)
                      return ((trajIDSet.contains(point.objID)));
                  else
                      return true;
              }
          });
  
          return filteredStream.keyBy(new KeySelector<Point, String>() {
              @Override
              public String getKey(Point p) throws Exception {
                  return p.objID;
              }
          }).window(SlidingEventTimeWindows.of(Time.seconds(windowSize), Time.seconds(windowSlideStep)))
                  .apply(new TStatsQueryWFunction());
      }
  }
  