class Example {
    private static final String AWS_ES_INDEX_NAME = "milestone-aggregation-index";
    private static final String AWS_ES_DOCUMENT_TYPE = "milestone-aggregation-record";

    private static final String TIMER_NAME = "timerName";
    private static final String TIMER_START = "longTimer.start";
    private static final String TIMER_END = "longTimer.end";
    private static final String TIMER_PAUSE = "longTimer.pause";
    private static final String TIMER_RESUME = "longTimer.resume";
    private static final String INPUT_METRIC_NAME = "TimeWaitingForActivity";
    private static final String OUTPUT_METRIC_NAME = "TimeWaitingForActivity";

    @Inject
    @Named(EnvironmentModule.FLINK_CONFIG_KEY)
    private static Map<String, String> flinkConfig;

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

        Injector injector = Guice.createInjector(new EnvironmentModule(args), new KivaPickWorkflowProcessorModule());

        // Retrieve execution environment based on running cluster

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        env.enableCheckpointing(Integer.valueOf(flinkConfig.get(EnvironmentModule.CHECKPOINT_INTERVAL_MILLIS_KEY)));

        env.setStateBackend(injector.getInstance(FsStateBackend.class));

        // Get data resource from KinesisConnectorWrapper

        KinesisConnectorWrapper kinesisConnector = injector.getInstance(KinesisConnectorWrapper.class);

        kinesisConnector.addSourceToEnv(env);

        DataStream<MilestoneBatch> source = kinesisConnector.getSource();

        // Flatten each milestone in the batch to MilestoneEvent model

        DataStream<MilestoneEvent> metricEvent = source.flatMap(injector.getInstance(MilestoneBatchFlatMap.class))
                .filter(evt -> evt.getData() != null && !evt.getData().isEmpty()
                        && evt.getData().get(TIMER_NAME).equals(INPUT_METRIC_NAME));

        // Set water marker for stream data

        DataStream<MilestoneEvent> eventInOrder = metricEvent
                .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<MilestoneEvent>(Time
                        .seconds(Integer.valueOf(flinkConfig.get(EnvironmentModule.STREAM_OUT_OF_ORDER_TIME_KEY)))) {
                    private static final long serialVersionUID = 5795391437252719084L;

                    @Override
                    public long extractTimestamp(MilestoneEvent element) {
                        return element.getTimeStamp().getTime();
                    }
                });

        DataStream<MilestoneEvent> partitionedEvent = eventInOrder.keyBy(new KeySelector<MilestoneEvent, String>() {
            private static final long serialVersionUID = 699212529568331455L;

            @Override
            public String getKey(MilestoneEvent event) {
                return event.getPickerId() + event.getWarehouseId();
            }
        });
        // Define pattern

        Pattern<MilestoneEvent, ?> kivaWaitForActivityPattern = Pattern.<MilestoneEvent>begin("start")
                .where(new SimpleCondition<MilestoneEvent>() {
                    private static final long serialVersionUID = 5407961589198363269L;

                    @Override
                    public boolean filter(MilestoneEvent event) {
                        return event.getMilestoneType().equals(TIMER_START);
                    }
                }).next("pauseOrResume").oneOrMore().optional().consecutive()
                .where(new SimpleCondition<MilestoneEvent>() {
                    private static final long serialVersionUID = -1947909131821791451L;

                    @Override
                    public boolean filter(MilestoneEvent event) throws Exception {
                        return event.getMilestoneType().equals(TIMER_PAUSE)
                                || event.getMilestoneType().equals(TIMER_RESUME);
                    }
                }).next("end").where(new SimpleCondition<MilestoneEvent>() {
                    private static final long serialVersionUID = 1562355907068840844L;

                    @Override
                    public boolean filter(MilestoneEvent event) {
                        return event.getMilestoneType().equals(TIMER_END);
                    }
                }).within(Time.seconds(Integer.valueOf(flinkConfig.get(EnvironmentModule.PATTERN_DETECT_TIMEOUT_KEY))));

        PatternStream<MilestoneEvent> patternStream = CEP.pattern(partitionedEvent, kivaWaitForActivityPattern);

        DataStream<MilestoneAggregation> waitForActivityTime = patternStream
                .select(new PatternSelectFunction<MilestoneEvent, MilestoneAggregation>() {
                    private static final long serialVersionUID = 4832891120845616503L;

                    @Override
                    public MilestoneAggregation select(Map<String, List<MilestoneEvent>> pattern) throws Exception {

                        MilestoneEvent timerStart = Iterables.getFirst(pattern.get("start"), null);

                        List<MilestoneEvent> timerPauseOrResumeList = pattern.get("pauseOrResume");

                        MilestoneEvent timerEnd = Iterables.getFirst(pattern.get("end"), null);

                        long timeAccumulater = 0L;

                        // Accumulate timer start to first pause
                        List<MilestoneEvent> startList = new ArrayList<MilestoneEvent>();
                        List<MilestoneEvent> endList = new ArrayList<MilestoneEvent>();

                        startList.add(timerStart);
                        // deal with duplicated pause or resume
                        int eventIndex = 0;
                        while (timerPauseOrResumeList != null && eventIndex < timerPauseOrResumeList.size()) {
                            MilestoneEvent currentEvent = timerPauseOrResumeList.get(eventIndex);
                            String currentState = currentEvent.getMilestoneType();
                            int tmpIndex = eventIndex + 1;
                            if (currentState.equals(TIMER_PAUSE)) {
                                endList.add(currentEvent);
                                while (tmpIndex < timerPauseOrResumeList.size() && timerPauseOrResumeList.get(tmpIndex)
                                        .getMilestoneType().equals(currentState)) {
                                    tmpIndex++;
                                }
                            } else if (currentState.equals(TIMER_RESUME)) {
                                startList.add(currentEvent);
                                while (tmpIndex < timerPauseOrResumeList.size() && timerPauseOrResumeList.get(tmpIndex)
                                        .getMilestoneType().equals(currentState)) {
                                    tmpIndex++;
                                }
                            }
                            eventIndex = tmpIndex;
                        }
                        endList.add(timerEnd);
                        // accumulate time
                        eventIndex = 0;
                        while (eventIndex < Math.min(startList.size(), endList.size())) {
                            timeAccumulater += endList.get(eventIndex).getTimeStamp().getTime()
                                    - startList.get(eventIndex).getTimeStamp().getTime();
                            eventIndex++;
                        }

                        long timeDiffInMillis = TimeUnit.MILLISECONDS.toMillis(timeAccumulater);
                        Date processTime = new Date();

                        Map<String, String> data = new HashMap<String, String>();
                        data.put("pickStart", timerStart.getDate());
                        data.put("pickEnd", timerEnd.getDate());
                        return new MilestoneAggregation(timerStart.getWarehouseId(), timerStart.getPickerId(),
                                OUTPUT_METRIC_NAME, timeDiffInMillis, processTime, timerEnd.getDate(), data,
                                UUID.randomUUID());
                    }
                });

        // Add source to elasticsearch sink
        ElasticsearchJestSinkWrapper elasticsearchConnector = injector.getInstance(ElasticsearchJestSinkWrapper.class);

        elasticsearchConnector.addSourceToSink(waitForActivityTime, AWS_ES_INDEX_NAME, AWS_ES_DOCUMENT_TYPE);

//        source.print();
//        partitionedEvent.print();
//        waitForActivityTime.print();

        env.execute();
    }
}