Reconnect both roles of a topic that publishes and subscribes

src/roslibpy/core.py

@@ -230,6 +230,7 @@ def __init__(
 
         self._subscribe_id = None
         self._advertise_id = None
+        self._connect_messages = {}
 
         if self.compression is None:
             self.compression = "none"
@@ -294,7 +295,9 @@ def unsubscribe(self):
 
         # Do not try to reconnect when manually unsubscribing
         if self.reconnect_on_close:
-            self.ros.off("close", self._reconnect_topic)
+            self._connect_messages.pop("subscribe", None)
+            if not self._connect_messages:
+                self.ros.off("close", self._reconnect_topic)
 
         self.ros.off(self.name)
         self.ros.send_on_ready(Message({"op": "unsubscribe", "id": self._subscribe_id, "topic": self.name}))
@@ -348,17 +351,21 @@ def _reset_advertise_id(self, _proto):
         self._advertise_id = None
 
     def _connect_topic(self, message):
-        self._connect_message = message
-        self.ros.send_on_ready(message)
-
         if self.reconnect_on_close:
-            self.ros.on("close", self._reconnect_topic)
+            if not self._connect_messages:
+                self.ros.on("close", self._reconnect_topic)
+            self._connect_messages[message["op"]] = message
+        self.ros.send_on_ready(message)
 
     def _reconnect_topic(self, _proto):
         # Delay a bit the event hookup because
         #  1) _proto is not yet nullified, and
         #  2) reconnect anyway takes a few seconds
-        self.ros.call_later(1, lambda: self.ros.send_on_ready(self._connect_message))
+        def reconnect():
+            for message in list(self._connect_messages.values()):
+                self.ros.send_on_ready(message)
+
+        self.ros.call_later(1, reconnect)
 
     def unadvertise(self):
         """Unregister as a publisher for the topic."""
@@ -367,7 +374,9 @@ def unadvertise(self):
 
         # Do not try to reconnect when manually unadvertising
         if self.reconnect_on_close:
-            self.ros.off("close", self._reconnect_topic)
+            self._connect_messages.pop("advertise", None)
+            if not self._connect_messages:
+                self.ros.off("close", self._reconnect_topic)
 
         self.ros.send_on_ready(
             Message(

tests/test_topic_reconnect_roles.py

@@ -0,0 +1,58 @@
+from unittest.mock import Mock
+import pytest
+from roslibpy import Topic
+
+
+@pytest.mark.parametrize("first", ["subscribe", "advertise"])
+def test_dual_role_topic_reconnects_both_roles(first):
+    ros = Mock()
+    ros.id_counter = 1
+    delayed = []
+    ros.call_later.side_effect = lambda delay, callback: delayed.append(callback)
+    topic = Topic(ros, "/state", "std_msgs/String")
+    actions = {"subscribe": lambda: topic.subscribe(lambda _: None), "advertise": topic.advertise}
+    actions[first]()
+    actions["advertise" if first == "subscribe" else "subscribe"]()
+    assert sum(call.args[0] == "close" for call in ros.on.call_args_list) == 1
+    ros.send_on_ready.reset_mock()
+    topic._reconnect_topic(None)
+    for callback in delayed:
+        callback()
+    operations = [call.args[0]["op"] for call in ros.send_on_ready.call_args_list]
+    assert sorted(operations) == ["advertise", "subscribe"]
+
+
+@pytest.mark.parametrize("removed,remaining", [("unsubscribe", "advertise"), ("unadvertise", "subscribe")])
+def test_removing_one_role_preserves_other_reconnect(removed, remaining):
+    ros = Mock()
+    ros.id_counter = 1
+    callbacks = []
+    ros.call_later.side_effect = lambda delay, callback: callbacks.append(callback)
+    topic = Topic(ros, "/state", "std_msgs/String")
+    topic.subscribe(lambda _: None)
+    topic.advertise()
+    topic._reconnect_topic(None)
+    getattr(topic, removed)()
+    assert not any(call.args[0] == "close" for call in ros.off.call_args_list)
+    ros.send_on_ready.reset_mock()
+    for callback in callbacks:
+        callback()
+    assert [call.args[0]["op"] for call in ros.send_on_ready.call_args_list] == [remaining]
+
+
+def test_removing_both_roles_cancels_pending_reconnect():
+    ros = Mock()
+    ros.id_counter = 1
+    callbacks = []
+    ros.call_later.side_effect = lambda delay, callback: callbacks.append(callback)
+    topic = Topic(ros, "/state", "std_msgs/String")
+    topic.subscribe(lambda _: None)
+    topic.advertise()
+    topic._reconnect_topic(None)
+    topic.unsubscribe()
+    topic.unadvertise()
+    assert any(call.args == ("close", topic._reconnect_topic) for call in ros.off.call_args_list)
+    ros.send_on_ready.reset_mock()
+    for callback in callbacks:
+        callback()
+    ros.send_on_ready.assert_not_called()