package com.github.projectflink.streaming;

import com.dataartisans.flink.example.eventpattern.Event;
import com.dataartisans.flink.example.eventpattern.EventsGenerator;
import com.dataartisans.flink.example.eventpattern.kafka.EventDeSerializer;
import com.github.projectflink.streaming.utils.PimpedKafkaSink;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;
import org.apache.flink.streaming.connectors.kafka.api.config.PartitionerWrapper;
import org.apache.flink.streaming.connectors.kafka.partitioner.SerializableKafkaPartitioner;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Properties;


public class KafkaGenerator {

	private static final Logger LOG = LoggerFactory.getLogger(KafkaGenerator.class);

	public static void main(String[] args) throws Exception {
		StreamExecutionEnvironment see = StreamExecutionEnvironment.getExecutionEnvironment();
		final ParameterTool pt = ParameterTool.fromArgs(args);
		see.getConfig().setGlobalJobParameters(pt);
		if(pt.has("p")) {
			see.setParallelism(pt.getInt("p"));
		}

		DataStreamSource<Event> src = see.addSource(new RichParallelSourceFunction<Event>() {

			int min;
			int max;
			EventsGenerator eg;
			boolean running = true;

			@Override
			public void open(Configuration parameters) throws Exception {
				super.open(parameters);

				int range = Integer.MAX_VALUE / getRuntimeContext().getNumberOfParallelSubtasks();
				this.min = range * getRuntimeContext().getIndexOfThisSubtask();
				this.max = min + range;
				eg = new EventsGenerator();
				LOG.info("Creating new EventsGenerator from {} to {}", min, max);
			}


			@Override
			public void run(SourceContext<Event> sourceContext) throws Exception {
				long generated = 0;
				long logfreq = pt.getInt("logfreq");
				long lastLog = -1;
				long lastElements = 0;

				while(running) {
					Event gen = eg.next(min, max);
					sourceContext.collect(gen);
				//	LOG.info("Generated event {}", gen);

					generated++;
					if (generated % logfreq == 0) {
						// throughput over entire time
						long now = System.currentTimeMillis();

						// throughput for the last "logfreq" elements
						if(lastLog == -1) {
							// init (the first)
							lastLog = now;
							lastElements = generated;
						} else {
							long timeDiff = now - lastLog;
							long elementDiff = generated - lastElements;
							double ex = (1000/(double)timeDiff);
							LOG.info("During the last {} ms, we generated {} elements. That's {} elements/second/core", timeDiff, elementDiff, elementDiff*ex);
							// reinit
							lastLog = now;
							lastElements = generated;
						}
					}

				}
			}

			@Override
			public void cancel() {
				running = false;
			}
		});

		String zkServer = pt.get("zookeeper");
		Properties props = pt.getProperties();

	/*	SerializableKafkaPartitioner part = new PimpedKafkaSink.LocalKafkaPartitioner(zkServer, pt.getRequired("topic"));
		props.put("partitioner.class", PartitionerWrapper.class.getCanonicalName());
		// java serialization will do the rest.
		props.put(PartitionerWrapper.SERIALIZED_WRAPPER_NAME, part); */

		src.addSink(new PimpedKafkaSink<Event>(pt.getRequired("brokerList"), pt.getRequired("topic"), props, new EventDeSerializer()));


		see.execute();
	}
}

@@ -2106,16 +2106,17 @@
 %09%09%09while
+ 
 (running
@@ -2115,24 +2115,24 @@
 (running) %7B%0A
-
 %09%09%09%09%09Event g
@@ -2188,16 +2188,17 @@
 t(gen);%0A
+%09
 %09%09%09%09//%09L
@@ -2233,16 +2233,16 @@
 , gen);%0A
-
 %0A%09%09%09%09%09ge
@@ -2432,16 +2432,17 @@
 %09%09%09%09%09%09if
+ 
 (lastLog
@@ -2666,17 +2666,19 @@
 1000
-/
+ / 
 (double)
 time
@@ -2673,16 +2673,17 @@
 (double)
+ 
 timeDiff
@@ -2818,17 +2818,19 @@
 mentDiff
-*
+ * 
 ex);%0A%09%09%09
@@ -3085,11 +3085,46 @@
 );%0A%0A
-%09/*
+%0A%09%09if(pt.has(%22localPartitioner%22)) %7B%0A%09%09
 %09Ser
@@ -3234,16 +3234,17 @@
 pic%22));%0A
+%09
 %09%09props.
@@ -3316,16 +3316,17 @@
 e());%0A%09%09
+%09
 // java 
@@ -3357,16 +3357,17 @@
 e rest.%0A
+%09
 %09%09props.
@@ -3424,11 +3424,12 @@
 rt);
- */
+%0A%09%09%7D
 %0A%0A%09%09

add option for local partitioner