/**
 * Copyright 2016 Milinda Pathirage
 * <p>
 * Licensed 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
 * <p>
 * http://www.apache.org/licenses/LICENSE-2.0
 * <p>
 * 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.pathirage.fdbench.tools;

import com.beust.jcommander.JCommander;
import com.beust.jcommander.Parameter;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;
import java.util.Random;
import java.util.concurrent.locks.LockSupport;

public class SimpleKafkaProducer {

  private static Random random = new Random(System.currentTimeMillis());

  public static void main(String[] args) {
    Options options = new Options();
    new JCommander(options, args);

    System.out.println(String.format("Starting produce with message-size: %s, brokers: %s, message-rate: %s, topic: %s, and duration: %s seconds",
        options.messageSize, options.kafkaBrokers, options.messageRate, options.topic, options.duration));

    KafkaProducer<byte[], byte[]> kafkaProducer = new KafkaProducer<byte[], byte[]>(getProducerProperties(options.kafkaBrokers));

    long startTime = System.currentTimeMillis();
    int i = 0;
    while ((System.currentTimeMillis() - startTime) < options.duration * 1000) {
      long interval = (long) poissonRandomInterArrivalDelay((1 / options.messageRate) * 1000000000);
      // This is not a high accuracy sleep. But will work for microsecond sleep times
      // http://www.rationaljava.com/2015/10/measuring-microsecond-in-java.html
      System.out.println("Waiting for  " + interval / 1000000000 + "seconds");
      waitNanos(interval);
      kafkaProducer.send(new ProducerRecord<byte[], byte[]>(options.topic, generateRandomMessage(options.messageSize)));
      i++;
      if (i % 1000 == 0) {
        System.out.println("Sent " + i + " messages!");
      }
    }

    System.exit(0);
  }

  private static void waitNanos(long nanos){
    long start = System.nanoTime();
    long end=0;
    do{
      end = System.nanoTime();
    }while(start + nanos >= end);
  }

  private static Properties getProducerProperties(String brokers) {
    Properties producerProps = new Properties();

    producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokers);
    producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArraySerializer");
    producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArraySerializer");
    producerProps.put(ProducerConfig.CLIENT_ID_CONFIG, "simple-producer");

    return producerProps;
  }

  private static byte[] generateRandomMessage(int size) {
    byte[] payload = new byte[size];

    for (int i = 0; i < payload.length; i++) {
      payload[i] = (byte) (random.nextInt(26) + 65);
    }

    return payload;
  }

  private static double poissonRandomInterArrivalDelay(double L) {
    return Math.log(1.0-Math.random())/-L;
  }

  public static class Options {
    @Parameter(names = {"--msg-size", "-m"})
    public int messageSize = 100;

    @Parameter(names = {"--brokers", "-b"})
    public String kafkaBrokers;

    @Parameter(names = {"--msg-rate", "-r"})
    public int messageRate = 100;

    @Parameter(names = {"--duration", "-d"})
    public int duration = 120;

    @Parameter(names = {"--topic", "-t"})
    public String topic = "test";
  }
}

@@ -949,55 +949,8 @@
 dom;
-%0Aimport java.util.concurrent.locks.LockSupport;
 %0A%0Apu
@@ -1975,87 +1975,8 @@
 tml%0A
-      System.out.println(%22Waiting for  %22 + interval / 1000000000 + %22seconds%22);%0A
     
@@ -2296,16 +2296,17 @@
 g nanos)
+ 
 %7B%0A    lo
@@ -2347,17 +2347,19 @@
 long end
-=
+ = 
 0;%0A    d
@@ -2359,16 +2359,17 @@
 ;%0A    do
+ 
 %7B%0A      
@@ -2398,21 +2398,23 @@
 );%0A    %7D
+ 
 while
+ 
 (start +
@@ -3312,24 +3312,34 @@
 (1.0
--Math.random())/
+ - random.nextDouble()) / 
 -L;%0A

remove debug logs and move to Random class from Math.random