Added consumer part for the benchmark

benchmarking/src/main/java/org/dna/mqtt/bechnmark/Consumer.java

@@ -0,0 +1,80 @@
+package org.dna.mqtt.bechnmark;
+
+import java.net.URISyntaxException;
+import org.fusesource.mqtt.client.BlockingConnection;
+import org.fusesource.mqtt.client.MQTT;
+import org.fusesource.mqtt.client.Message;
+import org.fusesource.mqtt.client.QoS;
+import org.fusesource.mqtt.client.Topic;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ *
+ * @author andrea
+ */
+public class Consumer implements Runnable {
+    
+    static final int PUB_LOOP = 10000;
+    
+    private static final Logger LOG = LoggerFactory.getLogger(Consumer.class);
+    
+    private String m_clientID;
+
+    public Consumer(String clientID) {
+        m_clientID = clientID;
+    }
+    
+    public void run() {
+        MQTT mqtt = new MQTT();
+        try {
+//            mqtt.setHost("test.mosquitto.org", 1883);
+            mqtt.setHost("localhost", 1883);
+        } catch (URISyntaxException ex) {
+            LOG.error(null, ex);
+            return;
+        }
+        
+        mqtt.setClientId(m_clientID);
+        BlockingConnection connection = mqtt.blockingConnection();
+        try {
+            connection.connect();
+        } catch (Exception ex) {
+            LOG.error("Cant't CONNECT to the server", ex);
+            return;
+        }
+        
+        try {
+            Topic[] topics = {new Topic("/topic", QoS.AT_MOST_ONCE)};
+            byte[] qoses = connection.subscribe(topics);
+            LOG.info("Subscribed to topic");
+        } catch (Exception ex) {
+            LOG.error("Cant't PUSBLISH to the server", ex);
+            return;
+        }
+            
+        Message message = null;
+        for (int i = 0; i < PUB_LOOP; i++) {
+            try {
+                message = connection.receive();
+            } catch (Exception ex) {
+                LOG.error(null, ex);
+                return;
+            }
+            byte[] payload = message.getPayload();
+            StringBuffer sb = new StringBuffer().append("Topic: ").append(message.getTopic())
+                    .append(", payload: ").append(new String(payload));
+            LOG.info(sb.toString());
+        }
+            
+        try {
+            LOG.info("Disconneting");
+            connection.disconnect();
+            LOG.info("Disconnected");
+        } catch (Exception ex) {
+            LOG.error("Cant't DISCONNECT to the server", ex);
+        }
+        
+    }
+    
+}

benchmarking/src/main/java/org/dna/mqtt/bechnmark/Main.java

@@ -14,10 +14,15 @@ import java.util.concurrent.Executors;
 public class Main {
     private static final int PUBLISHER_POOL_SIZE = 2;
     
-    public static void main(String[] args) {
+    public static void main(String[] args) throws InterruptedException {
         ExecutorService pool = Executors.newFixedThreadPool(PUBLISHER_POOL_SIZE);
         
-        pool.submit(new Producer("Client1"));
+        pool.submit(new Consumer("Cons1"));
+        
+        //force wait to let the consumer be registered before the producer
+        Thread.sleep(1000);
+        
+        pool.submit(new Producer("Prod1"));
         
         pool.shutdown();
     }

benchmarking/src/main/java/org/dna/mqtt/bechnmark/Producer.java

@@ -1,7 +1,6 @@
 package org.dna.mqtt.bechnmark;
 
 import java.net.URISyntaxException;
-import java.util.logging.Level;
 import org.fusesource.mqtt.client.BlockingConnection;
 import org.fusesource.mqtt.client.MQTT;
 import org.fusesource.mqtt.client.QoS;
@@ -20,6 +19,8 @@ public class Producer implements Runnable {
     
     private String m_clientID;
     
+    static final int PUB_LOOP = 10000;
+    
     public Producer(String clientID) {
         m_clientID = clientID;
     }
@@ -44,12 +45,15 @@ public class Producer implements Runnable {
         }
         
         //TODO loop
-        try {
-            LOG.info("Publishing");
-            connection.publish("/topic", "Hello world MQTT!!".getBytes(), QoS.AT_MOST_ONCE, false);
-        } catch (Exception ex) {
-            LOG.error("Cant't PUSBLISH to the server", ex);
-            return;
+        for (int i = 0; i < PUB_LOOP; i++) {
+            try {
+//                LOG.info("Publishing");
+                String payload = "Hello world MQTT!!" + i;
+                connection.publish("/topic", payload.getBytes(), QoS.AT_MOST_ONCE, false);
+            } catch (Exception ex) {
+                LOG.error("Cant't PUBLISH to the server", ex);
+                return;
+            }
         }
         try {
             LOG.info("Disconneting");