From 1ea176334fadf25d1eba14f1f400a6d2b88f525b Mon Sep 17 00:00:00 2001 From: wyzula-jan Date: Fri, 3 Jul 2026 22:00:26 +0200 Subject: [PATCH] fix(dispatcher): select the wrapper subscribed to the requested topic on disconnect --- bec_widgets/utils/bec_dispatcher.py | 36 +++++++++++++------ tests/unit_tests/test_bec_dispatcher.py | 48 +++++++++++++++++++++++++ 2 files changed, 73 insertions(+), 11 deletions(-) diff --git a/bec_widgets/utils/bec_dispatcher.py b/bec_widgets/utils/bec_dispatcher.py index 0a0ad4be..6c2f64e0 100644 --- a/bec_widgets/utils/bec_dispatcher.py +++ b/bec_widgets/utils/bec_dispatcher.py @@ -265,28 +265,42 @@ class BECDispatcher: """ Disconnect a slot from a topic. + Among the wrappers matching the callback (and ``cb_info``, when given), the one + actually subscribed to the requested topics is released. If none of them is + subscribed to these topics, nothing is released and a warning names the actual + subscriptions — a blind unregister on the wrong wrapper would silently no-op. + Args: slot(Callable): The slot to disconnect topics EndpointInfo | str | list[EndpointInfo] | list[str]: A topic or list of topics to unsub from. cb_info(dict | None): When the same slot was registered multiple times with different cb_info payloads (e.g. per-signal async subscriptions), pass the - payload to select the exact registration; without it the first wrapper - matching the callback is used. + payload to narrow the candidate registrations before the topic match. """ # find the right slot to disconnect from ; # slot callbacks are wrapped in QtThreadSafeCallback objects, # but the slot we receive here is the original callable - for connected_slot in self._registered_slots.values(): - if connected_slot.cb != slot: - continue - if cb_info is not None and connected_slot.cb_info != cb_info: - continue - break - else: + topics_str, _ = self.client.connector.extract_raw_endpoints_from_info(topics) + requested = set(topics_str) + candidates = [ + connected_slot + for connected_slot in self._registered_slots.values() + if connected_slot.cb == slot and (cb_info is None or connected_slot.cb_info == cb_info) + ] + # Among wrappers for this callback, pick the one actually subscribed to the + # requested topics; a bare unregister on the wrong wrapper would silently no-op + # and leave the real subscription alive. + connected_slot = next((s for s in candidates if s.topics & requested), None) + if connected_slot is None: + if candidates: + logger.warning( + f"disconnect_slot({slot!r}): no registration matches topics " + f"{sorted(requested)}; the slot is subscribed to " + f"{sorted(set().union(*(s.topics for s in candidates)))}. Nothing released." + ) return self.client.connector.unregister(topics, cb=connected_slot) - topics_str, _ = self.client.connector.extract_raw_endpoints_from_info(topics) - self._registered_slots[connected_slot].topics.difference_update(set(topics_str)) + self._registered_slots[connected_slot].topics.difference_update(requested) if not self._registered_slots[connected_slot].topics: del self._registered_slots[connected_slot] diff --git a/tests/unit_tests/test_bec_dispatcher.py b/tests/unit_tests/test_bec_dispatcher.py index 651b8e37..0f6392a5 100644 --- a/tests/unit_tests/test_bec_dispatcher.py +++ b/tests/unit_tests/test_bec_dispatcher.py @@ -114,6 +114,54 @@ def test_dispatcher_disconnect_one(bec_dispatcher_w_connector, qtbot, send_msg_e cb2.assert_called_once() +@pytest.mark.parametrize("topics_msg_list", [(("topic1", dummy_msg),)]) +def test_dispatcher_disconnect_wrong_topic_is_safe_noop( + bec_dispatcher_w_connector, qtbot, send_msg_event +): + bec_dispatcher = bec_dispatcher_w_connector + cb1 = mock.Mock(spec=[]) + + bec_dispatcher.connect_slot(cb1, "topic1") + # disconnecting a topic the slot is NOT subscribed to must not release topic1 + bec_dispatcher.disconnect_slot(cb1, "topic-wrong") + assert len(bec_dispatcher.client.connector._managed_connection._topics_cb) == 1 + + send_msg_event.set() + qtbot.wait(10) + cb1.assert_called_once() + + bec_dispatcher.disconnect_slot(cb1, "topic1") + assert len(bec_dispatcher.client.connector._managed_connection._topics_cb) == 0 + + +@pytest.mark.parametrize("topics_msg_list", [(("topic1", dummy_msg), ("topic2", dummy_msg))]) +def test_dispatcher_disconnect_selects_wrapper_by_topic( + bec_dispatcher_w_connector, qtbot, send_msg_event +): + # The same callback registered twice with different cb_info produces two wrappers; + # disconnecting a topic must release it from the wrapper that actually holds it, + # not silently no-op on the first wrapper that matches the callback. + bec_dispatcher = bec_dispatcher_w_connector + cb1 = mock.Mock(spec=[]) + + bec_dispatcher.connect_slot(cb1, "topic1", cb_info={"scan": "a"}) + bec_dispatcher.connect_slot(cb1, "topic2", cb_info={"scan": "b"}) + assert len(bec_dispatcher.client.connector._managed_connection._topics_cb) == 2 + + bec_dispatcher.disconnect_slot(cb1, "topic2") + remaining = bec_dispatcher.client.connector._managed_connection._topics_cb + assert len(remaining) == 1 + assert "topic1" in remaining + + # release the remaining subscription explicitly instead of leaning on fixture teardown + bec_dispatcher.disconnect_slot(cb1, "topic1") + assert len(bec_dispatcher.client.connector._managed_connection._topics_cb) == 0 + + # unblock the fixture's message generator so connector.shutdown() can join + send_msg_event.set() + qtbot.wait(10) + + @pytest.mark.parametrize("topics_msg_list", [(("topic1", dummy_msg),)]) def test_dispatcher_2_cb_same_topic(bec_dispatcher_w_connector, qtbot, send_msg_event): # test for BEC issue #276