fix(Ros.close): block until the WebSocket is actually torn down

Previously Ros.close() returned as soon as proto.send_close() had been
dispatched onto the reactor thread: it set wait_disconnect inside the
on_ready callback, before clientConnectionLost ever fired. Downstream
callers (compas_fab's RosClient.__exit__ -> Ros.close() -> immediately
construct a new Ros) could then race the previous socket's TCP teardown,
and on contended reactors the next Ros.run() occasionally tripped its
10s connection timeout.

close() now registers a one-shot listener on the factory's "close" event
(emitted from clientConnectionLost when the underlying TCP connection
has actually been lost) and only sets wait_disconnect from there. The
"closing" event still fires synchronously inside the on_ready callback
before send_close(), so handlers that depend on that ordering are
unaffected.

If the disconnect doesn't complete within the timeout the listener is
detached before raising RosTimeoutError, so a subsequent reconnect
cycle on the same factory won't leak a reference to this Ros instance.

Verified by a new test_close_blocks_until_disconnect case in
tests/ros1/test_lifecycle.py.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

Author: gonzalocasas

CHANGELOG.rst

@@ -14,6 +14,13 @@ Unreleased
 
 **Changed**
 
+* ``Ros.close()`` now blocks until the underlying WebSocket has actually torn
+  down — it waits for the factory's ``"close"`` event (emitted from
+  ``clientConnectionLost``) instead of returning as soon as ``send_close()`` is
+  dispatched on the reactor thread. Downstream callers that immediately
+  construct a new ``Ros`` no longer race the previous socket's cleanup. The
+  ``"closing"`` event semantics are unchanged.
+
 **Fixed**
 
 * ``TwistedEventLoopManager`` no longer leaks one ``PythonLoggingObserver`` per

src/roslibpy/ros.py

@@ -76,20 +76,44 @@ def _unset_connecting_flag(*args):
         self.factory.connect()
 
     def close(self, timeout=CONNECTION_TIMEOUT):
-        """Disconnect from ROS."""
-        if self.is_connected:
-            wait_disconnect = threading.Event()
+        """Disconnect from ROS.
+
+        Blocks until the underlying WebSocket has actually torn down
+        (``clientConnectionLost`` fired). Prior to 2.1, ``close()``
+        returned as soon as ``send_close()`` had been dispatched on the
+        reactor thread; callers that immediately constructed a new ``Ros``
+        could race the previous socket's cleanup, occasionally causing
+        the next ``run()`` to time out under contention.
+        """
+        if not self.is_connected:
+            return
+
+        wait_disconnect = threading.Event()
 
-            def _wrapper_callback(proto):
-                self.emit("closing")
-                proto.send_close()
-                wait_disconnect.set()
-                return proto
+        def _on_close(*_args):
+            wait_disconnect.set()
 
-            self.factory.on_ready(_wrapper_callback)
+        # `clientConnectionLost` on the factory emits the "close" event once
+        # the websocket has actually been torn down. Use `once` so the
+        # listener is auto-removed on first fire — protects against the
+        # ReconnectingClientFactory triggering a later non-manual close.
+        self.factory.once("close", _on_close)
+
+        def _wrapper_callback(proto):
+            self.emit("closing")
+            proto.send_close()
+            return proto
+
+        self.factory.on_ready(_wrapper_callback)
 
-            if not wait_disconnect.wait(timeout):
-                raise RosTimeoutError("Failed to disconnect to ROS")
+        if not wait_disconnect.wait(timeout):
+            # Detach the listener so it doesn't fire on a later reconnect
+            # cycle and leak a reference to this Ros instance.
+            try:
+                self.factory.off("close", _on_close)
+            except Exception:
+                pass
+            raise RosTimeoutError("Failed to disconnect to ROS")
 
     def run(self, timeout=CONNECTION_TIMEOUT):
         """Kick-starts a non-blocking event loop.

tests/ros1/test_lifecycle.py

@@ -10,6 +10,8 @@
 
 from __future__ import print_function
 
+import threading
+
 from roslibpy import Ros
 
 HOST = "127.0.0.1"
@@ -48,3 +50,26 @@ def test_log_observer_does_not_leak():
     )
 
 
+def test_close_blocks_until_disconnect():
+    """``Ros.close()`` must not return until the WebSocket is actually closed.
+
+    Before A2 the call returned once ``send_close`` had been dispatched on
+    the reactor thread — the protocol's ``onClose`` would fire shortly
+    after, leaving a brief window where ``is_connected`` could still read
+    True. Now ``close()`` blocks on ``clientConnectionLost`` so the
+    contract matches the name.
+    """
+    ros = Ros(URL)
+    ros.run()
+    assert ros.is_connected
+
+    closed_event = threading.Event()
+    ros.on("close", lambda _: closed_event.set())
+
+    ros.close()
+
+    # The "close" event must have fired by the time close() returned.
+    assert closed_event.is_set(), "close() returned before clientConnectionLost fired"
+    assert not ros.is_connected, "is_connected still True after close()"
+
+