Merge branch 'master' of git://github.com/koho/paho.mqtt.python into koho-master

examples/client_sub-class.py

@@ -25,6 +25,9 @@ class MyMQTTClass(mqtt.Client):
     def on_connect(self, mqttc, obj, flags, rc):
         print("rc: "+str(rc))
 
+    def on_connect_fail(self, mqttc, obj):
+        print("Connect failed")
+
     def on_message(self, mqttc, obj, msg):
         print(msg.topic+" "+str(msg.qos)+" "+str(msg.payload))
 

src/paho/mqtt/client.py

@@ -475,6 +475,8 @@ class Client(object):
       Decorator: @client.connect_callback() (```client``` is the name of the
         instance which this callback is being attached to)
 
+    on_connect_fail(client, userdata): called when the client failed to connect to the broker.
+
     on_disconnect(client, userdata, rc): called when the client disconnects from the broker.
       The rc parameter indicates the disconnection state:
         MQTT_ERR_SUCCESS: the callback was called in response to a disconnect() call.
@@ -689,6 +691,7 @@ class Client(object):
         # No default callbacks
         self._on_log = None
         self._on_connect = None
+        self._on_connect_fail = None
         self._on_subscribe = None
         self._on_message = None
         self._on_publish = None
@@ -1809,6 +1812,7 @@ class Client(object):
                 try:
                     self.reconnect()
                 except (socket.error, OSError, WebsocketConnectionError):
+                    self._handle_on_connect_fail()
                     if not retry_first_connection:
                         raise
                     self._easy_log(
@@ -1847,6 +1851,7 @@ class Client(object):
                     try:
                         self.reconnect()
                     except (socket.error, OSError, WebsocketConnectionError):
+                        self._handle_on_connect_fail()
                         self._easy_log(
                             MQTT_LOG_DEBUG, "Connection failed, retrying")
 
@@ -1960,6 +1965,25 @@ class Client(object):
             return func
         return decorator
 
+    @property
+    def on_connect_fail(self):
+        """If implemented, called when the client failed to connect
+        to the broker."""
+        return self._on_connect_fail
+
+    @on_connect_fail.setter
+    def on_connect_fail(self, func):
+        """ Define the connection failure callback implementation
+
+        Expected signature is:
+            on_connect_fail(client, userdata)
+
+        client:     the client instance for this callback
+        userdata:   the private user data as set in Client() or userdata_set()
+        """
+        with self._callback_mutex:
+            self._on_connect_fail = func
+
     @property
     def on_subscribe(self):
         """If implemented, called when the broker responds to a subscribe
@@ -3544,6 +3568,15 @@ class Client(object):
                         if not self.suppress_exceptions:
                             raise
 
+    def _handle_on_connect_fail(self):
+        if self.on_connect_fail:
+            with self._in_callback_mutex:
+                try:
+                    self.on_connect_fail(self, self._userdata)
+                except Exception as err:
+                    self._easy_log(
+                        MQTT_LOG_ERR, 'Caught exception in on_connect_fail: %s', err)
+
     def _thread_main(self):
         self.loop_forever(retry_first_connection=True)
 

tests/test_mqttv5.py

@@ -41,6 +41,7 @@ class Callbacks:
         self.unsubscribeds = []
         self.disconnecteds = []
         self.connecteds = []
+        self.conn_failures = []
 
     def __str__(self):
         return str(self.messages) + str(self.messagedicts) + str(self.publisheds) + \
@@ -54,6 +55,9 @@ class Callbacks:
         self.connecteds.append({"userdata": userdata, "flags": flags,
                                 "reasonCode": reasonCode, "properties": properties})
 
+    def on_connect_fail(self, client, userdata):
+        self.conn_failures.append({"userdata": userdata})
+
     def wait(self, alist, timeout=2):
         interval = 0.2
         total = 0
@@ -62,6 +66,9 @@ class Callbacks:
             total += interval
         return alist.pop(0)  # if len(alist) > 0 else None
 
+    def wait_connect_fail(self):
+        return self.wait(self.conn_failures, timeout=10)
+
     def wait_connected(self):
         return self.wait(self.connecteds)
 
@@ -105,6 +112,7 @@ class Callbacks:
         client.on_unsubscribe = self.unsubscribed
         client.on_message = self.on_message
         client.on_disconnect = self.on_disconnect
+        client.on_connect_fail = self.on_connect_fail
         client.on_log = self.on_log
 
 
@@ -224,6 +232,17 @@ class Test(unittest.TestCase):
         callback.clear()
         aclient.loop_stop()
 
+    def test_connect_fail(self):
+        clientid = "connection failure"
+
+        fclient, fcallback = self.new_client(clientid)
+
+        fclient.user_data_set(1)
+        fclient.connect_async("localhost", 1)
+        response = fcallback.wait_connect_fail()
+        self.assertEqual(response["userdata"], 1)
+        fclient.loop_stop()
+
     def test_retained_message(self):
         qos0topic = "fromb/qos 0"
         qos1topic = "fromb/qos 1"