bug #44: remove webrtcbin after client disconnection. sync webrtcbin …

…state to shared pipeline status

src/gst_signalling/gst_abstract_role.py

@@ -1,5 +1,6 @@
 import asyncio
 import logging
+from threading import Thread
 from typing import Any, Dict, NamedTuple, Optional
 
 import gi
@@ -69,8 +70,11 @@ async def on_end_session(session_id: str) -> None:
         Gst.init(None)
 
         self._pipeline = Gst.Pipeline.new()
+        # pipeline will only contain dynamically added webrtcbins
+        self._pipeline.set_state(Gst.State.PLAYING)
 
     def __del__(self) -> None:
+        self._pipeline.set_state(Gst.State.NULL)
         Gst.deinit()
 
     def make_send_sdp(self, sdp: Any, type: str, session_id: str) -> None:  # sdp is GstWebRTC.WebRTCSessionDescription
@@ -87,12 +91,8 @@ def init_webrtc(self, session_id: str) -> Gst.Element:
         assert webrtc
 
         webrtc.set_property("bundle-policy", "max-bundle")
-        # webrtc.set_property("stun-server", None)
-        # webrtc.set_property("turn-server", None)
-
         webrtc.connect("on-ice-candidate", self.send_ice_candidate_message, session_id)
 
-        self._pipeline.set_state(Gst.State.READY)
         self._pipeline.add(webrtc)
 
         return webrtc
@@ -131,13 +131,12 @@ def handle_ice_message(self, webrtc: Gst.Element, ice_msg: Dict[str, Any]) -> No
 
     async def close_session(self, session_id: str) -> None:
         self.logger.info("close session")
-        """
-        session = self.sessions.pop(session_id)
 
-        self.emit("close_session", session)
-
-        await session.pc.close()
-        """
+        session = self.sessions.pop(session_id)
+        self._pipeline.remove(session.pc)
+        session.pc.set_state(Gst.State.NULL)
+        # self.emit("close_session", session)
+        # await session.pc.close()
 
     async def send_sdp(self, session_id: str, sdp: Dict[str, Dict[str, str]]) -> None:
         await self.signalling.send_peer_message(session_id, "sdp", sdp)

src/gst_signalling/gst_producer.py

@@ -45,8 +45,7 @@ async def setup_session(self, session_id: str, peer_id: str) -> GstSession:
         pc = session.pc
         pc.connect("on-negotiation-needed", self.on_negotiation_needed, session_id)
 
-        self._pipeline.set_state(Gst.State.PLAYING)
-
+        pc.sync_state_with_parent()
         self.emit("new_session", session)
 
         return session