#3706 refactor element signal tracking

so we can re-use the same code for screen capture and bus messages

xpra/codecs/gstreamer/capture.py

@@ -55,8 +55,10 @@ def create_pipeline(self, capture_element:str="ximagesrc"):
         if not self.setup_pipeline_and_bus(elements):
             raise RuntimeError("failed to setup gstreamer pipeline")
         self.sink   = self.pipeline.get_by_name("sink")
-        self.sink.connect("new-sample", self.on_new_sample)
-        self.sink.connect("new-preroll", self.on_new_preroll)
+        def sh(sig, handler):
+            self.element_connect(self.sink, sig, handler)
+        sh("new-sample", self.on_new_sample)
+        sh("new-preroll", self.on_new_preroll)
 
     def on_new_sample(self, _bus):
         sample = self.sink.emit("pull-sample")

xpra/codecs/gstreamer/codec_common.py

@@ -52,10 +52,8 @@ def init_context(self, encoding, width, height, colorspace, options=None):
         self.src.set_property("format", Gst.Format.TIME)
         #self.src.set_caps(Gst.Caps.from_string(CAPS))
         self.sink   = self.pipeline.get_by_name("sink")
-        self.sink_handlers = []
         def sh(sig, handler):
-            sid = self.sink.connect(sig, handler)
-            self.sink_handlers.append(sid)
+            self.element_connect(self.sink, sig, handler)
         sh("new-sample", self.on_new_sample)
         sh("new-preroll", self.on_new_preroll)
         self.start()
@@ -131,11 +129,6 @@ def get_type(self) -> str:
         return "gstreamer"
 
     def clean(self):
-        handlers = self.sink_handlers
-        if handlers and self.sink:
-            self.handlers = []
-            for x in handlers:
-                self.sink.disconnect(x)
         super().cleanup()
         self.width = 0
         self.height = 0

xpra/gst_pipeline.py

@@ -35,10 +35,10 @@ class Pipeline(GObject.GObject):
     def __init__(self):
         super().__init__()
         self.bus = None
-        self.bus_message_handler_id = None
         self.bitrate = -1
         self.pipeline = None
         self.pipeline_str = ""
+        self.element_handlers = {}
         self.start_time = 0
         self.state : str = "stopped"
         self.info : dict = {}
@@ -48,6 +48,20 @@ def __init__(self):
         self.emit_info_timer : int = 0
         self.file = None
 
+    def element_connect(self, element, sig, handler):
+        """ keeps track of signal ids so we can cleanup later """
+        sid = element.connect(sig, handler)
+        self.element_handlers.setdefault(element, []).append(sid)
+
+    def elements_disconnect(self):
+        handlers = self.element_handlers
+        if not handlers:
+            return
+        self.element_handlers = {}
+        for element, sids in handlers.items():
+            for sid in sids:
+                element.disconnect(sid)
+
     def update_state(self, state):
         log("update_state(%s)", state)
         self.state = state
@@ -105,7 +119,7 @@ def setup_pipeline_and_bus(self, elements):
             self.cleanup()
             return False
         self.bus = self.pipeline.get_bus()
-        self.bus_message_handler_id = self.bus.connect("message", self.on_message)
+        self.element_connect(self.bus, "message", self.on_message)
         self.bus.add_signal_watch()
         self.info["pipeline"] = self.pipeline_str
         return True
@@ -152,18 +166,14 @@ def stop(self):
     def cleanup(self):
         log("Pipeline.cleanup()")
         self.cancel_emit_info_timer()
+        self.elements_disconnect()
         self.stop()
         b = self.bus
         self.bus = None
         log("Pipeline.cleanup() bus=%s", b)
         if not b:
             return
         b.remove_signal_watch()
-        bmhid = self.bus_message_handler_id
-        log("Pipeline.cleanup() bus_message_handler_id=%s", bmhid)
-        if bmhid:
-            self.bus_message_handler_id = None
-            b.disconnect(bmhid)
         self.pipeline = None
         self.state = None
         self.info = {}