/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.flink.streaming.examples.windowing;

import org.apache.flink.annotation.Internal;
import org.apache.flink.api.common.eventtime.Watermark;
import org.apache.flink.api.common.eventtime.WatermarkGenerator;
import org.apache.flink.api.common.eventtime.WatermarkGeneratorSupplier;
import org.apache.flink.api.common.eventtime.WatermarkOutput;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringEncoder;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.MemorySize;
import org.apache.flink.connector.file.sink.FileSink;
import org.apache.flink.connector.file.src.FileSource;
import org.apache.flink.core.fs.Path;
import org.apache.flink.formats.csv.CsvReaderFormat;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.DefaultRollingPolicy;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.windowing.assigners.EventTimeSessionWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
import java.nio.file.Files;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;

import static java.nio.file.StandardOpenOption.APPEND;
import static java.nio.file.StandardOpenOption.CREATE;

/**
 * An example of session windowing that keys events by ID and groups and counts them in session with
 * gaps of 3 milliseconds.
 */
public class SessionWindowing {
    private static final Logger LOG = LoggerFactory.getLogger(SessionWindowing.class);
    public static class Pojo {
        public String character;
        public Long event_time;
        public Integer val;

//        public Pojo() {
//        }
//
//        public Pojo(String character, Long event_time, Integer val) {
//            this.character = character;
//            this.event_time = event_time;
//            this.val = val;
//        }

        public String toString() {
            return this.character + "," + this.event_time + "," + this.val;
        }
    }

//    public static void write(final String s) throws IOException {
//        String filename= "/home/ubuntu/flink/flink-examples/flink-examples-streaming/target/out/log.txt";
//        FileWriter fw = new FileWriter(filename,true); //the true will append the new data
//        fw.write(s + "\n");//appends the string to the file
//        fw.close();
//    }

    @Internal
    public static class MyPerElementWatermarkGenerator implements WatermarkGenerator<Pojo> {

        private long maxTimestamp;

        public MyPerElementWatermarkGenerator() {
            maxTimestamp = Long.MIN_VALUE;
        }


        @Override
        public void onEvent(Pojo event, long eventTimestamp, WatermarkOutput output) {
//            if (eventTimestamp > maxTimestamp) {
//                maxTimestamp = eventTimestamp;
//                output.emitWatermark(new Watermark(maxTimestamp));
//            }
            if (event.event_time > maxTimestamp)
            {
                maxTimestamp = event.event_time;
            }

            LOG.debug("maxTimestamp: " + maxTimestamp);
            output.emitWatermark(new Watermark(maxTimestamp));
        }

        @Override
        public void onPeriodicEmit(WatermarkOutput output) {}

//        public WatermarkGeneratorSupplier<Pojo> getSupplier() {
//            return (ctx) -> new MyPerElementWatermarkGenerator();
//        }

    }

