frontends (tools/webrtc): Better error handling and logging
This commit is contained in:
Goffi
parent
47564ec828
commit
2fce8ba7d6
1 changed files with 134 additions and 40 deletions
|
|
@ -296,6 +296,7 @@ class WebRTC:
|
|||
mline_index += 1
|
||||
|
||||
self.media_types = media_types
|
||||
log.debug(f"Media types mapping: {media_types}")
|
||||
|
||||
def _a_call(self, method, *args, **kwargs):
|
||||
"""Call an async method in main thread"""
|
||||
|
|
@ -335,6 +336,10 @@ class WebRTC:
|
|||
|
||||
@param candidate_string: The ice candidate string to be parsed.
|
||||
"""
|
||||
# Skip empty candidates
|
||||
if not candidate_string or candidate_string.strip() == "":
|
||||
return None
|
||||
|
||||
pattern = re.compile(
|
||||
r"candidate:(?P<foundation>\S+) (?P<component_id>\d+) (?P<transport>\S+) "
|
||||
r"(?P<priority>\d+) (?P<address>\S+) (?P<port>\d+) typ "
|
||||
|
|
@ -417,15 +422,21 @@ class WebRTC:
|
|||
|
||||
if bundle_ufrag and bundle_pwd:
|
||||
for mid in bundle_media:
|
||||
media = mid_media_map[mid]
|
||||
self.ufrag[media] = bundle_ufrag
|
||||
self.pwd[media] = bundle_pwd
|
||||
media = mid_media_map.get(mid)
|
||||
if media:
|
||||
self.ufrag[media] = bundle_ufrag
|
||||
self.pwd[media] = bundle_pwd
|
||||
log.debug(
|
||||
f"Set bundle ICE credentials for {media}: ufrag={bundle_ufrag[:4]}..."
|
||||
)
|
||||
else:
|
||||
log.warning(f"No media type found for mid {mid}")
|
||||
|
||||
def reset_instance(self):
|
||||
"""Inits or resets the instance variables to their default state."""
|
||||
self.role: str | None = None
|
||||
if self.pipeline is not None:
|
||||
self.pipeline.set_state(Gst.State.NULL)
|
||||
self.set_pipeline_state(Gst.State.NULL)
|
||||
self.pipeline = None
|
||||
self._remote_video_pad = None
|
||||
self.sid: str | None = None
|
||||
|
|
@ -635,6 +646,8 @@ class WebRTC:
|
|||
if self.webrtcbin is None:
|
||||
raise exceptions.InternalError("Can't get the pipeline.")
|
||||
|
||||
self.monitor_connection_state()
|
||||
|
||||
# If video or audio sources are not created, ``get_by_name`` will return None.
|
||||
self.video_src = self.pipeline.get_by_name("video_src")
|
||||
self.muted_src = self.pipeline.get_by_name("muted_src")
|
||||
|
|
@ -716,30 +729,68 @@ class WebRTC:
|
|||
def create_data_channel(self, dc_data: SourcesDataChannel | SinksDataChannel) -> None:
|
||||
"""Create a Data Channel and connect relevant callbacks."""
|
||||
assert self.pipeline is not None
|
||||
if isinstance(dc_data, SourcesDataChannel):
|
||||
# Data channel configuration for compatibility with browser defaults
|
||||
data_channel_options = Gst.Structure.new_empty("data-channel-options")
|
||||
data_channel_options.set_value("ordered", True)
|
||||
data_channel_options.set_value("protocol", "")
|
||||
match dc_data:
|
||||
case SourcesDataChannel():
|
||||
# Data channel configuration for compatibility with browser defaults
|
||||
data_channel_options = Gst.Structure.new_empty("data-channel-options")
|
||||
data_channel_options.set_value("ordered", True)
|
||||
data_channel_options.set_value("protocol", "")
|
||||
|
||||
# Create the data channel
|
||||
self.pipeline.set_state(Gst.State.READY)
|
||||
self.data_channels[dc_data.name] = data_channel = self.webrtcbin.emit(
|
||||
"create-data-channel", dc_data.name, data_channel_options
|
||||
)
|
||||
if data_channel is None:
|
||||
log.error("Failed to create data channel")
|
||||
return
|
||||
data_channel.connect("on-open", dc_data.dc_open_cb)
|
||||
elif isinstance(dc_data, SinksDataChannel):
|
||||
self.webrtcbin.connect("on-data-channel", dc_data.dc_on_data_channel)
|
||||
else:
|
||||
raise ValueError("Only SourcesDataChannel or SinksDataChannel are allowed.")
|
||||
# Create the data channel
|
||||
self.pipeline.set_state(Gst.State.READY)
|
||||
self.data_channels[dc_data.name] = data_channel = self.webrtcbin.emit(
|
||||
"create-data-channel", dc_data.name, data_channel_options
|
||||
)
|
||||
if data_channel is None:
|
||||
log.error("Failed to create data channel")
|
||||
return
|
||||
data_channel.connect("on-open", dc_data.dc_open_cb)
|
||||
case SinksDataChannel():
|
||||
self.webrtcbin.connect("on-data-channel", dc_data.dc_on_data_channel)
|
||||
case _:
|
||||
raise ValueError(
|
||||
"Only SourcesDataChannel or SinksDataChannel are allowed."
|
||||
)
|
||||
|
||||
def set_pipeline_state(self, state: Gst.State) -> bool:
|
||||
"""Safely set pipeline state with verification.
|
||||
|
||||
@param state: The target state
|
||||
@return: True if state change was successful
|
||||
"""
|
||||
if not self.pipeline:
|
||||
log.error("Pipeline not initialized")
|
||||
return False
|
||||
|
||||
ret = self.pipeline.set_state(state)
|
||||
if ret == Gst.StateChangeReturn.FAILURE:
|
||||
log.error(f"Failed to change pipeline state to {state}")
|
||||
return False
|
||||
|
||||
__, pending_state = self.pipeline.get_state(Gst.CLOCK_TIME_NONE)[:2]
|
||||
if pending_state != Gst.State.VOID_PENDING:
|
||||
log.warning(f"State change to {state} still pending")
|
||||
|
||||
return True
|
||||
|
||||
def monitor_connection_state(self):
|
||||
"""Monitor WebRTC connection state"""
|
||||
|
||||
def on_notify_connection_state(pspect, param):
|
||||
state = self.webrtcbin.get_property("connection-state")
|
||||
log.info(f"WebRTC connection state: {state}")
|
||||
|
||||
if state == GstWebRTC.WebRTCPeerConnectionState.FAILED:
|
||||
log.error("WebRTC connection failed")
|
||||
elif state == GstWebRTC.WebRTCPeerConnectionState.CONNECTED:
|
||||
log.info("WebRTC connection established")
|
||||
|
||||
self.webrtcbin.connect("notify::connection-state", on_notify_connection_state)
|
||||
|
||||
def start_pipeline(self) -> None:
|
||||
"""Starts the GStreamer pipeline."""
|
||||
log.debug("starting the pipeline")
|
||||
self.pipeline.set_state(Gst.State.PLAYING)
|
||||
self.set_pipeline_state(Gst.State.PLAYING)
|
||||
|
||||
def on_negotiation_needed(self, webrtc):
|
||||
"""Initiate SDP offer when negotiation is needed."""
|
||||
|
|
@ -933,13 +984,27 @@ class WebRTC:
|
|||
self.pipeline.add(q, conv, videoscale, capsfilter)
|
||||
|
||||
self.pipeline.sync_children_states()
|
||||
|
||||
ret = pad.link(q.get_static_pad("sink"))
|
||||
if ret != Gst.PadLinkReturn.OK:
|
||||
log.error(f"Error linking pad: {ret}")
|
||||
q.link(conv)
|
||||
conv.link(videoscale)
|
||||
videoscale.link(capsfilter)
|
||||
self.link_element_or_pad(capsfilter.link, remote_video_sink)
|
||||
log.error(f"Error linking pad to queue: {ret}")
|
||||
return
|
||||
|
||||
if not q.link(conv):
|
||||
log.error("Failed to link queue to videoconvert")
|
||||
return
|
||||
|
||||
if not conv.link(videoscale):
|
||||
log.error("Failed to link videoconvert to videoscale")
|
||||
return
|
||||
|
||||
if not videoscale.link(capsfilter):
|
||||
log.error("Failed to link videoscale to capsfilter")
|
||||
return
|
||||
|
||||
if not self.link_element_or_pad(capsfilter.link, remote_video_sink):
|
||||
log.error("Failed to link capsfilter to sink")
|
||||
return
|
||||
|
||||
else:
|
||||
self.pipeline.add(q, conv)
|
||||
|
|
@ -1010,16 +1075,29 @@ class WebRTC:
|
|||
if not self.pwd:
|
||||
sdp = self.webrtcbin.props.local_description.sdp.as_text()
|
||||
self.extract_ufrag_pwd(sdp)
|
||||
|
||||
ice_data = {}
|
||||
for media_type, candidates in self.local_candidates_buffer.items():
|
||||
if media_type not in self.ufrag or media_type not in self.pwd:
|
||||
log.error(f"Missing ICE credentials for {media_type}")
|
||||
log.debug(f"Available ufrag keys: {list(self.ufrag.keys())}")
|
||||
log.debug(f"Available pwd keys: {list(self.pwd.keys())}")
|
||||
continue
|
||||
|
||||
ice_data[media_type] = {
|
||||
"ufrag": self.ufrag[media_type],
|
||||
"pwd": self.pwd[media_type],
|
||||
"candidates": candidates,
|
||||
}
|
||||
await self.bridge.ice_candidates_add(
|
||||
self.sid, data_format.serialise(ice_data), self.profile
|
||||
)
|
||||
|
||||
if ice_data:
|
||||
log.debug(f"Sending ICE data for media types: {list(ice_data.keys())}")
|
||||
await self.bridge.ice_candidates_add(
|
||||
self.sid, data_format.serialise(ice_data), self.profile
|
||||
)
|
||||
else:
|
||||
log.error("No ICE data to send - missing credentials!")
|
||||
|
||||
self.local_candidates_buffer.clear()
|
||||
|
||||
def _remote_sdp_set(self, promise) -> None:
|
||||
|
|
@ -1079,6 +1157,11 @@ class WebRTC:
|
|||
@param mlineindex: The mline index.
|
||||
@param candidate: The ICE candidate.
|
||||
"""
|
||||
# Empty candidate signals end of candidates for this mline
|
||||
if not candidate_sdp or candidate_sdp.strip() == "":
|
||||
log.debug(f"ICE gathering complete for mline {mline_index}")
|
||||
return
|
||||
|
||||
log.debug(
|
||||
f"Local ICE candidate. MLine Index: {mline_index}, Candidate: {candidate_sdp}"
|
||||
)
|
||||
|
|
@ -1086,16 +1169,20 @@ class WebRTC:
|
|||
if parsed_candidate is None:
|
||||
log.warning(f"Can't parse candidate: {candidate_sdp}")
|
||||
return
|
||||
|
||||
try:
|
||||
media_type = self.media_types[mline_index]
|
||||
except KeyError:
|
||||
raise exceptions.InternalError("can't find media type")
|
||||
log.error(f"Can't find media type for mline {mline_index}")
|
||||
log.debug(f"{self.media_types=}")
|
||||
return
|
||||
|
||||
if self.sid is None:
|
||||
log.debug("buffering local ICE candidate")
|
||||
log.debug(f"buffering local ICE candidate for {media_type}")
|
||||
self.local_candidates_buffer.setdefault(media_type, []).append(
|
||||
parsed_candidate
|
||||
)
|
||||
|
||||
else:
|
||||
sdp = self.webrtcbin.props.local_description.sdp.as_text()
|
||||
assert sdp is not None
|
||||
|
|
@ -1119,6 +1206,8 @@ class WebRTC:
|
|||
keys and corresponding ICE data as values.
|
||||
|
||||
@raise exceptions.InternalError: Raised when sdp mline index is not found.
|
||||
|
||||
|
||||
"""
|
||||
if not self.sdp_set:
|
||||
log.debug("buffering remote ICE candidate")
|
||||
|
|
@ -1136,9 +1225,9 @@ class WebRTC:
|
|||
except Exception as e:
|
||||
raise exceptions.InternalError(f"Can't find sdp mline index: {e}")
|
||||
self.webrtcbin.emit("add-ice-candidate", mline_index, candidate_sdp)
|
||||
log.warning(
|
||||
log.debug(
|
||||
f"Remote ICE candidate added. MLine Index: {mline_index}, "
|
||||
f"{candidate_sdp}"
|
||||
f"Candidate: {candidate_sdp}"
|
||||
)
|
||||
|
||||
def on_ice_gathering_state_change(self, pspec, __):
|
||||
|
|
@ -1147,9 +1236,14 @@ class WebRTC:
|
|||
|
||||
def on_ice_connection_state(self, pspec, __):
|
||||
state = self.webrtcbin.props.ice_connection_state
|
||||
log.info(f"ICE connection state changed to: {state}")
|
||||
|
||||
if state == GstWebRTC.WebRTCICEConnectionState.FAILED:
|
||||
log.error("ICE connection failed")
|
||||
log.info(f"ICE connection state changed to {state}")
|
||||
log.error("ICE connection failed - check firewall/STUN/TURN configuration")
|
||||
elif state == GstWebRTC.WebRTCICEConnectionState.CONNECTED:
|
||||
log.info("ICE connection established successfully")
|
||||
elif state == GstWebRTC.WebRTCICEConnectionState.COMPLETED:
|
||||
log.info("ICE connectivity checks completed")
|
||||
|
||||
def on_bus_error(self, bus: Gst.Bus, message: Gst.Message) -> None:
|
||||
"""Handles the GStreamer bus error messages.
|
||||
|
|
@ -1238,7 +1332,7 @@ class WebRTC:
|
|||
f"Invalid source: {source!r}, use one of {'video', 'muted', 'desktop'}"
|
||||
)
|
||||
|
||||
self.pipeline.set_state(Gst.State.PAUSED)
|
||||
self.set_pipeline_state(Gst.State.PAUSED)
|
||||
|
||||
# Create a new desktop source if necessary
|
||||
if source == "desktop":
|
||||
|
|
@ -1271,7 +1365,7 @@ class WebRTC:
|
|||
if pad is not None:
|
||||
self.video_selector.props.active_pad = pad
|
||||
|
||||
self.pipeline.set_state(Gst.State.PLAYING)
|
||||
self.set_pipeline_state(Gst.State.PLAYING)
|
||||
|
||||
def _setup_desktop_source(self, properties: dict[str, object] | None) -> None:
|
||||
"""Set up a new desktop source.
|
||||
|
|
|
|||
Loading…
Reference in a new issue