From ae4132adba0714e8cad2ea4ceeab7eeb4bf700fe Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Stef=C3=A1n=20Freyr=20Stef=C3=A1nsson?=
 <stfs@activitystream.com>
Date: Thu, 18 Jul 2013 19:42:07 +0000
Subject: [PATCH] Adding a test producer application.

---
 .../rabbitmq/RabbitMQProducerMain.java        | 182 ++++++++++++++++++
 1 file changed, 182 insertions(+)
 create mode 100644 examples/src/main/java/druid/examples/rabbitmq/RabbitMQProducerMain.java

diff --git a/examples/src/main/java/druid/examples/rabbitmq/RabbitMQProducerMain.java b/examples/src/main/java/druid/examples/rabbitmq/RabbitMQProducerMain.java
new file mode 100644
index 00000000000..a070bfff929
--- /dev/null
+++ b/examples/src/main/java/druid/examples/rabbitmq/RabbitMQProducerMain.java
@@ -0,0 +1,182 @@
+package druid.examples.rabbitmq;
+
+import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.Connection;
+import com.rabbitmq.client.ConnectionFactory;
+import org.apache.commons.cli.*;
+
+import java.text.SimpleDateFormat;
+import java.util.*;
+
+/**
+ *
+ */
+public class RabbitMQProducerMain
+{
+  public static void main(String[] args)
+      throws Exception
+  {
+    // We use a List to keep track of option insertion order. See below.
+    final List<Option> optionList = new ArrayList<Option>();
+
+    optionList.add(OptionBuilder.withLongOpt("help")
+        .withDescription("display this help message")
+        .create("h"));
+    optionList.add(OptionBuilder.withLongOpt("hostname")
+        .hasArg()
+        .withDescription("the hostname of the AMQP broker [defaults to AMQP library default]")
+        .create("b"));
+    optionList.add(OptionBuilder.withLongOpt("port")
+        .hasArg()
+        .withDescription("the port of the AMQP broker [defaults to AMQP library default]")
+        .create("n"));
+    optionList.add(OptionBuilder.withLongOpt("username")
+        .hasArg()
+        .withDescription("username to connect to the AMQP broker [defaults to AMQP library default]")
+        .create("u"));
+    optionList.add(OptionBuilder.withLongOpt("password")
+        .hasArg()
+        .withDescription("password to connect to the AMQP broker [defaults to AMQP library default]")
+        .create("p"));
+    optionList.add(OptionBuilder.withLongOpt("vhost")
+        .hasArg()
+        .withDescription("name of virtual host on the AMQP broker [defaults to AMQP library default]")
+        .create("v"));
+    optionList.add(OptionBuilder.withLongOpt("exchange")
+        .isRequired()
+        .hasArg()
+        .withDescription("name of the AMQP exchange [required - no default]")
+        .create("e"));
+    optionList.add(OptionBuilder.withLongOpt("key")
+        .hasArg()
+        .withDescription("the routing key to use when sending messages [default: 'default.routing.key']")
+        .create("k"));
+    optionList.add(OptionBuilder.withLongOpt("type")
+        .hasArg()
+        .withDescription("the type of exchange to create [default: 'topic']")
+        .create("t"));
+    optionList.add(OptionBuilder.withLongOpt("durable")
+        .withDescription("if set, a durable exchange will be declared [default: not set]")
+        .create("d"));
+    optionList.add(OptionBuilder.withLongOpt("autodelete")
+        .withDescription("if set, an auto-delete exchange will be declared [default: not set]")
+        .create("a"));
+    optionList.add(OptionBuilder.withLongOpt("single")
+        .withDescription("if set, only a single message will be sent [default: not set]")
+        .create("s"));
+    optionList.add(OptionBuilder.withLongOpt("start")
+        .hasArg()
+        .withDescription("time to use to start sending messages from [default: 2010-01-01T00:00:00]")
+        .create());
+    optionList.add(OptionBuilder.withLongOpt("stop")
+        .hasArg()
+        .withDescription("time to use to send messages until (format: '2013-07-18T23:45:59') [default: current time]")
+        .create());
+    optionList.add(OptionBuilder.withLongOpt("interval")
+        .hasArg()
+        .withDescription("the interval to add to the timestamp between messages in seconds [default: 10]")
+        .create());
+    optionList.add(OptionBuilder.withLongOpt("delay")
+        .hasArg()
+        .withDescription("the delay between sending messages in milliseconds [default: 100]")
+        .create());
+
+    // An extremely silly hack to maintain the above order in the help formatting.
+    HelpFormatter formatter = new HelpFormatter();
+    // Add a comparator to the HelpFormatter using the ArrayList above to sort by insertion order.
+    formatter.setOptionComparator(new Comparator(){
+      @Override
+      public int compare(Object o1, Object o2)
+      {
+        // I know this isn't fast, but who cares! The list is short.
+        return optionList.indexOf(o1) - optionList.indexOf(o2);
+      }
+    });
+
+    // Now we can add all the options to an Options instance. This is dumb!
+    Options options = new Options();
+    for (Option option : optionList) {
+      options.addOption(option);
+    }
+
+    CommandLine cmd = null;
+
+    try{
+      cmd = new BasicParser().parse(options, args);
+
+    }
+    catch(ParseException e){
+      formatter.printHelp("RabbitMQProducerMain", e.getMessage(), options, null);
+      System.exit(1);
+    }
+
+    if(cmd.hasOption("h")) {
+      formatter.printHelp("RabbitMQProducerMain", options);
+      System.exit(2);
+    }
+
+    ConnectionFactory factory = new ConnectionFactory();
+
+    if(cmd.hasOption("b")){
+      factory.setHost(cmd.getOptionValue("b"));
+    }
+    if(cmd.hasOption("u")){
+      factory.setUsername(cmd.getOptionValue("u"));
+    }
+    if(cmd.hasOption("p")){
+      factory.setPassword(cmd.getOptionValue("p"));
+    }
+    if(cmd.hasOption("v")){
+      factory.setVirtualHost(cmd.getOptionValue("v"));
+    }
+    if(cmd.hasOption("n")){
+      factory.setPort(Integer.parseInt(cmd.getOptionValue("n")));
+    }
+
+    String exchange = cmd.getOptionValue("e");
+    String routingKey = "default.routing.key";
+    if(cmd.hasOption("k")){
+      routingKey = cmd.getOptionValue("k");
+    }
+
+    boolean durable = cmd.hasOption("d");
+    boolean autoDelete = cmd.hasOption("a");
+    String type = cmd.getOptionValue("t", "topic");
+    boolean single = cmd.hasOption("single");
+    int interval = Integer.parseInt(cmd.getOptionValue("interval", "10"));
+    int delay = Integer.parseInt(cmd.getOptionValue("delay", "100"));
+
+    SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss");
+    Date stop = sdf.parse(cmd.getOptionValue("stop", sdf.format(new Date())));
+
+    Random r = new Random();
+    Calendar timer = Calendar.getInstance();
+    timer.setTime(sdf.parse(cmd.getOptionValue("start", "2010-01-01T00:00:00")));
+
+    String msg_template = "{\"utcdt\": \"%s\", \"wp\": %d, \"gender\": \"%s\", \"age\": %d}";
+
+    Connection connection = factory.newConnection();
+    Channel channel = connection.createChannel();
+
+    channel.exchangeDeclare(exchange, type, durable, autoDelete, null);
+
+    do{
+      int wp = (10 + r.nextInt(90)) * 100;
+      String gender = r.nextBoolean() ? "male" : "female";
+      int age = 20 + r.nextInt(70);
+
+      String line = String.format(msg_template, sdf.format(timer.getTime()), wp, gender, age);
+
+      channel.basicPublish(exchange, routingKey, null, line.getBytes());
+
+      System.out.println("Sent message: " + line);
+
+      timer.add(Calendar.SECOND, interval);
+
+      Thread.sleep(delay);
+    }while((!single && stop.after(timer.getTime())));
+
+    connection.close();
+  }
+
+}
\ No newline at end of file