diff --git a/src/roslibpy/core.py b/src/roslibpy/core.py index 2153dcc..3cbd8c4 100644 --- a/src/roslibpy/core.py +++ b/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( diff --git a/tests/test_topic_reconnect_roles.py b/tests/test_topic_reconnect_roles.py new file mode 100644 index 0000000..91c0df5 --- /dev/null +++ b/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()