|
| 1 | +import umqtt.robust |
| 2 | +import time |
| 3 | + |
| 4 | +# A common expectation of the robust client is that it should re-subscribe to |
| 5 | +# topics after a reconnect(). This feature adds to code size and isn't required |
| 6 | +# in all circumstances, so hasn't been included by default. |
| 7 | + |
| 8 | +# You can easily inherit from umqtt.robust.MQTTClient to add this feature... |
| 9 | + |
| 10 | + |
| 11 | +class MyMQTTClient(umqtt.robust.MQTTClient): |
| 12 | + def __init__(self, *args, **kwargs): |
| 13 | + super().__init__(*args, **kwargs) |
| 14 | + self.topics = [] |
| 15 | + |
| 16 | + def connect(self, clean_session=True): |
| 17 | + if not super().connect(clean_session): |
| 18 | + # Session was not restored - need to resubscribe |
| 19 | + for topic in self.topics: |
| 20 | + self.subscribe(topic) |
| 21 | + |
| 22 | + return False # Session was not restored |
| 23 | + |
| 24 | + return True # Session was restored |
| 25 | + |
| 26 | + def subscribe(self, topic): |
| 27 | + print("Subscribing to", topic) |
| 28 | + super().subscribe(topic) |
| 29 | + if topic not in self.topics: |
| 30 | + self.topics.append(topic) |
| 31 | + |
| 32 | + |
| 33 | +# Change the server to test on your MQTT broker |
| 34 | +c = MyMQTTClient("test_client", "localhost", keepalive=5) |
| 35 | +c.DEBUG = True |
| 36 | + |
| 37 | +c.set_callback(print) |
| 38 | + |
| 39 | +c.connect() |
| 40 | +c.subscribe(b"test/topic/a") |
| 41 | + |
| 42 | +c.publish(b"test/topic/a", b"message 1") |
| 43 | +c.wait_msg() |
| 44 | + |
| 45 | +# Connection breaks once keepalive expires |
| 46 | +time.sleep(8) |
| 47 | + |
| 48 | +c.publish(b"test/topic/a", b"message 2") # publish() doesn't detect OSError, message 2 is lost |
| 49 | +c.check_msg() # check_msg() detects OSError and will reconnect() |
| 50 | + |
| 51 | +c.publish(b"test/topic/a", b"message 3") |
| 52 | +c.wait_msg() |
0 commit comments