fix: adopt Paho MQTT v2 callbacks

Migrate transport callbacks to Paho's version 2 API and classify connection and disconnect reasons by their documented semantics. Propagate broker-rejected SUBACKs through operation tracking, including early acknowledgements, and leave reconnect timing to the SDK.

Remove obsolete reconnect-delay and private thread workarounds now that Paho 2.1 is required.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: cartertinney

azure-iot-device/azure/iot/device/common/mqtt_transport.py

@@ -283,13 +283,15 @@ def on_complete(cancelled=False):
             logger.debug("{}({}): subscribing to {}".format(self.name, op.name, op.topic))
 
             @pipeline_thread.invoke_on_pipeline_thread_nowait
-            def on_complete(cancelled=False):
+            def on_complete(cancelled=False, error=None):
                 if cancelled:
                     op.complete(
                         error=pipeline_exceptions.OperationCancelled(
                             "Operation cancelled before SUBACK received"
                         )
                     )
+                elif error is not None:
+                    op.complete(error=error)
                 else:
                     logger.debug(
                         "{}({}): SUBACK received. completing op.".format(self.name, op.name)

azure-iot-device/azure/iot/device/common/pipeline/pipeline_stages_mqtt.py

@@ -28,7 +28,7 @@ classifiers = [
 dependencies = [
     "deprecation>=2.1.0,<3.0.0",
     "janus",
-    "paho-mqtt>=2.0.0,<3.0.0",
+    "paho-mqtt>=2.1.0,<3.0.0",
     "PySocks",
     "requests>=2.32.3,<3.0.0",
     "requests-unixsocket>=0.4.1",

pyproject.toml

@@ -602,7 +602,7 @@ def test_complete(self, mocker, stage, op):
         assert op.error is None
 
     @pytest.mark.it(
-        "Completes the operation with an OperationCancelled error upon cancellation of the MQTT unsubscribe by the MQTTTransport"
+        "Completes the operation with an OperationCancelled error upon cancellation of the MQTT publish by the MQTTTransport"
     )
     def test_complete_with_cancel(self, mocker, stage, op):
         # Begin publish
@@ -639,7 +639,7 @@ def op(self, mocker):
         )
 
     @pytest.mark.it("Performs an MQTT subscribe via the MQTTTransport")
-    def test_mqtt_publish(self, mocker, stage, op):
+    def test_mqtt_subscribe(self, mocker, stage, op):
         stage.run_op(op)
         assert stage.transport.subscribe.call_count == 1
         assert stage.transport.subscribe.call_args == mocker.call(
@@ -662,10 +662,23 @@ def test_complete(self, mocker, stage, op):
         assert op.error is None
 
     @pytest.mark.it(
-        "Completes the operation with an OperationCancelled error upon cancellation of the MQTT unsubscribe by the MQTTTransport"
+        "Completes the operation with an error received from the MQTT subscribe callback"
+    )
+    def test_complete_with_error(self, stage, op, arbitrary_exception):
+        stage.run_op(op)
+
+        assert not op.completed
+
+        stage.transport.subscribe.call_args[1]["callback"](error=arbitrary_exception)
+
+        assert op.completed
+        assert op.error is arbitrary_exception
+
+    @pytest.mark.it(
+        "Completes the operation with an OperationCancelled error upon cancellation of the MQTT subscribe by the MQTTTransport"
     )
     def test_complete_with_cancel(self, mocker, stage, op):
-        # Begin unsubscribe
+        # Begin subscribe
         stage.run_op(op)
 
         assert not op.completed
@@ -699,7 +712,7 @@ def op(self, mocker):
         )
 
     @pytest.mark.it("Performs an MQTT unsubscribe via the MQTTTransport")
-    def test_mqtt_publish(self, mocker, stage, op):
+    def test_mqtt_unsubscribe(self, mocker, stage, op):
         stage.run_op(op)
         assert stage.transport.unsubscribe.call_count == 1
         assert stage.transport.unsubscribe.call_args == mocker.call(
@@ -739,7 +752,7 @@ def test_complete_with_cancel(self, mocker, stage, op):
     @pytest.mark.it(
         "Completes the operation using the exception that was raised, if an exception was raised from the MQTTTransport"
     )
-    def test_publish_error(self, stage, op, arbitrary_exception):
+    def test_unsubscribe_error(self, stage, op, arbitrary_exception):
         stage.transport.unsubscribe.side_effect = arbitrary_exception
 
         stage.run_op(op)