    public static void main(String[] args) throws Exception {

        final ParameterTool params = ParameterTool.fromArgs(args);
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.getConfig().setGlobalJobParameters(params);
	env.setRuntimeMode(RuntimeExecutionMode.STREAMING);
        env.setParallelism(1);

        final boolean fileOutput = params.has("output");

        String path = "/home/ubuntu/flink/flink-examples/flink-examples-streaming/target/csv";
        final CsvReaderFormat<Pojo> format = CsvReaderFormat.forPojo(Pojo.class);
        File csvfile = new File(path);
        final FileSource<Pojo> csv_source =
                FileSource.forRecordStreamFormat(format, Path.fromLocalFile(csvfile)).
		monitorContinuously(Duration.ofMillis(1)).build();
        
	DataStream<Pojo> source = env.fromSource(
                csv_source,
                new WatermarkStrategy<Pojo>() {
                    @Override
                    public WatermarkGenerator<Pojo> createWatermarkGenerator(
                            WatermarkGeneratorSupplier.Context context) {
                        return new MyPerElementWatermarkGenerator();
                    }
                }
                .withTimestampAssigner((event, timestamp) -> event.event_time), "csv-source");

        // .<Pojo>forBoundedOutOfOrderness(Duration.ofSeconds(20))
//        DataStream<Pojo> source = env.fromSource(
//                csv_source, WatermarkStrategy
//                        // .<Pojo>forMonotonousTimestamps()
//                        //.<Pojo>forBoundedOutOfOrderness(Duration.ofMillis(1))
//                        .<Pojo>noWatermarks()
//                        .withTimestampAssigner((event, timestamp) -> event.event_time),
//                    "csv-source");
        // DataStream<Pojo> source = source_no_watermark.assignTimestampsAndWatermarks();
        // .withTimestampAssigner((event, timestamp) -> event.event_time)
//        DataStream<Pojo> source_raw = env.fromSource(csv_source,
//                WatermarkStrategy.<Pojo>noWatermarks(), "csv-source");
//        DataStream<Pojo> source = source_raw.assignTimestampsAndWatermarks(
//                new WatermarkStrategy<Pojo>() {
//                    @Override
//                    public WatermarkGenerator<Pojo> createWatermarkGenerator(
//                            WatermarkGeneratorSupplier.Context context) {
//                        return (WatermarkGenerator<Pojo>) new MyPerElementWatermarkGenerator();
//                    }
//                }
//        );




        // readFile is a deprecated API
        // DataStream<Tuple3<String, Long, Integer>> myinput = env.readFile

//        final List<Tuple3<String, Long, Integer>> input = new ArrayList<>();
//
//        input.add(new Tuple3<>("a", 1L, 1));
//        input.add(new Tuple3<>("b", 1L, 1));
//        input.add(new Tuple3<>("b", 3L, 1));
//        input.add(new Tuple3<>("b", 5L, 1));
//        // input.add(new Tuple3<>("b", 1000L, 1));
//        input.add(new Tuple3<>("c", 6L, 1));
//        // We expect to detect the session "a" earlier than this point (the old
//        // functionality can only detect here when the next starts)
//        input.add(new Tuple3<>("a", 10L, 1));
//        // We expect to detect session "b" and "c" at this point as well
//        input.add(new Tuple3<>("c", 11L, 1));
//
//        DataStream<Tuple3<String, Long, Integer>> source =
//                env.addSource(
//                        new SourceFunction<Tuple3<String, Long, Integer>>() {
//                            private static final long serialVersionUID = 1L;
//
//                            @Override
//                            public void run(SourceContext<Tuple3<String, Long, Integer>> ctx)
//                                    throws Exception {
//                                for (Tuple3<String, Long, Integer> value : input) {
//                                    ctx.collectWithTimestamp(value, value.f1);
//                                    ctx.emitWatermark(new Watermark(value.f1 - 1));
//                                }
//                                ctx.emitWatermark(new Watermark(Long.MAX_VALUE));
//                            }
//
//                            @Override
//                            public void cancel() {}
//                        });
//
//        // We create sessions for each id with max timeout of 3 time units
//        DataStream<Tuple3<String, Long, Integer>> aggregated =
//                source.keyBy(value -> value.f0)
//                        .window(EventTimeSessionWindows.withGap(Time.milliseconds(3L)))
//                        .sum(2);
//
//        if (fileOutput) {
//            aggregated
//                    .sinkTo(
//                            FileSink.<Tuple3<String, Long, Integer>>forRowFormat(
//                                            new Path(params.get("output")),
//                                            new SimpleStringEncoder<>())
//                                    .withRollingPolicy(
//                                            DefaultRollingPolicy.builder()
//                                                    .withMaxPartSize(MemorySize.ofMebiBytes(1))
//                                                    .withRolloverInterval(Duration.ofSeconds(10))
//                                                    .build())
//                                    .build())
//                    .name("output");
//        }
        DataStream<Pojo> aggregated = source.keyBy(value -> value.character)
                .window(EventTimeSessionWindows.withGap(Time.milliseconds(3L)))
                .sum("val");
        if (fileOutput) {
            aggregated.sinkTo(FileSink.<Pojo>forRowFormat(
                    new Path(params.get("output")),
                    new SimpleStringEncoder<>())
                    .withRollingPolicy(
                            DefaultRollingPolicy.builder()
                                    .withMaxPartSize(MemorySize.ofMebiBytes(1))
                                    .withRolloverInterval(Duration.ofSeconds(10))
                                    .build())
                    .build())
                    .name("output");
        }
        else {
            System.out.println("Printing result to stdout. Use --output to specify output path.");
            aggregated.print();
        }

        env.execute();
        System.out.println("Program ends");
    }
}