/*
 * 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 com.intel.flinkbench;

import com.intel.flinkbench.microbench.*;
import com.intel.flinkbench.util.BenchLogUtil;
import com.intel.hibench.streambench.common.ConfigLoader;
import com.intel.flinkbench.util.FlinkBenchConfig;
import com.intel.hibench.streambench.common.metrics.MetricsUtil;
import com.intel.hibench.streambench.common.metrics.KafkaReporter;

import com.intel.hibench.streambench.common.Platform;

public class RunBench {
    public static void main(String[] args) throws Exception {
        runAll(args);
    }

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

        if (args.length < 2)
            BenchLogUtil.handleError("Usage: RunBench <ConfigFile> <FrameworkName>");

        FlinkBenchConfig conf = new FlinkBenchConfig();

        ConfigLoader cl = new ConfigLoader(args[0]);
        
        KafkaReporter reporter = getReporter(cl);

        conf.master = cl.getProperty("hibench.streambench.brokerList");
        conf.zkHost = cl.getProperty("hibench.streambench.zookeeper.host");
        conf.workerCount = Integer.parseInt(cl.getProperty("hibench.streambench.storm.worker_count"));
        conf.benchName = cl.getProperty("hibench.streambench.benchname");
        conf.topic = cl.getProperty("hibench.streambench.topic_name");
        conf.consumerGroup = cl.getProperty("hibench.streambench.consumer_group");
        
        String benchName = conf.benchName;

        BenchLogUtil.logMsg("Benchmark starts.." + benchName +
                "   Frameworks:" + "Flink");

        if (benchName.equals("wordcount")) {
            conf.separator = cl.getProperty("hibench.streambench.separator");
            WordCount wordCount = new WordCount();
            wordCount.processStream(conf);
        } else if (benchName.equals("identity")) {
            Identity identity = new Identity();
            identity.processStream(reporter, conf);
        } else if (benchName.equals("sample")) {
            conf.prob = Double.parseDouble(cl.getProperty("hibench.streambench.prob"));
            Sample sample = new Sample();
            sample.processStream(conf);
        } else if (benchName.equals("project")) {
            conf.separator = cl.getProperty("hibench.streambench.separator");
            conf.fieldIndex = Integer.parseInt(cl.getProperty("hibench.streambench.field_index"));
            Projection project = new Projection();
            project.processStream(conf);
        } else if (benchName.equals("grep")) {
            conf.pattern = cl.getProperty("hibench.streambench.pattern");
            Grep grep = new Grep();
            grep.processStream(conf);
        } else if (benchName.equals("distinctcount")) {
            DistinctCount distinct = new DistinctCount();
            distinct.processStream(conf);
        } else if (benchName.equals("statistics")) {
            Statistics numeric = new Statistics();
            numeric.processStream(conf);
        }
    }

    private static KafkaReporter getReporter(ConfigLoader config) {
        String topic = config.getProperty("hibench.streambench.topic_name");
        String brokerList = config.getProperty("hibench.streambench.brokerList");
        long recordPerInterval = Long.parseLong(config.getProperty("hibench.streambench.prepare.periodic.recordPerInterval"));
        int intervalSpan = Integer.parseInt(config.getProperty("hibench.streambench.prepare.periodic.intervalSpan"));
        String reporterTopic = MetricsUtil.getTopic(Platform.FLINK, topic, recordPerInterval, intervalSpan);
        return new KafkaReporter(reporterTopic, brokerList);
    }
}

@@ -1155,16 +1155,79 @@
 eporter;
+%0Aimport com.intel.hibench.streambench.common.StreamBenchConfig;
 %0A%0Aimport
@@ -3915,40 +3915,37 @@
 rty(
-%22hibench.s
+S
 tream
-b
+B
 ench
-.topic_name%22
+Config.KAFKA_TOPIC
 );%0A 
@@ -3994,40 +3994,43 @@
 rty(
-%22hibench.s
+S
 tream
-b
+B
 ench
-.brokerList%22
+Config.KAFKA_BROKER_LIST
 );%0A 
@@ -4099,64 +4099,54 @@
 rty(
-%22hibench.s
+S
 tream
-b
+B
 ench
-.prepare.periodic.recordPerInterval%22
+Config.DATAGEN_RECORDS_PRE_INTERVAL
 ));%0A
@@ -4212,59 +4212,47 @@
 rty(
-%22hibench.s
+S
 tream
-b
+B
 ench
-.prepare.periodic.intervalSpan%22
+Config.DATAGEN_INTERVAL_SPAN
 ));%0A

modify record config.