package com.bloomberg.bask.stub;

import org.json.JSONTokener;
import org.json.JSONArray;

import java.io.InputStream;
import java.io.FileInputStream;
import java.util.Properties;
import java.util.Random;
import java.util.concurrent.TimeUnit;

import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * This simple class is reads tweets from a file and then sends them on
 * to a Kafka queue. It is meant to stand in place for Solr which
 * would read tweets from Twitter's streaming API, store them, and then
 * send them to a Kafka queue.
 *
 * @author Kiran Hampal
 * @author Jack Markham
 */
public class TweetProducer {

    private static final String source = "src/test/resources/sample_tweets.json",
                                topic  = "documents",
                                broker = "192.168.50.4:9092";

    private final Logger logger = LoggerFactory.getLogger (TweetProducer.class);

    public static void main (String[] args) {
        TweetProducer producer = new TweetProducer ();
        producer.run ();
    }

    private ProducerConfig config () {
        Properties props = new Properties ();
        props.put ("metadata.broker.list", broker);
        props.put ("serializer.class", "kafka.serializer.StringEncoder");
        props.put ("producer.type", "sync");
        props.put ("request.required.acks", "1");
        return new ProducerConfig (props);
    }

    /**
     * Sends each JSON object in an array to Kafka, with a delay of 1-6 seconds
     */
    private void sendJSON (JSONArray jsonArray) throws Exception {
        Producer<String, String> producer = new Producer<String, String> (config ());
        for (int i = 0; i < jsonArray.length (); i++) {
            logger.debug ("Sending message {}", i);
            String msg = jsonArray.get (i).toString ();
            KeyedMessage<String, String> data = new KeyedMessage<String, String> (topic, msg);
            producer.send (data);
            TimeUnit.SECONDS.sleep (randomTime (6));
        }
        producer.close ();
    }

    /**
     * Returns a random number between max and 1
     */
    private int randomTime (int max) {
        Random random = new Random ();
        return random.nextInt (max) + 1;
    }

    private void run () {
        try {
            InputStream inputStream = new FileInputStream (source);
            sendJSON (new JSONArray (new JSONTokener (inputStream)));
        } catch (Exception e) {
            logger.error ("Unable to send message to Kafka", e);
        }
    }
}

@@ -237,16 +237,78 @@
 eUnit;%0A%0A
+import joptsimple.OptionParser;%0Aimport joptsimple.OptionSet;%0A%0A
 import k
@@ -886,73 +886,249 @@
 son%22
-,
+;
 %0A    
-                            topic  = %22documents%22,%0A         
+private String topic, broker;%0A%0A    private final Logger logger = LoggerFactory.getLogger (TweetProducer.class);%0A%0A    public static void main (String%5B%5D args) %7B%0A        OptionParser parser = new OptionParser ();%0A        parser.accepts (%0A
     
@@ -1127,32 +1127,42 @@
 s (%0A            
+%22broker%22,%0A
            broke
@@ -1160,166 +1160,318 @@
     
-broker = %22192.168.50.4:9092%22;%0A%0A    private final Logger logger = LoggerFactory.getLogger (TweetProducer.class);%0A%0A
+ %22The kafka broker in the form of Host:IP%22%0A        ).withRequiredArg ().ofType (String.class).required ();%0A        parser.accepts (%0A            %22topic%22,%0A            %22The topic to send messages to%22%0A        ).withRequiredArg ().ofType (String.class).required ();%0A 
     
-public static void main (String%5B%5D 
+   OptionSet options = parser.parse (
 args)
- %7B
+;%0A
 %0A   
@@ -1523,34 +1523,261 @@
 er (
-);%0A        producer.run ()
+%0A            (String) options.valueOf (%22broker%22),%0A            (String) options.valueOf (%22topic%22)%0A        );%0A        producer.run ();%0A    %7D%0A%0A    public TweetProducer (String broker, String topic) %7B%0A        this.broker = broker;%0A        this.topic = topic
 ;%0A  
@@ -2968,30 +2968,29 @@
     %7D%0A%0A    p
-rivate
+ublic
  void run ()

Allow commandline argument in the test/fake tweet producer