@@ -163,7 +163,8 @@ def on_disconnect_returned():
163163 # an unexpected-drop callback, it was queued first and consumed this
164164 # operation. Otherwise, complete the explicit disconnection path now.
165165 logger .info ("{}: MQTT disconnected" .format (self .name ))
166- self ._handle_mqtt_disconnected ()
166+ self .send_event_up (pipeline_events_base .DisconnectedEvent ())
167+ self ._complete_pending_connection_op_after_disconnect ()
167168
168169 try :
169170 self .transport .disconnect (clear_inflight = op .hard )
@@ -294,7 +295,25 @@ def _on_mqtt_message_received(self, topic, payload):
294295 def _on_mqtt_connection_dropped (self , cause ):
295296 """Handle a transport-reported unexpected connection loss."""
296297 logger .info ("{}: MQTT connection dropped unexpectedly: {}" .format (self .name , cause ))
297- self ._handle_mqtt_disconnected (cause )
298+ self .send_event_up (pipeline_events_base .DisconnectedEvent ())
299+ try :
300+ self ._reconcile_mqtt_operation_tracking_after_connection_drop ()
301+
302+ # Higher layers will see that we're disconnected and may reconnect as necessary.
303+ error = transport_exceptions .ConnectionDroppedError ("Unexpected disconnection" )
304+ error .__cause__ = cause
305+ self .report_background_exception (error )
306+ finally :
307+ # Completion callbacks can synchronously start work on a replacement connection.
308+ self ._complete_pending_connection_op_after_disconnect (cause )
309+
310+ @pipeline_thread .runs_on_pipeline_thread
311+ def _reconcile_mqtt_operation_tracking_after_connection_drop (self ):
312+ """Reconcile MQTT operation tracking with the connection recovery policy.
313+
314+ This cannot be encapsulated inside the MQTTTransport because it has to do with
315+ connection_retry policy.
316+ """
298317
299318 # If there is no connection retry, complete tracked MQTT operations as cancelled so
300319 # they do not remain pending indefinitely.
@@ -317,25 +336,9 @@ def _on_mqtt_connection_dropped(self, cause):
317336 )
318337 self .transport ._op_manager .stop_tracking_non_publish_operations ()
319338
320- # Higher layers will see that we're disconnected and may reconnect as necessary.
321- error = transport_exceptions .ConnectionDroppedError ("Unexpected disconnection" )
322- error .__cause__ = cause
323- self .report_background_exception (error )
324-
325339 @pipeline_thread .runs_on_pipeline_thread
326- def _handle_mqtt_disconnected (self , cause = None ):
327- """Apply disconnected-state effects on the pipeline thread.
328-
329- Called after either an unexpected transport callback or a successful explicit disconnect.
330-
331- :param Exception cause: The Exception that caused the disconnection, if any (optional)
332- """
333- # Send an event to tell other pipeline stages that we're disconnected. Do this before
334- # we do anything else (in case upper stages have any "are we connected" logic.)
335- # NOTE: Other stages rely on the fact that this occurs before any op that may be in
336- # progress is completed. Be careful with changing the order things occur here.
337- self .send_event_up (pipeline_events_base .DisconnectedEvent ())
338-
340+ def _complete_pending_connection_op_after_disconnect (self , cause = None ):
341+ """Complete a pending connection operation after disconnection effects are applied."""
339342 if self ._pending_connection_op :
340343
341344 connection_op = self ._pending_connection_op
0 commit comments