Skip to content

Commit 63dea9a

Browse files
randomsyncrandomsync
committed
Add 'Subscribe' and 'Unsubscribe' keywords
Subscribe keyword returns a list of messages returned within the specified period. Unsubscribe removes the client's subscription so that no further messages are received.
1 parent 49c268d commit 63dea9a

File tree

6 files changed

+149
-2
lines changed

6 files changed

+149
-2
lines changed

‎doc/MQTTLibrary.html‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -457,7 +457,7 @@
457457
jQuery.extend({highlight:function(e,t,n,r){if(e.nodeType===3){var i=e.data.match(t);if(i){var s=document.createElement(n||"span");s.className=r||"highlight";var o=e.splitText(i.index);o.splitText(i[0].length);var u=o.cloneNode(true);s.appendChild(u);o.parentNode.replaceChild(s,o);return 1}}else if(e.nodeType===1&&e.childNodes&&!/(script|style)/i.test(e.tagName)&&!(e.tagName===n.toUpperCase()&&e.className===r)){for(var a=0;a<e.childNodes.length;a++){a+=jQuery.highlight(e.childNodes[a],t,n,r)}}return 0}});jQuery.fn.unhighlight=function(e){var t={className:"highlight",element:"span"};jQuery.extend(t,e);return this.find(t.element+"."+t.className).each(function(){var e=this.parentNode;e.replaceChild(this.firstChild,this);e.normalize()}).end()};jQuery.fn.highlight=function(e,t){var n={className:"highlight",element:"span",caseSensitive:false,wordsOnly:false};jQuery.extend(n,t);if(e.constructor===String){e=[e]}e=jQuery.grep(e,function(e,t){return e!=""});e=jQuery.map(e,function(e,t){return e.replace(/[-[\]{}()*+?.,\\^$|#\s]/g,"\\$&")});if(e.length==0){return this}var r=n.caseSensitive?"":"i";var i="("+e.join("|")+")";if(n.wordsOnly){i="\\b"+i+"\\b"}var s=new RegExp(i,r);return this.each(function(){jQuery.highlight(this,s,n.element,n.className)})}
458458
</script>
459459
<script type="text/javascript">
460-
libdoc = {"doc":"<p>A keyword library for Robot Framework. It provides keywords for performing various operations on an MQTT broker. See <a href=\"http://mqtt.org/\">http://mqtt.org/\x3c/a> for more details on MQTT specification.\x3c/p>","generated":"2015-02-22 21:37:30","inits":[],"keywords":[{"args":"broker, port=1883, client_id=, clean_session=True","doc":"<p>Connect to an MQTT broker. This is a pre-requisite step for publish and subscribe keywords.\x3c/p>\n<p><span class=\"name\">broker\x3c/span> MQTT broker host\x3c/p>\n<p><span class=\"name\">port\x3c/span> broker port (default 1883)\x3c/p>\n<p><span class=\"name\">client_id\x3c/span> if not specified, a random id is generated\x3c/p>\n<p><span class=\"name\">clean_session\x3c/span> specifies the clean session flag for the connection\x3c/p>\n<p>Examples:\x3c/p>\n<p>Connect to a broker with default port and client id\x3c/p>\n<table border=\"1\">\n<tr>\n<td>Connect\x3c/td>\n<td>127.0.0.1\x3c/td>\n\x3c/tr>\n\x3c/table>\n<p>Connect to a broker by specifying the port and client id explicitly\x3c/p>\n<table border=\"1\">\n<tr>\n<td>Connect\x3c/td>\n<td>127.0.0.1\x3c/td>\n<td>1883\x3c/td>\n<td>test.client\x3c/td>\n\x3c/tr>\n\x3c/table>\n<p>Connect to a broker with clean session flag set to false\x3c/p>\n<table border=\"1\">\n<tr>\n<td>Connect\x3c/td>\n<td>127.0.0.1\x3c/td>\n<td>clean_session=${false}\x3c/td>\n\x3c/tr>\n\x3c/table>","name":"Connect","shortdoc":"Connect to an MQTT broker. This is a pre-requisite step for publish"},{"args":"","doc":"<p>Disconnect from MQTT Broker.\x3c/p>\n<p>Example:\x3c/p>\n<table border=\"1\">\n<tr>\n<td>Disconnect\x3c/td>\n\x3c/tr>\n\x3c/table>","name":"Disconnect","shortdoc":"Disconnect from MQTT Broker."},{"args":"topic, message=None, qos=0, retain=False","doc":"<p>Publish a message to a topic with specified qos and retained flag. It is required that a connection has been established using <a href=\"#Connect\" class=\"name\">Connect\x3c/a> keyword before using this keyword.\x3c/p>\n<p><span class=\"name\">topic\x3c/span> topic to which the message will be published\x3c/p>\n<p><span class=\"name\">message\x3c/span> message payload to publish\x3c/p>\n<p><span class=\"name\">qos\x3c/span> qos of the message\x3c/p>\n<p><span class=\"name\">retain\x3c/span> retained flag\x3c/p>\n<p>Examples:\x3c/p>\n<table border=\"1\">\n<tr>\n<td>Publish\x3c/td>\n<td>test/test\x3c/td>\n<td>test message\x3c/td>\n<td>1\x3c/td>\n<td>${false}\x3c/td>\n\x3c/tr>\n\x3c/table>","name":"Publish","shortdoc":"Publish a message to a topic with specified qos and retained flag."},{"args":"topic, qos, payload, timeout=1","doc":"<p>Subscribe to a topic and validate that the specified payload is received within timeout. It is required that a connection has been established using <a href=\"#Connect\" class=\"name\">Connect\x3c/a> keyword. The payload can be specified as a python regular expression. If the specified payload is not received within timeout, an AssertionError is thrown.\x3c/p>\n<p><span class=\"name\">topic\x3c/span> topic to subscribe to\x3c/p>\n<p><span class=\"name\">qos\x3c/span> quality of service for the subscription\x3c/p>\n<p><span class=\"name\">payload\x3c/span> payload (message) that is expected to arrive\x3c/p>\n<p><span class=\"name\">timeout\x3c/span> time to wait for the payload to arrive\x3c/p>\n<p>Examples:\x3c/p>\n<table border=\"1\">\n<tr>\n<td>Subscribe And Validate\x3c/td>\n<td>test/test\x3c/td>\n<td>1\x3c/td>\n<td>test message\x3c/td>\n\x3c/tr>\n\x3c/table>","name":"Subscribe And Validate","shortdoc":"Subscribe to a topic and validate that the specified payload is"}],"name":"MQTTLibrary","named_args":true,"scope":"global","version":""};
460+
libdoc = {"doc":"<p>A keyword library for Robot Framework. It provides keywords for performing various operations on an MQTT broker. See <a href=\"http://mqtt.org/\">http://mqtt.org/\x3c/a> for more details on MQTT specification.\x3c/p>","generated":"2015-02-24 23:11:57","inits":[],"keywords":[{"args":"broker, port=1883, client_id=, clean_session=True","doc":"<p>Connect to an MQTT broker. This is a pre-requisite step for publish and subscribe keywords.\x3c/p>\n<p><span class=\"name\">broker\x3c/span> MQTT broker host\x3c/p>\n<p><span class=\"name\">port\x3c/span> broker port (default 1883)\x3c/p>\n<p><span class=\"name\">client_id\x3c/span> if not specified, a random id is generated\x3c/p>\n<p><span class=\"name\">clean_session\x3c/span> specifies the clean session flag for the connection\x3c/p>\n<p>Examples:\x3c/p>\n<p>Connect to a broker with default port and client id\x3c/p>\n<table border=\"1\">\n<tr>\n<td>Connect\x3c/td>\n<td>127.0.0.1\x3c/td>\n\x3c/tr>\n\x3c/table>\n<p>Connect to a broker by specifying the port and client id explicitly\x3c/p>\n<table border=\"1\">\n<tr>\n<td>Connect\x3c/td>\n<td>127.0.0.1\x3c/td>\n<td>1883\x3c/td>\n<td>test.client\x3c/td>\n\x3c/tr>\n\x3c/table>\n<p>Connect to a broker with clean session flag set to false\x3c/p>\n<table border=\"1\">\n<tr>\n<td>Connect\x3c/td>\n<td>127.0.0.1\x3c/td>\n<td>clean_session=${false}\x3c/td>\n\x3c/tr>\n\x3c/table>","name":"Connect","shortdoc":"Connect to an MQTT broker. This is a pre-requisite step for publish"},{"args":"","doc":"<p>Disconnect from MQTT Broker.\x3c/p>\n<p>Example:\x3c/p>\n<table border=\"1\">\n<tr>\n<td>Disconnect\x3c/td>\n\x3c/tr>\n\x3c/table>","name":"Disconnect","shortdoc":"Disconnect from MQTT Broker."},{"args":"topic, message=None, qos=0, retain=False","doc":"<p>Publish a message to a topic with specified qos and retained flag. It is required that a connection has been established using <a href=\"#Connect\" class=\"name\">Connect\x3c/a> keyword before using this keyword.\x3c/p>\n<p><span class=\"name\">topic\x3c/span> topic to which the message will be published\x3c/p>\n<p><span class=\"name\">message\x3c/span> message payload to publish\x3c/p>\n<p><span class=\"name\">qos\x3c/span> qos of the message\x3c/p>\n<p><span class=\"name\">retain\x3c/span> retained flag\x3c/p>\n<p>Examples:\x3c/p>\n<table border=\"1\">\n<tr>\n<td>Publish\x3c/td>\n<td>test/test\x3c/td>\n<td>test message\x3c/td>\n<td>1\x3c/td>\n<td>${false}\x3c/td>\n\x3c/tr>\n\x3c/table>","name":"Publish","shortdoc":"Publish a message to a topic with specified qos and retained flag."},{"args":"topic, qos, timeout=1, limit=1","doc":"<p>Subscribe to a topic and return a list of message payloads received within the specified time.\x3c/p>\n<p><span class=\"name\">topic\x3c/span> topic to subscribe to\x3c/p>\n<p><span class=\"name\">qos\x3c/span> quality of service for the subscription\x3c/p>\n<p><span class=\"name\">timeout\x3c/span> duration of subscription\x3c/p>\n<p><span class=\"name\">limit\x3c/span> the max number of payloads that will be returned. Specify 0 for no limit\x3c/p>\n<p>Examples:\x3c/p>\n<p>Subscribe and get a list of all messages received within 5 seconds\x3c/p>\n<table border=\"1\">\n<tr>\n<td>${messages}=\x3c/td>\n<td>Subscribe\x3c/td>\n<td>test/test\x3c/td>\n<td>qos=1\x3c/td>\n<td>timeout=5\x3c/td>\n<td>limit=0\x3c/td>\n\x3c/tr>\n\x3c/table>\n<p>Subscribe and get 1st message received within 60 seconds\x3c/p>\n<table border=\"1\">\n<tr>\n<td>@{messages}=\x3c/td>\n<td>Subscribe\x3c/td>\n<td>test/test\x3c/td>\n<td>qos=1\x3c/td>\n<td>timeout=60\x3c/td>\n<td>limit=1\x3c/td>\n\x3c/tr>\n<tr>\n<td>Length should be\x3c/td>\n<td>${messages}\x3c/td>\n<td>1\x3c/td>\n<td>\x3c/td>\n<td>\x3c/td>\n<td>\x3c/td>\n\x3c/tr>\n\x3c/table>","name":"Subscribe","shortdoc":"Subscribe to a topic and return a list of message payloads received"},{"args":"topic, qos, payload, timeout=1","doc":"<p>Subscribe to a topic and validate that the specified payload is received within timeout. It is required that a connection has been established using <a href=\"#Connect\" class=\"name\">Connect\x3c/a> keyword. The payload can be specified as a python regular expression. If the specified payload is not received within timeout, an AssertionError is thrown.\x3c/p>\n<p><span class=\"name\">topic\x3c/span> topic to subscribe to\x3c/p>\n<p><span class=\"name\">qos\x3c/span> quality of service for the subscription\x3c/p>\n<p><span class=\"name\">payload\x3c/span> payload (message) that is expected to arrive\x3c/p>\n<p><span class=\"name\">timeout\x3c/span> time to wait for the payload to arrive\x3c/p>\n<p>Examples:\x3c/p>\n<table border=\"1\">\n<tr>\n<td>Subscribe And Validate\x3c/td>\n<td>test/test\x3c/td>\n<td>1\x3c/td>\n<td>test message\x3c/td>\n\x3c/tr>\n\x3c/table>","name":"Subscribe And Validate","shortdoc":"Subscribe to a topic and validate that the specified payload is"},{"args":"topic","doc":"<p>Unsubscribe the client from the specified topic.\x3c/p>\n<p><span class=\"name\">topic\x3c/span> topic to unsubscribe from\x3c/p>\n<p>Example:\x3c/p>\n<table border=\"1\">\n<tr>\n<td>Unsubscribe\x3c/td>\n<td>test/mqtt_test\x3c/td>\n\x3c/tr>\n\x3c/table>","name":"Unsubscribe","shortdoc":"Unsubscribe the client from the specified topic."}],"name":"MQTTLibrary","named_args":true,"scope":"global","version":"0.3.0"};
461461
</script>
462462
<title></title>
463463
</head>

‎src/MQTTLibrary/MQTTKeywords.py‎

Lines changed: 56 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -66,6 +66,45 @@ def publish(self, topic, message=None, qos=0, retain=False):
6666
% (topic, message, qos, retain), 'INFO')
6767
self._mqttc.publish(topic, message, int(qos), retain)
6868

69+
def subscribe(self, topic, qos, timeout=1, limit=1):
70+
""" Subscribe to a topic and return a list of message payloads received
71+
within the specified time.
72+
73+
`topic` topic to subscribe to
74+
75+
`qos` quality of service for the subscription
76+
77+
`timeout` duration of subscription
78+
79+
`limit` the max number of payloads that will be returned. Specify 0
80+
for no limit
81+
82+
Examples:
83+
84+
Subscribe and get a list of all messages received within 5 seconds
85+
| ${messages}= | Subscribe | test/test | qos=1 | timeout=5 | limit=0 |
86+
87+
Subscribe and get 1st message received within 60 seconds
88+
| @{messages}= | Subscribe | test/test | qos=1 | timeout=60 | limit=1 |
89+
| Length should be | ${messages} | 1 |
90+
91+
"""
92+
seconds = convert_time(timeout)
93+
self._messages = []
94+
limit = int(limit)
95+
96+
self.builtin.log('Subscribing to topic: %s' % topic, 'INFO')
97+
self._mqttc.subscribe(str(topic), int(qos))
98+
self._mqttc.on_message = self._on_message_list
99+
100+
timer_start = time.time()
101+
while time.time() < timer_start + seconds:
102+
if limit == 0 or len(self._messages) < limit:
103+
self._mqttc.loop()
104+
else:
105+
break
106+
return self._messages
107+
69108
def subscribe_and_validate(self, topic, qos, payload, timeout=1):
70109

71110
""" Subscribe to a topic and validate that the specified payload is
@@ -104,6 +143,18 @@ def subscribe_and_validate(self, topic, qos, payload, timeout=1):
104143
if not self._verified:
105144
raise AssertionError("The expected payload didn't arrive in the topic")
106145

146+
def unsubscribe(self, topic):
147+
148+
""" Unsubscribe the client from the specified topic.
149+
150+
`topic` topic to unsubscribe from
151+
152+
Example:
153+
| Unsubscribe | test/mqtt_test |
154+
155+
"""
156+
self._mqttc.unsubscribe(str(topic))
157+
107158
def disconnect(self):
108159

109160
""" Disconnect from MQTT Broker.
@@ -118,3 +169,8 @@ def _on_message(self, client, userdata, message):
118169
self.builtin.log('Received message: %s on topic: %s with QoS: %s'
119170
% (str(message.payload), message.topic, str(message.qos)), 'DEBUG')
120171
self._verified = re.match(self._payload, str(message.payload))
172+
173+
def _on_message_list(self, client, userdata, message):
174+
self.builtin.log('Received message: %s on topic: %s with QoS: %s'
175+
% (str(message.payload), message.topic, str(message.qos)), 'DEBUG')
176+
self._messages.append(message.payload)

‎src/MQTTLibrary/__init__.py‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -11,3 +11,4 @@ class MQTTLibrary(MQTTKeywords):
1111
"""
1212

1313
ROBOT_LIBRARY_SCOPE = 'GLOBAL'
14+
ROBOT_LIBRARY_VERSION = __version__

‎src/MQTTLibrary/version.py‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1 +1 @@
1-
VERSION = '0.2.0.dev1'
1+
VERSION = '0.3.0'

‎tests/keywords.txt‎

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -29,3 +29,24 @@
2929
| | Subscribe and Validate
3030
| | ... | ${topic} | ${qos} | ${message} | ${timeout}
3131
| | [Teardown] | Disconnect
32+
33+
| Subscribe and Get Messages
34+
| | [Arguments] | ${broker.uri}=${broker.uri} | ${port}=${broker.port}
35+
| | ... | ${client.id}=${client.id} | ${topic}=${topic}
36+
| | ... | ${qos}=1 | ${timeout}=1s
37+
| | ... | ${limit}=1
38+
| | Connect | ${broker.uri} | ${broker.port} | ${client.id} | ${false}
39+
| | @{messages} | Subscribe | ${topic} | ${qos} | ${timeout} | ${limit}
40+
| | [Teardown] | Disconnect
41+
| | [Return] | @{messages}
42+
43+
| Subscribe and Unsubscribe
44+
| | [Arguments] | ${broker.uri}=${broker.uri} | ${port}=${broker.port}
45+
| | ... | ${client.id}=${client.id} | ${topic}=${topic}
46+
| | ... | ${qos}=1 | ${timeout}=1s
47+
| | ... | ${limit}=1
48+
| | Connect | ${broker.uri} | ${broker.port} | ${client.id} | ${false}
49+
| | @{messages} | Subscribe | ${topic} | ${qos} | ${timeout} | ${limit}
50+
| | Unsubscribe | ${topic}
51+
| | [Teardown] | Disconnect
52+
| | [Return] | @{messages}

‎tests/pubsub.txt‎

Lines changed: 69 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -59,3 +59,72 @@
5959
| | ... | Subscribe to MQTT Broker and Validate | client.id=${client} | topic=${topic} | message=whatever
6060
| | Publish to MQTT Broker and Disconnect | topic=${topic} | message=${message} | qos=1
6161
| | Subscribe to MQTT Broker and Validate | client.id=${client} | topic=${topic} | message=${regex}
62+
63+
| Subscribe for the first time and validate that no messages are received
64+
| | Sleep | 1s
65+
| | ${time} | Get Time | epoch
66+
| | ${client} | Catenate | SEPARATOR=. | robot.mqtt | ${time}
67+
| | ${topic} | Set Variable | test/mqtt_test_sub
68+
| | @{messages} | Subscribe and Get Messages | client.id=${client} | topic=${topic} | timeout=5s
69+
| | LOG | ${messages}
70+
| | Length Should Be | ${messages} | 0
71+
72+
| Subscribe, publish 1 message and validate it is received
73+
| | Sleep | 1s
74+
| | ${time} | Get Time | epoch
75+
| | ${client} | Catenate | SEPARATOR=. | robot.mqtt | ${time}
76+
| | ${topic} | Set Variable | test/mqtt_test_sub
77+
| | Subscribe and Get Messages | client.id=${client} | topic=${topic}
78+
| | Publish to MQTT Broker and Disconnect | topic=${topic} | message=test message | qos=1
79+
| | @{messages} | Subscribe and Get Messages | client.id=${client} | topic=${topic}
80+
| | LOG | ${messages}
81+
| | Length Should Be | ${messages} | 1
82+
| | Should Be Equal As Strings | @{messages}[0] | test message
83+
84+
| Subscribe with no limit, publish multiple messages and validate they are received
85+
| | Sleep | 1s
86+
| | ${time} | Get Time | epoch
87+
| | ${client} | Catenate | SEPARATOR=. | robot.mqtt | ${time}
88+
| | ${topic} | Set Variable | test/mqtt_test_sub
89+
| | Subscribe and Get Messages | client.id=${client} | topic=${topic}
90+
| | Publish to MQTT Broker and Disconnect | topic=${topic} | message=test message1 | qos=1
91+
| | Publish to MQTT Broker and Disconnect | topic=${topic} | message=test message2 | qos=1
92+
| | Publish to MQTT Broker and Disconnect | topic=${topic} | message=test message3 | qos=1
93+
| | @{messages} | Subscribe and Get Messages | client.id=${client} | topic=${topic} | limit=0
94+
| | LOG | ${messages}
95+
| | Length Should Be | ${messages} | 3
96+
| | Should Be Equal As Strings | @{messages}[0] | test message1
97+
| | Should Be Equal As Strings | @{messages}[1] | test message2
98+
| | Should Be Equal As Strings | @{messages}[2] | test message3
99+
100+
| Subscribe with limit
101+
| | Sleep | 1s
102+
| | ${time} | Get Time | epoch
103+
| | ${client} | Catenate | SEPARATOR=. | robot.mqtt | ${time}
104+
| | ${topic} | Set Variable | test/mqtt_test_sub
105+
| | Subscribe and Get Messages | client.id=${client} | topic=${topic}
106+
| | Publish to MQTT Broker and Disconnect | topic=${topic} | message=test message1 | qos=1
107+
| | Publish to MQTT Broker and Disconnect | topic=${topic} | message=test message2 | qos=1
108+
| | Publish to MQTT Broker and Disconnect | topic=${topic} | message=test message3 | qos=1
109+
| | @{messages} | Subscribe and Get Messages | client.id=${client} | topic=${topic} | limit=1
110+
| | LOG | ${messages}
111+
| | Length Should Be | ${messages} | 1
112+
| | Should Be Equal As Strings | @{messages}[0] | test message1
113+
| | @{messages} | Subscribe and Get Messages | client.id=${client} | topic=${topic} | limit=2
114+
| | LOG | ${messages}
115+
| | Length Should Be | ${messages} | 2
116+
| | Should Be Equal As Strings | @{messages}[0] | test message2
117+
| | Should Be Equal As Strings | @{messages}[1] | test message3
118+
119+
| Unsubscribe and validate no messages are received
120+
| | Sleep | 1s
121+
| | ${time} | Get Time | epoch
122+
| | ${client} | Catenate | SEPARATOR=. | robot.mqtt | ${time}
123+
| | ${topic} | Set Variable | test/mqtt_test_sub
124+
| | Subscribe and Get Messages | client.id=${client} | topic=${topic}
125+
| | Publish to MQTT Broker and Disconnect | topic=${topic} | message=test message1 | qos=1
126+
| | @{messages} | Subscribe and Unsubscribe | client.id=${client} | topic=${topic}
127+
| | Publish to MQTT Broker and Disconnect | topic=${topic} | message=test message2 | qos=1
128+
| | @{messages} | Subscribe and Get Messages | client.id=${client} | topic=${topic}
129+
| | LOG | ${messages}
130+
| | Length Should Be | ${messages} | 0

0 commit comments

Comments
 (0)