/**
 * Hello world!
 *
 */
public class FlinkApp{

    private static final ObjectMapper jsonParser = new ObjectMapper();
    private static final String region = "cn-northwest-1";
    private static final String inputStreamName = "bot-webb-splunk-kinesis";

    private static final String FAIRNESS_FIREHOUSE = "fairness-firehouse";
    private static final String EDGEKV_FIREHOUSE = "bot-webb-splunk-firehose-edgekv";


    private static DataStream<String> createSourceFromStaticConfig(StreamExecutionEnvironment env) {
        Properties inputProperties = new Properties();
        inputProperties.setProperty(ConsumerConfigConstants.AWS_REGION, region);
        inputProperties.setProperty(ConsumerConfigConstants.STREAM_INITIAL_POSITION, "LATEST");

        return env.addSource(new FlinkKinesisConsumer<>(inputStreamName,
                new SimpleStringSchema(), inputProperties));
    }


    private static KinesisFirehoseSink<String> fairnessFirehoseSink() {
        Properties sinkProperties = new Properties();
        sinkProperties.setProperty(AWS_REGION, region);

        return KinesisFirehoseSink.<String>builder()
                .setFirehoseClientProperties(sinkProperties)
                .setSerializationSchema(new SimpleStringSchema())
                .setDeliveryStreamName(FAIRNESS_FIREHOUSE)
                .build();
    }

    private static KinesisFirehoseSink<String> edgeKVFirehoseSink() {
        Properties sinkProperties = new Properties();
        sinkProperties.setProperty(AWS_REGION, region);

        return KinesisFirehoseSink.<String>builder()
                .setFirehoseClientProperties(sinkProperties)
                .setSerializationSchema(new SimpleStringSchema())
                .setDeliveryStreamName(EDGEKV_FIREHOUSE)
                .build();
    }




    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        final OutputTag<String> edgeKVOutput = new OutputTag<String>("edgekv-output"){};

        DataStream<DataRecord> input = createSourceFromStaticConfig(env).map(data -> jsonParser.readValue(data, DataRecord.class)).name("xxxxxxx-kinesis source");

        final SingleOutputStreamOperator<String> mainDS = input.process(new ProcessFunction<DataRecord, String>() {

            @Override
            public void processElement(DataRecord dataRecord, ProcessFunction<DataRecord, String>.Context ctx, Collector<String> out) throws Exception {
                if (dataRecord.getMetadata() != null && dataRecord.getMetadata().size() > 0) {
                    List<String> destinations = dataRecord.getMetadata().get(0).get("destination");
                    if (destinations != null && destinations.contains("fairness") && destinations.contains("edgeKV")) {
                        // emit data to edgeKV DataStream
                        ctx.output(edgeKVOutput, dataRecord.toString());
                        // emit data to fairness DataStream
                        out.collect(dataRecord.toString());
                    } else if (destinations != null && destinations.contains("fairness")) {
                        out.collect(dataRecord.toString());
                    } else if (destinations != null && destinations.contains("edgeKV")) {
                        ctx.output(edgeKVOutput, dataRecord.toString());
                    }
                }
            }
        }).name("Data Processor");

        DataStream<String> sideOutput = mainDS.getSideOutput(edgeKVOutput);

        sideOutput.sinkTo(fairnessFirehoseSink()).name("sink to EdgeKV Firehouse");
        mainDS.sinkTo(edgeKVFirehoseSink()).name("sink to Fairness Firehouse");

        env.execute("Minimum Stock Price");
    }
}