MQTT: Mosquitto broker queue test

bench.py

@@ -46,6 +46,7 @@ def run_clients(lang, *args):
     "py_redis": ["py", "--redis", "--unbuffered"],
     "py_redis_buffered": ["py", "--redis"],
     "py_zmq": ["py"],
+    "py_mqtt": ["py","--mqtt"],
     "go_redis": ["go", "--redis"],
     "go_zmq": ["go"],
 }
@@ -55,16 +56,17 @@ def run_clients(lang, *args):
     "py_redis": "red",
     "py_redis_buffered": "green",
     "py_zmq": "blue",
+    "py_mqtt": "coral",
     "go_redis": "violet",
     "go_zmq": "orange",
 }
 
 # Groups of runs mapped to each graph.
 plots = {
-    "two-queues-1": ["py_zmq", "py_redis"],
-    "two-queues-2": ["py_zmq", "py_redis", "py_redis_buffered"],
+    "two-queues-1": ["py_zmq", "py_redis", "py_mqtt"],
+    "two-queues-2": ["py_zmq", "py_redis", "py_redis_buffered", "py_mqtt"],
     "two-queues-3": ["py_zmq", "py_redis", "py_redis_buffered",
-                     "go_zmq", "go_redis"],
+                     "go_zmq", "go_redis", "py_mqtt"],
 }
 
 # Store all results in an output directory.
@@ -88,7 +90,7 @@ def run_clients(lang, *args):
     with open(output_path(names[0] + ".dat"), "r") as f:
         clients = len(f.read().split())
     with open(name + ".p", "w") as f:
-        lines = ", ".join([line % (l, l.replace("_", " "), colours[l])
+        lines = ", ".join([line % (output_path(l), l.replace("_", " "), colours[l])
                            for l in names])
         f.write(plotfile % {"name": name, "lines": lines, "clients": clients})
     Popen(["gnuplot", name + ".p"], stderr=PIPE)

mqtt_pubsub.py

@@ -0,0 +1,77 @@
+
+import time
+import mosquitto
+import os
+
+
+class MQTTPubSub(object):
+
+    def __init__(self, host="127.0.0.1"):
+        #create an mqtt client
+        mypid = os.getpid()
+        client_uniq = "two_queue_"+str(mypid)
+        self.pubsu = mosquitto.Mosquitto(client_uniq)
+        self.pubsu.connect(host)
+        self.pubsu.on_message = self.on_message
+        self.channels = set()
+        self.received = False
+        self.message  = ""
+
+    def on_message(self, obj, msg):
+        self.message = msg.topic+" "+msg.payload
+        self.received = True
+
+    def publish(self, channel, message):
+        self.pubsu.publish(channel, message)
+
+    def subscribe(self, channels):
+        if type(channels) == list:
+            for channel in channels:
+                self.channels.add(channel)
+                self.pubsu.subscribe(channel)
+        else:
+            self.channels.add(channels)
+            self.pubsu.subscribe(channels)
+
+    def unsubscribe(self, channels):
+        if type(channels) == list:
+            for channel in channels:
+                self.channels.remove(channel)
+                self.pubsu.unsubscribe(channel)
+        else:
+            self.channels.remove(channels)
+            self.pubsu.unsubscribe(channels)
+
+    def pubsub(self):
+        return self
+
+    def recv(self):
+        while self.received == False:
+            self.pubsu.loop();
+        self.received = False
+        return self.message
+
+    def listen(self):
+        while True:
+            channel, _, data = self.recv().partition(" ")
+            yield {"type": "message", "channel": channel, "data": data}
+
+
+if __name__ == "__main__":
+    receiver = MQTTPubSub(host="127.0.0.1")
+    pubsub = receiver.pubsub()
+    last = time.time()
+    messages = 0
+    pubsub.subscribe(["teste"])
+    try:
+        while True:
+            pubsub.publish("teste",pubsub.recv())
+            #print pubsub.recv()
+            messages += 1
+            now = time.time()
+            if now - last > 1:
+                print "%s msg/sec" % messages
+                last = now
+                messages = 0
+    except (KeyboardInterrupt, SystemExit):
+        pass

test_client.py

@@ -9,6 +9,7 @@
 
 import buffered_redis
 import zmq_pubsub
+import mqtt_pubsub
 
 
 def new_client():
@@ -21,6 +22,8 @@ def new_client():
             Client = redis.Redis
         else:
             Client = buffered_redis.BufferedRedis
+    elif args.mqtt:
+        Client = mqtt_pubsub.MQTTPubSub
     else:
         Client = zmq_pubsub.ZMQPubSub
     return Client(host=args.host)
@@ -96,6 +99,7 @@ def get_metrics():
     parser.add_argument("--message-size", type=int, default=20)
     parser.add_argument("--redis", action="store_true")
     parser.add_argument("--unbuffered", action="store_true")
+    parser.add_argument("--mqtt", action="store_true")
     parser.add_argument("--quiet", action="store_true")
     args = parser.parse_args()
     channels = [str(i) for i in range(args.num_channels)]