bec/bec_widgets
1
0
Fork 0
mirror of https://github.com/bec-project/bec_widgets.git synced 2026-07-26 21:22:58 +02:00
Code Activity

fix: migrate _convert_endpointinfo to public method

Browse Source
This commit is contained in:
David Perl
2026-06-27 08:51:46 +02:00
committed by wakonig_k
parent c13d8cf07b
commit 794de6a2d9
1 changed files with 3 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
bec_widgets/utils/bec_dispatcher.py
+3 -3
View File
@@ -211,7 +211,7 @@ class BECDispatcher:
self._registered_slots[qt_slot] = qt_slot
qt_slot = self._registered_slots[qt_slot]
self.client.connector.register(topics, cb=qt_slot, **kwargs)
topics_str, _ = self.client.connector._convert_endpointinfo(topics)
topics_str, _ = self.client.connector.extract_raw_endpoints_from_info(topics)
qt_slot.topics.update(set(topics_str))
else:
logger.warning(f"Attempted to create duplicate stream subscription for {topics=}")
@@ -240,7 +240,7 @@ class BECDispatcher:
if not self._registered_slots[connected_slot].topics:
del self._registered_slots[connected_slot]
def disconnect_topics(self, topics: Union[str, list]):
def disconnect_topics(self, topics: str | list):
"""
Disconnect all slots from a topic.
@@ -248,7 +248,7 @@ class BECDispatcher:
topics(Union[str, list]): The topic(s) to disconnect from
"""
self.client.connector.unregister(topics)
topics_str, _ = self.client.connector._convert_endpointinfo(topics)
topics_str, _ = self.client.connector.extract_raw_endpoints_from_info(topics)
remove_slots = []
for connected_slot in self._registered_slots.values():