Created a Consumer for non blocking IO in benchmarks

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

@@ -13,13 +13,13 @@ import org.slf4j.LoggerFactory;
  *
  * @author andrea
  */
-public class Consumer implements Runnable {
+public class ConsumerBlocking implements Runnable {
     
-    private static final Logger LOG = LoggerFactory.getLogger(Consumer.class);
+    private static final Logger LOG = LoggerFactory.getLogger(ConsumerBlocking.class);
     
     private String m_clientID;
 
-    public Consumer(String clientID) {
+    public ConsumerBlocking(String clientID) {
         m_clientID = clientID;
     }
     

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

@@ -0,0 +1,85 @@
+package org.dna.mqtt.bechnmark;
+
+import java.net.URISyntaxException;
+import org.fusesource.mqtt.client.Future;
+import org.fusesource.mqtt.client.FutureConnection;
+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 ConsumerFuture implements Runnable {
+    
+    private static final Logger LOG = LoggerFactory.getLogger(ConsumerFuture.class);
+    
+    private String m_clientID;
+
+    public ConsumerFuture(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);
+        FutureConnection connection = mqtt.futureConnection();
+        Future<Void> futConn = connection.connect();
+        
+        try {
+            futConn.await();
+        } catch (Exception ex) {
+            LOG.error("Cant't CONNECT to the server", ex);
+            return;
+        }
+
+        
+        Topic[] topics = {new Topic("/topic", QoS.AT_MOST_ONCE)};
+        Future<byte[]> futSub = connection.subscribe(topics);
+        try {
+            byte[] qoses = futSub.await();
+            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 < Producer.PUB_LOOP; i++) {
+            Future<Message> futReceive = connection.receive();
+            try {
+                message = futReceive.await();
+            } 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());
+        }
+            
+        Future<Void> f4 =  connection.disconnect();
+        try {
+            LOG.info("Disconneting");
+            f4.await();
+            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

@@ -13,17 +13,21 @@ import java.util.concurrent.Executors;
  */
 public class Main {
     private static final int PUBLISHER_POOL_SIZE = 2;
+    private static final int NUM_CONCURRENT_PRODS = 2;
     
     public static void main(String[] args) throws InterruptedException {
-        ExecutorService pool = Executors.newFixedThreadPool(PUBLISHER_POOL_SIZE);
+        ExecutorService consumerPool = Executors.newFixedThreadPool(PUBLISHER_POOL_SIZE);
+        ExecutorService producerPool = Executors.newFixedThreadPool(PUBLISHER_POOL_SIZE);
         
-        pool.submit(new Consumer("Cons1"));
+        consumerPool.submit(new ConsumerBlocking("Cons1"));
         
         //force wait to let the consumer be registered before the producer
         Thread.sleep(1000);
+        int len = Producer.PUB_LOOP / NUM_CONCURRENT_PRODS;
+        producerPool.submit(new Producer("Prod1", 0, len));
+        producerPool.submit(new Producer("Prod2", len, len));
         
-        pool.submit(new Producer("Prod1"));
-        
-        pool.shutdown();
+        consumerPool.shutdown();
+        producerPool.shutdown();
     }
 }

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

@@ -21,8 +21,13 @@ public class Producer implements Runnable {
     
     public static final int PUB_LOOP = 1000000;
     
-    public Producer(String clientID) {
+    private static int m_starIndex;
+    private static int m_len;
+    
+    public Producer(String clientID, int start, int len) {
         m_clientID = clientID;
+        m_starIndex = start;
+        m_len = len;
     }
 
     public void run() {
@@ -45,7 +50,7 @@ public class Producer implements Runnable {
         }
         
         //TODO loop
-        for (int i = 0; i < PUB_LOOP; i++) {
+        for (int i = m_starIndex; i < m_starIndex + m_len; i++) {
             try {
 //                LOG.info("Publishing");
                 String payload = "Hello world MQTT!!" + i;