From fc92a8b671a6735a722f1da192121cddd5554eb6 Mon Sep 17 00:00:00 2001 From: Josenivaldo Benito Junior Date: Wed, 28 Aug 2013 13:55:52 -0300 Subject: [PATCH 1/8] Mosquitto Python client: initial support --- mqtt_pubsub.py | 61 ++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 61 insertions(+) create mode 100644 mqtt_pubsub.py diff --git a/mqtt_pubsub.py b/mqtt_pubsub.py new file mode 100644 index 0000000..f18d697 --- /dev/null +++ b/mqtt_pubsub.py @@ -0,0 +1,61 @@ + +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.pubsub = mqttc.connect(host) + #self.pub = context.socket(zmq.PUSH) + #self.pub.connect("tcp://%s:%s" % (host, 5562)) + #self.sub = context.socket(zmq.SUB) + #self.sub.connect("tcp://%s:%s" % (host, 5561)) + self.channels = set() + + def publish(self, channel, message): + self.pubsub.publish(channel, message) + + def subscribe(self, channels): + for channel in channels: + self.channels.add(channel) + self.pubsub.subscribe(channel) + + def unsubscribe(self, channels): + for channel in channels: + self.channels.remove(channel) + self.pubsub.unsubscribe(channel) + + def pubsub(self): + return self + + def listen(self): + while True: + channel, _, data = self.sub.recv().partition(" ") + yield {"type": "message", "channel": channel, "data": data} + + +def serve(quiet): + context = zmq.Context() + receiver = context.socket(zmq.PULL) + receiver.bind("tcp://*:%s" % 5562) + sender = context.socket(zmq.PUB) + sender.bind("tcp://*:%s" % 5561) + last = time.time() + messages = 0 + try: + while True: + sender.send(receiver.recv()) + if not quiet: + messages += 1 + now = time.time() + if now - last > 1: + print "%s msg/sec" % messages + last = now + messages = 0 + except (KeyboardInterrupt, SystemExit): + pass From 248b12e8cc1a568294f29a7e537fab5a3bd179ec Mon Sep 17 00:00:00 2001 From: Josenivaldo Benito Junior Date: Wed, 28 Aug 2013 14:45:21 -0300 Subject: [PATCH 2/8] Mosquitto support File was based on zmq_pubsub, now most converted to use Mosquitto. Still need to implement listen function. --- mqtt_pubsub.py | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/mqtt_pubsub.py b/mqtt_pubsub.py index f18d697..b85a1ff 100644 --- a/mqtt_pubsub.py +++ b/mqtt_pubsub.py @@ -10,13 +10,18 @@ def __init__(self, host="127.0.0.1"): #create an mqtt client mypid = os.getpid() client_uniq = "two_queue_"+str(mypid) - self.pubsub = mqttc.connect(host) + self.pubsub = mosquitto.Mosquitto(client_uniq) + self.pubsub.connect(host) #self.pub = context.socket(zmq.PUSH) #self.pub.connect("tcp://%s:%s" % (host, 5562)) #self.sub = context.socket(zmq.SUB) #self.sub.connect("tcp://%s:%s" % (host, 5561)) self.channels = set() + def on_message(mosq, obj, msg): + ret = msg.topic+" "+msg.payload + return ret + def publish(self, channel, message): self.pubsub.publish(channel, message) @@ -34,7 +39,7 @@ def pubsub(self): return self def listen(self): - while True: + while self.pubsub.loop() == 0: channel, _, data = self.sub.recv().partition(" ") yield {"type": "message", "channel": channel, "data": data} From 9354b8360ceace9319c74c71981d2b11a18743c2 Mon Sep 17 00:00:00 2001 From: Josenivaldo Benito Junior Date: Wed, 28 Aug 2013 17:06:14 -0300 Subject: [PATCH 3/8] Test Function for subscribe -still need to rewrite listen method. --- mqtt_pubsub.py | 34 ++++++++++++++-------------------- 1 file changed, 14 insertions(+), 20 deletions(-) diff --git a/mqtt_pubsub.py b/mqtt_pubsub.py index b85a1ff..203d6f4 100644 --- a/mqtt_pubsub.py +++ b/mqtt_pubsub.py @@ -10,8 +10,9 @@ def __init__(self, host="127.0.0.1"): #create an mqtt client mypid = os.getpid() client_uniq = "two_queue_"+str(mypid) - self.pubsub = mosquitto.Mosquitto(client_uniq) - self.pubsub.connect(host) + self.pubsu = mosquitto.Mosquitto(client_uniq) + self.pubsu.connect(host) + self.pubsu.on_message = self.on_message #self.pub = context.socket(zmq.PUSH) #self.pub.connect("tcp://%s:%s" % (host, 5562)) #self.sub = context.socket(zmq.SUB) @@ -20,47 +21,40 @@ def __init__(self, host="127.0.0.1"): def on_message(mosq, obj, msg): ret = msg.topic+" "+msg.payload + print("Message received on topic "+msg.topic+" with QoS "+str(msg.qos)+" and payload "+msg.payload) return ret def publish(self, channel, message): - self.pubsub.publish(channel, message) + self.pubsu.publish(channel, message) def subscribe(self, channels): for channel in channels: + print("channel "+channel) self.channels.add(channel) - self.pubsub.subscribe(channel) + self.pubsu.subscribe(channel) def unsubscribe(self, channels): for channel in channels: self.channels.remove(channel) - self.pubsub.unsubscribe(channel) + self.pubsu.unsubscribe(channel) def pubsub(self): return self def listen(self): - while self.pubsub.loop() == 0: + while self.pubsu.loop() == 0: channel, _, data = self.sub.recv().partition(" ") yield {"type": "message", "channel": channel, "data": data} -def serve(quiet): - context = zmq.Context() - receiver = context.socket(zmq.PULL) - receiver.bind("tcp://*:%s" % 5562) - sender = context.socket(zmq.PUB) - sender.bind("tcp://*:%s" % 5561) +if __name__ == "__main__": + receiver = MQTTPubSub(host="127.0.0.1") + pubsub = receiver.pubsub() last = time.time() messages = 0 + pubsub.subscribe(["teste"]) try: while True: - sender.send(receiver.recv()) - if not quiet: - messages += 1 - now = time.time() - if now - last > 1: - print "%s msg/sec" % messages - last = now - messages = 0 + print("Retorno: "+str(pubsub.pubsu.loop_read())) except (KeyboardInterrupt, SystemExit): pass From d1d44212987ee526471aef7774c13b46f43b665c Mon Sep 17 00:00:00 2001 From: Josenivaldo Benito Junior Date: Wed, 28 Aug 2013 17:44:12 -0300 Subject: [PATCH 4/8] Mosquitto: Python draft completed Instead of rewriting listen function, wrote a blocking recv function to behave similar as ZeroMQ recv. --- mqtt_pubsub.py | 28 ++++++++++++++++++++-------- 1 file changed, 20 insertions(+), 8 deletions(-) diff --git a/mqtt_pubsub.py b/mqtt_pubsub.py index 203d6f4..d939269 100644 --- a/mqtt_pubsub.py +++ b/mqtt_pubsub.py @@ -18,18 +18,18 @@ def __init__(self, host="127.0.0.1"): #self.sub = context.socket(zmq.SUB) #self.sub.connect("tcp://%s:%s" % (host, 5561)) self.channels = set() + self.received = True + self.message = "" - def on_message(mosq, obj, msg): - ret = msg.topic+" "+msg.payload - print("Message received on topic "+msg.topic+" with QoS "+str(msg.qos)+" and payload "+msg.payload) - return ret + 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): for channel in channels: - print("channel "+channel) self.channels.add(channel) self.pubsu.subscribe(channel) @@ -41,9 +41,15 @@ def unsubscribe(self, 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 self.pubsu.loop() == 0: - channel, _, data = self.sub.recv().partition(" ") + while True: + channel, _, data = self.recv().partition(" ") yield {"type": "message", "channel": channel, "data": data} @@ -55,6 +61,12 @@ def listen(self): pubsub.subscribe(["teste"]) try: while True: - print("Retorno: "+str(pubsub.pubsu.loop_read())) + pubsub.publish("teste",pubsub.recv()) + messages += 1 + now = time.time() + if now - last > 1: + print "%s msg/sec" % messages + last = now + messages = 0 except (KeyboardInterrupt, SystemExit): pass From 4fdcb1972dfb183f1469d773cec30c391187081b Mon Sep 17 00:00:00 2001 From: Josenivaldo Benito Junior Date: Wed, 28 Aug 2013 20:30:31 -0300 Subject: [PATCH 5/8] Mosquitto: Test client is able to run it Rewrite the subscribe to avoid error when subscribing. This allowed listen to work correct with test_client script. --- mqtt_pubsub.py | 29 +++++++++++++++++------------ 1 file changed, 17 insertions(+), 12 deletions(-) diff --git a/mqtt_pubsub.py b/mqtt_pubsub.py index d939269..b345d5c 100644 --- a/mqtt_pubsub.py +++ b/mqtt_pubsub.py @@ -13,30 +13,34 @@ def __init__(self, host="127.0.0.1"): self.pubsu = mosquitto.Mosquitto(client_uniq) self.pubsu.connect(host) self.pubsu.on_message = self.on_message - #self.pub = context.socket(zmq.PUSH) - #self.pub.connect("tcp://%s:%s" % (host, 5562)) - #self.sub = context.socket(zmq.SUB) - #self.sub.connect("tcp://%s:%s" % (host, 5561)) self.channels = set() - self.received = True + self.received = False self.message = "" def on_message(self, obj, msg): self.message = msg.topic+" "+msg.payload - self.received = True + self.received = True def publish(self, channel, message): self.pubsu.publish(channel, message) def subscribe(self, channels): - for channel in channels: - self.channels.add(channel) - self.pubsu.subscribe(channel) + 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): - for channel in channels: - self.channels.remove(channel) - self.pubsu.unsubscribe(channel) + 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 @@ -62,6 +66,7 @@ def listen(self): try: while True: pubsub.publish("teste",pubsub.recv()) + #print pubsub.recv() messages += 1 now = time.time() if now - last > 1: From eb3e8163d200307552b7b0126fb366062b330f61 Mon Sep 17 00:00:00 2001 From: Josenivaldo Benito Junior Date: Thu, 29 Aug 2013 12:32:00 -0300 Subject: [PATCH 6/8] MQTT: Convert test_client to support MQTT test --- test_client.py | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/test_client.py b/test_client.py index b9a0dd7..535bbeb 100755 --- a/test_client.py +++ b/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)] From 93bdc963604d0128a0d2085e250ba5f1f266afb1 Mon Sep 17 00:00:00 2001 From: Josenivaldo Benito Junior Date: Thu, 29 Aug 2013 12:56:21 -0300 Subject: [PATCH 7/8] MQTT: Bench support for Mosquitto client Bench now runs test_client with "--mqtt" option and plot results together ZeroMQ and Redis results --- bench.py | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/bench.py b/bench.py index 14de0ef..5d397fa 100755 --- a/bench.py +++ b/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. From 1c559a4975104c32e01a629081c306b18fae912f Mon Sep 17 00:00:00 2001 From: Josenivaldo Benito Junior Date: Thu, 29 Aug 2013 17:07:28 -0300 Subject: [PATCH 8/8] Bench: put output path in .dat files for gnuplot Gnuplot was not working while running bench.py. The .png files were being generated as null. The fact is bench.py was calling gnuplot from working directory and not from output directory. The result was gnuplot did not find that .dat files with data to plot. This fix avoids manual generation of .png file by issuing gnuplot command inside output directory. --- bench.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/bench.py b/bench.py index 5d397fa..cfaf61a 100755 --- a/bench.py +++ b/bench.py @@ -90,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)