1/*
2 * Copyright (C) 2017-2022 Igalia S.L. All rights reserved.
3 * Copyright (C) 2022 Metrological Group B.V.
4 *
5 * This library is free software; you can redistribute it and/or
6 * modify it under the terms of the GNU Lesser General Public
7 * License as published by the Free Software Foundation; either
8 * version 2 of the License, or (at your option) any later version.
9 *
10 * This library is distributed in the hope that it will be useful,
11 * but WITHOUT ANY WARRANTY; without even the implied warranty of
12 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
13 * Lesser General Public License for more details.
14 *
15 * You should have received a copy of the GNU Lesser General Public
16 * License along with this library; if not, write to the Free Software
17 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
18 */
19
20#include "config.h"
21#include "GStreamerMediaEndpoint.h"
22
23#if USE(GSTREAMER_WEBRTC)
24
25#include "Document.h"
26#include "GStreamerCommon.h"
27#include "GStreamerDataChannelHandler.h"
28#include "GStreamerRtpReceiverBackend.h"
29#include "GStreamerRtpTransceiverBackend.h"
30#include "GStreamerSctpTransportBackend.h"
31#include "GStreamerWebRTCUtils.h"
32#include "JSDOMPromiseDeferred.h"
33#include "JSRTCStatsReport.h"
34#include "Logging.h"
35#include "MediaEndpointConfiguration.h"
36#include "NotImplemented.h"
37#include "RTCDataChannel.h"
38#include "RTCDataChannelEvent.h"
39#include "RTCIceCandidate.h"
40#include "RTCOfferOptions.h"
41#include "RTCPeerConnection.h"
42#include "RTCSctpTransportBackend.h"
43#include "RTCSessionDescription.h"
44#include "RTCStatsReport.h"
45#include "RTCTrackEvent.h"
46#include "RealtimeIncomingAudioSourceGStreamer.h"
47#include "RealtimeIncomingVideoSourceGStreamer.h"
48#include "RealtimeOutgoingAudioSourceGStreamer.h"
49#include "RealtimeOutgoingVideoSourceGStreamer.h"
50
51#include <gst/sdp/sdp.h>
52#include <wtf/MainThread.h>
53#include <wtf/Scope.h>
54#include <wtf/glib/RunLoopSourcePriority.h>
55#include <wtf/text/StringToIntegerConversion.h>
56
57GST_DEBUG_CATEGORY(webkit_webrtc_endpoint_debug);
58#define GST_CAT_DEFAULT webkit_webrtc_endpoint_debug
59
60namespace WebCore {
61
62GStreamerMediaEndpoint::GStreamerMediaEndpoint(GStreamerPeerConnectionBackend& peerConnection)
63 : m_peerConnectionBackend(peerConnection)
64 , m_statsCollector(GStreamerStatsCollector::create())
65#if !RELEASE_LOG_DISABLED
66 , m_statsLogTimer(*this, &GStreamerMediaEndpoint::gatherStatsForLogging)
67 , m_logger(peerConnection.logger())
68 , m_logIdentifier(peerConnection.logIdentifier())
69#endif
70{
71 ensureGStreamerInitialized();
72
73 static std::once_flag debugRegisteredFlag;
74 std::call_once(debugRegisteredFlag, [] {
75 GST_DEBUG_CATEGORY_INIT(webkit_webrtc_endpoint_debug, "webkitwebrtcendpoint", 0, "WebKit WebRTC end-point");
76 });
77}
78
79bool GStreamerMediaEndpoint::initializePipeline()
80{
81 static uint32_t nPipeline = 0;
82 auto pipelineName = makeString("webkit-webrt-pipeline-", nPipeline);
83 m_pipeline = gst_pipeline_new(pipelineName.ascii().data());
84
85 connectSimpleBusMessageCallback(m_pipeline.get(), [this](GstMessage* message) {
86 handleMessage(message);
87 });
88
89 auto binName = makeString("webkit-webrtcbin-", nPipeline++);
90 m_webrtcBin = makeGStreamerElement("webrtcbin", binName.ascii().data());
91 if (!m_webrtcBin)
92 return false;
93
94 m_statsCollector->setElement(m_webrtcBin.get());
95 g_signal_connect_swapped(m_webrtcBin.get(), "notify::ice-connection-state", G_CALLBACK(+[](GStreamerMediaEndpoint* endPoint) {
96 endPoint->onIceConnectionChange();
97 }), this);
98 g_signal_connect_swapped(m_webrtcBin.get(), "notify::ice-gathering-state", G_CALLBACK(+[](GStreamerMediaEndpoint* endPoint) {
99 endPoint->onIceGatheringChange();
100 }), this);
101 g_signal_connect_swapped(m_webrtcBin.get(), "on-negotiation-needed", G_CALLBACK(+[](GStreamerMediaEndpoint* endPoint) {
102 endPoint->onNegotiationNeeded();
103 }), this);
104 g_signal_connect_swapped(m_webrtcBin.get(), "on-ice-candidate", G_CALLBACK(+[](GStreamerMediaEndpoint* endPoint, guint sdpMLineIndex, gchararray candidate) {
105 endPoint->onIceCandidate(sdpMLineIndex, candidate);
106 }), this);
107 g_signal_connect_swapped(m_webrtcBin.get(), "pad-added", G_CALLBACK(+[](GStreamerMediaEndpoint* endPoint, GstPad* pad) {
108 // Ignore outgoing pad notifications.
109 if (GST_PAD_DIRECTION(pad) != GST_PAD_SRC)
110 return;
111
112 callOnMainThreadAndWait([protectedThis = Ref(*endPoint), pad] {
113 if (protectedThis->isStopped())
114 return;
115
116 protectedThis->addRemoteStream(pad);
117 });
118 }), this);
119 g_signal_connect_swapped(m_webrtcBin.get(), "pad-removed", G_CALLBACK(+[](GStreamerMediaEndpoint* endPoint, GstPad* pad) {
120 // Ignore outgoing pad notifications.
121 if (GST_PAD_DIRECTION(pad) != GST_PAD_SRC)
122 return;
123
124 callOnMainThreadAndWait([protectedThis = Ref(*endPoint), pad] {
125 if (protectedThis->isStopped())
126 return;
127
128 protectedThis->removeRemoteStream(pad);
129 });
130 }), this);
131
132 g_signal_connect_swapped(m_webrtcBin.get(), "on-data-channel", G_CALLBACK(+[](GStreamerMediaEndpoint* endPoint, GstWebRTCDataChannel* channel) {
133 endPoint->onDataChannel(channel);
134 }), this);
135
136 gst_bin_add(GST_BIN_CAST(m_pipeline.get()), m_webrtcBin.get());
137 return true;
138}
139
140GStreamerMediaEndpoint::~GStreamerMediaEndpoint()
141{
142 if (m_pipeline)
143 teardownPipeline();
144}
145
146void GStreamerMediaEndpoint::teardownPipeline()
147{
148 RELEASE_ASSERT(m_pipeline);
149#if !RELEASE_LOG_DISABLED
150 stopLoggingStats();
151#endif
152 m_statsCollector->setElement(nullptr);
153
154 g_signal_handlers_disconnect_by_data(m_webrtcBin.get(), this);
155 disconnectSimpleBusMessageCallback(m_pipeline.get());
156 gst_element_set_state(m_pipeline.get(), GST_STATE_NULL);
157
158 m_sources.clear();
159 m_remoteMLineInfos.clear();
160 m_remoteStreamsById.clear();
161 m_webrtcBin = nullptr;
162 m_pipeline = nullptr;
163}
164
165bool GStreamerMediaEndpoint::handleMessage(GstMessage* message)
166{
167 switch (GST_MESSAGE_TYPE(message)) {
168 case GST_MESSAGE_EOS:
169 GST_DEBUG_BIN_TO_DOT_FILE_WITH_TS(GST_BIN_CAST(m_pipeline.get()), GST_DEBUG_GRAPH_SHOW_ALL, "eos");
170 break;
171 case GST_MESSAGE_ELEMENT: {
172 const auto* data = gst_message_get_structure(message);
173 if (!g_strcmp0(gst_structure_get_name(data), "GstBinForwarded")) {
174 GRefPtr<GstMessage> subMessage;
175 gst_structure_get(data, "message", GST_TYPE_MESSAGE, &subMessage.outPtr(), nullptr);
176 if (GST_MESSAGE_TYPE(subMessage.get()) == GST_MESSAGE_EOS)
177 disposeElementChain(GST_ELEMENT_CAST(GST_MESSAGE_SRC(subMessage.get())));
178 }
179 break;
180 }
181 default:
182 break;
183 }
184 return true;
185}
186
187void GStreamerMediaEndpoint::disposeElementChain(GstElement* element)
188{
189 GST_DEBUG_OBJECT(m_pipeline.get(), "Got element EOS message from %" GST_PTR_FORMAT, element);
190
191 auto pad = adoptGRef(gst_element_get_static_pad(element, "sink"));
192 auto peer = adoptGRef(gst_pad_get_peer(pad.get()));
193
194 gst_element_set_locked_state(m_pipeline.get(), TRUE);
195 gst_pad_unlink(peer.get(), pad.get());
196 gst_bin_remove(GST_BIN_CAST(m_pipeline.get()), element);
197 gst_element_release_request_pad(m_webrtcBin.get(), peer.get());
198 gst_element_set_state(element, GST_STATE_NULL);
199 gst_element_set_locked_state(m_pipeline.get(), FALSE);
200}
201
202bool GStreamerMediaEndpoint::setConfiguration(MediaEndpointConfiguration& configuration)
203{
204 if (m_pipeline)
205 teardownPipeline();
206
207 if (!initializePipeline())
208 return false;
209
210 auto bundlePolicy = bundlePolicyFromConfiguration(configuration);
211 auto iceTransportPolicy = iceTransportPolicyFromConfiguration(configuration);
212 g_object_set(m_webrtcBin.get(), "bundle-policy", bundlePolicy, "ice-transport-policy", iceTransportPolicy, nullptr);
213
214 for (auto& server : configuration.iceServers) {
215 bool stunSet = false;
216 for (auto& url : server.urls) {
217 if (url.protocol().startsWith("turn")) {
218 auto valid = url.string().isolatedCopy().replace("turn:", "turn://").replace("turns:", "turns://");
219 URL validURL(URL(), valid);
220 // FIXME: libnice currently doesn't seem to handle IPv6 addresses very well.
221 if (validURL.host().startsWith('['))
222 continue;
223 validURL.setUser(server.username);
224 validURL.setPassword(server.credential);
225 bool result = false;
226 g_signal_emit_by_name(m_webrtcBin.get(), "add-turn-server", validURL.string().utf8().data(), &result);
227 if (!result)
228 GST_WARNING("Unable to use TURN server: %s", validURL.string().utf8().data());
229 }
230 if (!stunSet && url.protocol().startsWith("stun")) {
231 auto valid = url.string().isolatedCopy().replace("stun:", "stun://");
232 URL validURL(URL(), valid);
233 // FIXME: libnice currently doesn't seem to handle IPv6 addresses very well.
234 if (validURL.host().startsWith('['))
235 continue;
236 validURL.setUser(server.username);
237 validURL.setPassword(server.credential);
238 g_object_set(m_webrtcBin.get(), "stun-server", validURL.string().utf8().data(), nullptr);
239 stunSet = true;
240 }
241 }
242 }
243
244 // WIP: https://gitlab.freedesktop.org/gstreamer/gst-plugins-bad/-/merge_requests/302
245 GST_FIXME("%zu custom certificates not propagated to webrtcbin", configuration.certificates.size());
246
247 gst_element_set_state(m_pipeline.get(), GST_STATE_PLAYING);
248 GST_DEBUG_OBJECT(m_pipeline.get(), "End-point ready");
249 return true;
250}
251
252void GStreamerMediaEndpoint::restartIce()
253{
254 GST_DEBUG_OBJECT(m_pipeline.get(), "restarting ICE");
255 // WIP in https://gitlab.freedesktop.org/gstreamer/gst-plugins-bad/-/merge_requests/1877
256 initiate(true, gst_structure_new("webrtcbin-offer-options", "ice-restart", G_TYPE_BOOLEAN, TRUE, nullptr));
257}
258
259static std::optional<std::pair<RTCSdpType, String>> fetchDescription(GstElement* webrtcBin, const char* name)
260{
261 if (!webrtcBin)
262 return { };
263
264 GUniqueOutPtr<GstWebRTCSessionDescription> description;
265 g_object_get(webrtcBin, makeString(name, "-description").utf8().data(), &description.outPtr(), nullptr);
266 if (!description)
267 return { };
268
269 GUniquePtr<char> sdpString(gst_sdp_message_as_text(description->sdp));
270 GST_TRACE_OBJECT(webrtcBin, "%s-description SDP: %s", name, sdpString.get());
271 return { { fromSessionDescriptionType(*description.get()), sdpString.get() } };
272}
273
274static GstWebRTCSignalingState fetchSignalingState(GstElement* webrtcBin)
275{
276 GstWebRTCSignalingState state;
277 g_object_get(webrtcBin, "signaling-state", &state, nullptr);
278#ifndef GST_DISABLE_GST_DEBUG
279 GUniquePtr<char> desc(g_enum_to_string(GST_TYPE_WEBRTC_SIGNALING_STATE, state));
280 GST_DEBUG_OBJECT(webrtcBin, "Signaling state set to %s", desc.get());
281#endif
282 return state;
283}
284
285enum class GatherSignalingState { No, Yes };
286static std::optional<PeerConnectionBackend::DescriptionStates> descriptionsFromWebRTCBin(GstElement* webrtcBin, GatherSignalingState gatherSignalingState = GatherSignalingState::No)
287{
288 std::optional<RTCSdpType> currentLocalDescriptionSdpType, pendingLocalDescriptionSdpType, currentRemoteDescriptionSdpType, pendingRemoteDescriptionSdpType;
289 String currentLocalDescriptionSdp, pendingLocalDescriptionSdp, currentRemoteDescriptionSdp, pendingRemoteDescriptionSdp;
290 if (auto currentLocalDescription = fetchDescription(webrtcBin, "current-local")) {
291 auto [sdpType, description] = *currentLocalDescription;
292 currentLocalDescriptionSdpType = sdpType;
293 currentLocalDescriptionSdp = WTFMove(description);
294 }
295 if (auto pendingLocalDescription = fetchDescription(webrtcBin, "pending-local")) {
296 auto [sdpType, description] = *pendingLocalDescription;
297 pendingLocalDescriptionSdpType = sdpType;
298 pendingLocalDescriptionSdp = WTFMove(description);
299 }
300 if (auto currentRemoteDescription = fetchDescription(webrtcBin, "current-remote")) {
301 auto [sdpType, description] = *currentRemoteDescription;
302 currentRemoteDescriptionSdpType = sdpType;
303 currentRemoteDescriptionSdp = WTFMove(description);
304 }
305 if (auto pendingRemoteDescription = fetchDescription(webrtcBin, "pending-remote")) {
306 auto [sdpType, description] = *pendingRemoteDescription;
307 pendingRemoteDescriptionSdpType = sdpType;
308 pendingRemoteDescriptionSdp = WTFMove(description);
309 }
310
311 std::optional<RTCSignalingState> signalingState;
312 if (gatherSignalingState == GatherSignalingState::Yes)
313 signalingState = toSignalingState(fetchSignalingState(webrtcBin));
314
315 return PeerConnectionBackend::DescriptionStates {
316 signalingState,
317 currentLocalDescriptionSdpType, currentLocalDescriptionSdp,
318 pendingLocalDescriptionSdpType, pendingLocalDescriptionSdp,
319 currentRemoteDescriptionSdpType, currentRemoteDescriptionSdp,
320 pendingRemoteDescriptionSdpType, pendingRemoteDescriptionSdp
321 };
322}
323
324void GStreamerMediaEndpoint::doSetLocalDescription(const RTCSessionDescription* description)
325{
326 setDescription(description, true, [protectedThis = Ref(*this)] {
327 auto descriptions = descriptionsFromWebRTCBin(protectedThis->m_webrtcBin.get(), GatherSignalingState::Yes);
328 GRefPtr<GstWebRTCSCTPTransport> transport;
329 g_object_get(protectedThis->m_webrtcBin.get(), "sctp-transport", &transport.outPtr(), nullptr);
330 protectedThis->m_peerConnectionBackend.setLocalDescriptionSucceeded(WTFMove(descriptions), transport ? makeUnique<GStreamerSctpTransportBackend>(WTFMove(transport)) : nullptr);
331 }, [protectedThis = Ref(*this)](const GError* error) {
332 if (error && error->code == GST_WEBRTC_ERROR_INVALID_STATE)
333 protectedThis->m_peerConnectionBackend.setLocalDescriptionFailed(Exception { InvalidStateError, "Failed to set local answer sdp: no pending remote description."_s });
334 else
335 protectedThis->m_peerConnectionBackend.setLocalDescriptionFailed(Exception { OperationError, "Unable to apply session local description"_s });
336 });
337}
338
339void GStreamerMediaEndpoint::setRemoteDescriptionSucceeded()
340{
341 GST_DEBUG_OBJECT(m_pipeline.get(), "Acking remote description");
342 auto descriptions = descriptionsFromWebRTCBin(m_webrtcBin.get(), GatherSignalingState::Yes);
343 GRefPtr<GstWebRTCSCTPTransport> transport;
344 g_object_get(m_webrtcBin.get(), "sctp-transport", &transport.outPtr(), nullptr);
345 m_peerConnectionBackend.setRemoteDescriptionSucceeded(WTFMove(descriptions), transport ? makeUnique<GStreamerSctpTransportBackend>(WTFMove(transport)) : nullptr);
346}
347
348void GStreamerMediaEndpoint::doSetRemoteDescription(const RTCSessionDescription& description)
349{
350 setDescription(&description, false, [protectedThis = Ref(*this)] {
351 protectedThis->setRemoteDescriptionSucceeded();
352 }, [protectedThis = Ref(*this)](const GError* error) {
353 if (error && error->code == GST_WEBRTC_ERROR_INVALID_STATE)
354 protectedThis->m_peerConnectionBackend.setRemoteDescriptionFailed(Exception { InvalidStateError, "Failed to set remote answer sdp"_s });
355 else
356 protectedThis->m_peerConnectionBackend.setRemoteDescriptionFailed(Exception { OperationError, "Unable to apply session remote description"_s });
357 });
358#if !RELEASE_LOG_DISABLED
359 startLoggingStats();
360#endif
361}
362
363struct SetDescriptionCallData {
364 WTF_MAKE_STRUCT_FAST_ALLOCATED;
365 SetDescriptionCallData(GStreamerMediaEndpoint& endPoint, Function<void()>&& successCallback, Function<void(const GError*)>&& failureCallback)
366 : successCallback(WTFMove(successCallback))
367 , failureCallback(WTFMove(failureCallback))
368 , endPoint(endPoint) { }
369
370 Function<void()> successCallback;
371 Function<void(const GError*)> failureCallback;
372 GStreamerMediaEndpoint& endPoint;
373};
374
375void GStreamerMediaEndpoint::setDescription(const RTCSessionDescription* description, bool isLocal, Function<void()>&& successCallback, Function<void(const GError*)>&& failureCallback)
376{
377 GstSDPMessage* message;
378 auto sdpType = RTCSdpType::Offer;
379
380 if (description) {
381 if (gst_sdp_message_new_from_text(reinterpret_cast<const char*>(description->sdp().characters8()), &message) != GST_SDP_OK) {
382 failureCallback(nullptr);
383 return;
384 }
385 sdpType = description->type();
386 } else if (gst_sdp_message_new(&message) != GST_SDP_OK) {
387 failureCallback(nullptr);
388 return;
389 }
390
391 if (!isLocal)
392 storeRemoteMLineInfo(message);
393
394 auto type = toSessionDescriptionType(sdpType);
395 GST_DEBUG_OBJECT(m_pipeline.get(), "Creating %s session for SDP %s", isLocal ? "local" : "remote", gst_webrtc_sdp_type_to_string(type));
396
397 GUniquePtr<GstWebRTCSessionDescription> sessionDescription(gst_webrtc_session_description_new(type, message));
398 auto signalName = makeString("set-", isLocal ? "local" : "remote", "-description");
399 auto data = makeUnique<SetDescriptionCallData>(*this, WTFMove(successCallback), WTFMove(failureCallback));
400 g_signal_emit_by_name(m_webrtcBin.get(), signalName.ascii().data(), sessionDescription.get(), gst_promise_new_with_change_func([](GstPromise* rawPromise, gpointer userData) {
401 std::unique_ptr<SetDescriptionCallData> data(reinterpret_cast<SetDescriptionCallData*>(userData));
402 auto promise = adoptGRef(rawPromise);
403 auto result = gst_promise_wait(promise.get());
404 const auto* reply = gst_promise_get_reply(promise.get());
405 if (result != GST_PROMISE_RESULT_REPLIED || (reply && gst_structure_has_field(reply, "error"))) {
406 GUniqueOutPtr<GError> error;
407 if (reply) {
408 gst_structure_get(reply, "error", G_TYPE_ERROR, &error.outPtr(), nullptr);
409 GST_ERROR_OBJECT(data->endPoint.pipeline(), "Unable to set description, error: %s", error->message);
410 }
411 callOnMainThread([error = error.get(), protectedThis = Ref(data->endPoint), failureCallback = WTFMove(data->failureCallback)] {
412 if (protectedThis->isStopped())
413 return;
414 failureCallback(error);
415 });
416 return;
417 }
418 callOnMainThread([protectedThis = Ref(data->endPoint), successCallback = WTFMove(data->successCallback)] {
419 if (protectedThis->isStopped())
420 return;
421 successCallback();
422 });
423 }, data.release(), nullptr));
424}
425
426void GStreamerMediaEndpoint::storeRemoteMLineInfo(GstSDPMessage* message)
427{
428 m_remoteMLineInfos.clear();
429 unsigned totalMedias = gst_sdp_message_medias_len(message);
430 m_remoteMLineInfos.reserveCapacity(totalMedias);
431 GST_DEBUG_OBJECT(m_pipeline.get(), "Storing %u remote pending mlines", totalMedias);
432 for (unsigned i = 0; i < totalMedias; i++) {
433 const GstSDPMedia* media = gst_sdp_message_get_media(message, i);
434 const char* typ = gst_sdp_media_get_media(media);
435 if (!g_str_equal(typ, "audio") && !g_str_equal(typ, "video"))
436 continue;
437
438#ifndef GST_DISABLE_GST_DEBUG
439 GUniquePtr<char> mediaRepresentation(gst_sdp_media_as_text(media));
440 GST_LOG_OBJECT(m_pipeline.get(), "Processing media:\n%s", mediaRepresentation.get());
441#endif
442 const char* mid = gst_sdp_media_get_attribute_val(media, "mid");
443 if (!mid)
444 continue;
445
446 bool isInactive = false;
447 for (unsigned ii = 0; ii < gst_sdp_media_attributes_len(media); ii++) {
448 const GstSDPAttribute* attr = gst_sdp_media_get_attribute(media, ii);
449 if (!g_strcmp0(attr->key, "inactive")) {
450 isInactive = true;
451 break;
452 }
453 }
454 if (isInactive) {
455 GST_DEBUG_OBJECT(m_pipeline.get(), "Skipping inactive media");
456 continue;
457 }
458
459 m_mediaForMid.set(String(mid), g_str_equal(typ, "audio") ? RealtimeMediaSource::Type::Audio : RealtimeMediaSource::Type::Video);
460
461 // https://gitlab.freedesktop.org/gstreamer/gst-plugins-bad/-/merge_requests/1907
462 if (sdpMediaHasAttributeKey(media, "ice-lite")) {
463 GRefPtr<GObject> ice;
464 g_object_get(m_webrtcBin.get(), "ice-agent", &ice.outPtr(), nullptr);
465 g_object_set(ice.get(), "ice-lite", TRUE, nullptr);
466 }
467
468 auto caps = adoptGRef(gst_caps_new_empty());
469 Vector<int> payloadTypes;
470 unsigned totalFormats = gst_sdp_media_formats_len(media);
471 GST_DEBUG_OBJECT(m_pipeline.get(), "Media %s has %u formats", typ, totalFormats);
472 for (unsigned j = 0; j < totalFormats; j++) {
473 auto format = String(gst_sdp_media_get_format(media, j));
474 auto payloadType = parseInteger<int>(format);
475 if (!payloadType || !*payloadType) {
476 GST_WARNING_OBJECT(m_pipeline.get(), "Invalid payload type: %s", format.utf8().data());
477 continue;
478 }
479 auto formatCaps = adoptGRef(gst_sdp_media_get_caps_from_media(media, *payloadType));
480 if (!formatCaps) {
481 GST_WARNING_OBJECT(m_pipeline.get(), "No caps found for payload type %d", *payloadType);
482 continue;
483 }
484
485 // Relay SDP attributes to the caps, this is specially useful so that elements in
486 // webrtcbin will be able to enable RTP header extensions.
487 gst_sdp_media_attributes_to_caps(media, formatCaps.get());
488
489 gst_caps_append(caps.get(), formatCaps.leakRef());
490 m_ptCounter = std::max(m_ptCounter, *payloadType + 1);
491 payloadTypes.append(*payloadType);
492 }
493
494 unsigned totalCaps = gst_caps_get_size(caps.get());
495 if (totalCaps) {
496 for (unsigned j = 0; j < totalCaps; j++) {
497 GstStructure* structure = gst_caps_get_structure(caps.get(), j);
498 gst_structure_set_name(structure, "application/x-rtp");
499 }
500 GST_DEBUG_OBJECT(m_pipeline.get(), "Caching payload caps: %" GST_PTR_FORMAT, caps.get());
501 m_remoteMLineInfos.uncheckedAppend({ WTFMove(caps), false, WTFMove(payloadTypes) });
502 }
503 }
504}
505
506void GStreamerMediaEndpoint::configureAndLinkSource(RealtimeOutgoingMediaSourceGStreamer& source)
507{
508 auto caps = source.allowedCaps();
509 std::optional<std::pair<PendingMLineInfo, GRefPtr<GstCaps>>> found;
510 for (auto& mLineInfo : m_remoteMLineInfos) {
511 if (mLineInfo.isUsed)
512 continue;
513
514 for (unsigned i = 0; i < gst_caps_get_size(mLineInfo.caps.get()); i++) {
515 auto* structure = gst_caps_get_structure(mLineInfo.caps.get(), i);
516 auto caps2 = adoptGRef(gst_caps_new_full(gst_structure_copy(structure), nullptr));
517 auto intersected = adoptGRef(gst_caps_intersect(caps.get(), caps2.get()));
518 if (!gst_caps_is_empty(intersected.get())) {
519 found = std::make_pair(mLineInfo, WTFMove(intersected));
520 break;
521 }
522 }
523 if (found.has_value())
524 break;
525 }
526
527 bool payloadTypeWasSet = false;
528 if (found) {
529 GST_DEBUG_OBJECT(m_pipeline.get(), "Unused and compatible caps found for %" GST_PTR_FORMAT "... %" GST_PTR_FORMAT, caps.get(), found->second.get());
530 payloadTypeWasSet = source.setPayloadType(found->second);
531 found->first.isUsed = true;
532 } else
533 payloadTypeWasSet = source.setPayloadType(caps);
534
535 if (!payloadTypeWasSet)
536 return;
537
538 if (!source.pad()) {
539 source.setSinkPad(requestPad(m_requestPadCounter, source.allowedCaps()));
540 m_requestPadCounter++;
541 }
542
543 auto& sinkPad = source.pad();
544 RELEASE_ASSERT(!gst_pad_is_linked(sinkPad.get()));
545 auto sourceBin = source.bin();
546 gst_bin_add(GST_BIN_CAST(m_pipeline.get()), sourceBin.get());
547 source.link();
548
549 GUniquePtr<char> padId(gst_pad_get_name(sinkPad.get()));
550 auto dotFileName = makeString(GST_OBJECT_NAME(m_pipeline.get()), ".outgoing-", padId.get());
551 GST_DEBUG_BIN_TO_DOT_FILE_WITH_TS(GST_BIN(m_pipeline.get()), GST_DEBUG_GRAPH_SHOW_ALL, dotFileName.utf8().data());
552}
553
554GRefPtr<GstPad> GStreamerMediaEndpoint::requestPad(unsigned mlineIndex, const GRefPtr<GstCaps>& allowedCaps)
555{
556 auto padId = makeString("sink_", mlineIndex);
557 auto caps = adoptGRef(gst_caps_copy(allowedCaps.get()));
558
559 gst_caps_map_in_place(caps.get(), [](GstCapsFeatures*, GstStructure* structure, gpointer userData) -> gboolean {
560 auto* endPoint = reinterpret_cast<GStreamerMediaEndpoint*>(userData);
561 gst_structure_set(structure, "payload", G_TYPE_INT, endPoint->m_ptCounter++, nullptr);
562 return TRUE;
563 }, this);
564
565 // FIXME: Restricting caps and codec-preferences breaks caps negotiation of outgoing sources in
566 // some case (jitsi).
567 GST_DEBUG_OBJECT(m_pipeline.get(), "Requesting pad %s restricted to %" GST_PTR_FORMAT, padId.utf8().data(), caps.get());
568 auto* padTemplate = gst_element_get_pad_template(m_webrtcBin.get(), "sink_%u");
569 auto sinkPad = adoptGRef(gst_element_request_pad(m_webrtcBin.get(), padTemplate, padId.utf8().data(), caps.get()));
570 GRefPtr<GstWebRTCRTPTransceiver> transceiver;
571 g_object_get(sinkPad.get(), "transceiver", &transceiver.outPtr(), nullptr);
572 g_object_set(transceiver.get(), "codec-preferences", caps.get(), nullptr);
573 return sinkPad;
574}
575
576bool GStreamerMediaEndpoint::addTrack(GStreamerRtpSenderBackend& sender, MediaStreamTrack& track, const FixedVector<String>&)
577{
578 GStreamerRtpSenderBackend::Source source;
579 GRefPtr<GstWebRTCRTPSender> rtcSender;
580
581 if (track.privateTrack().hasAudio()) {
582 GST_DEBUG_OBJECT(m_pipeline.get(), "Adding outgoing audio source");
583 auto audioSource = RealtimeOutgoingAudioSourceGStreamer::create(track.privateTrack());
584 configureAndLinkSource(audioSource);
585
586 rtcSender = audioSource->sender();
587 source = WTFMove(audioSource);
588 } else {
589 RELEASE_ASSERT(track.privateTrack().hasVideo());
590 GST_DEBUG_OBJECT(m_pipeline.get(), "Adding outgoing video source");
591 auto videoSource = RealtimeOutgoingVideoSourceGStreamer::create(track.privateTrack());
592 configureAndLinkSource(videoSource);
593
594 rtcSender = videoSource->sender();
595 source = WTFMove(videoSource);
596 }
597
598 sender.setSource(WTFMove(source));
599
600 if (auto rtpSender = sender.rtcSender()) {
601 GST_DEBUG_OBJECT(m_pipeline.get(), "Already has a sender.");
602 return true;
603 }
604
605 sender.setRTCSender(WTFMove(rtcSender));
606
607 GST_DEBUG_OBJECT(m_pipeline.get(), "Sender configured");
608 return true;
609}
610
611void GStreamerMediaEndpoint::removeTrack(GStreamerRtpSenderBackend& sender)
612{
613 GST_DEBUG_OBJECT(m_pipeline.get(), "Removing track");
614 if (auto bin = sender.stopSource())
615 gst_bin_remove(GST_BIN_CAST(m_pipeline.get()), bin.get());
616 sender.clearSource();
617 onNegotiationNeeded();
618}
619
620void GStreamerMediaEndpoint::doCreateOffer(const RTCOfferOptions& options)
621{
622 // https://gitlab.freedesktop.org/gstreamer/gst-plugins-bad/-/merge_requests/1877
623 // FIXME: Plumb options.voiceActivityDetection.
624 initiate(true, gst_structure_new("webrtcbin-offer-options", "ice-restart", G_TYPE_BOOLEAN, options.iceRestart, nullptr));
625}
626
627void GStreamerMediaEndpoint::doCreateAnswer()
628{
629 initiate(false, nullptr);
630}
631
632void GStreamerMediaEndpoint::initiate(bool isInitiator, GstStructure* rawOptions)
633{
634 m_isInitiator = isInitiator;
635 const char* type = isInitiator ? "offer" : "answer";
636 GST_DEBUG_OBJECT(m_pipeline.get(), "Creating %s", type);
637 auto signalName = makeString("create-", type);
638 GUniquePtr<GstStructure> options(rawOptions);
639 g_signal_emit_by_name(m_webrtcBin.get(), signalName.ascii().data(), options.get(), gst_promise_new_with_change_func([](GstPromise* rawPromise, gpointer userData) {
640 auto* endpoint = reinterpret_cast<GStreamerMediaEndpoint*>(userData);
641 auto promise = adoptGRef(rawPromise);
642 auto result = gst_promise_wait(promise.get());
643 if (result != GST_PROMISE_RESULT_REPLIED) {
644 endpoint->createSessionDescriptionFailed({ });
645 return;
646 }
647
648 const auto* reply = gst_promise_get_reply(promise.get());
649 RELEASE_ASSERT(reply);
650 if (gst_structure_has_field(reply, "error")) {
651 GUniqueOutPtr<GError> promiseError;
652 gst_structure_get(reply, "error", G_TYPE_ERROR, &promiseError.outPtr(), nullptr);
653 GUniquePtr<GError> error(promiseError.release());
654 endpoint->createSessionDescriptionFailed(WTFMove(error));
655 return;
656 }
657
658 const char* type = endpoint->m_isInitiator ? "offer" : "answer";
659 GUniqueOutPtr<GstWebRTCSessionDescription> sessionDescription;
660 gst_structure_get(reply, type, GST_TYPE_WEBRTC_SESSION_DESCRIPTION, &sessionDescription.outPtr(), nullptr);
661
662#ifndef GST_DISABLE_GST_DEBUG
663 GUniquePtr<char> sdp(gst_sdp_message_as_text(sessionDescription->sdp));
664 GST_DEBUG_OBJECT(endpoint->pipeline(), "Created %s: %s", type, sdp.get());
665#endif
666 GUniquePtr<GstWebRTCSessionDescription> description(sessionDescription.release());
667 endpoint->createSessionDescriptionSucceeded(WTFMove(description));
668 }, this, nullptr));
669}
670
671void GStreamerMediaEndpoint::getStats(GstPad* pad, Ref<DeferredPromise>&& promise)
672{
673 m_statsCollector->getStats([promise = WTFMove(promise), protectedThis = Ref(*this)](auto&& report) mutable {
674 ASSERT(isMainThread());
675 if (protectedThis->isStopped() || !report)
676 return;
677
678 promise->resolve<IDLInterface<RTCStatsReport>>(report.releaseNonNull());
679 }, pad);
680}
681
682MediaStream& GStreamerMediaEndpoint::mediaStreamFromRTCStream(String label)
683{
684 auto mediaStream = m_remoteStreamsById.ensure(label, [label, this]() mutable {
685 auto& document = downcast<Document>(*m_peerConnectionBackend.connection().scriptExecutionContext());
686 return MediaStream::create(document, MediaStreamPrivate::create(document.logger(), { }, WTFMove(label)));
687 });
688 return *mediaStream.iterator->value;
689}
690
691void GStreamerMediaEndpoint::addRemoteStream(GstPad* pad)
692{
693 m_pendingIncomingStreams++;
694
695 auto caps = adoptGRef(gst_pad_query_caps(pad, nullptr));
696 const char* mediaType = capsMediaType(caps.get());
697 GST_DEBUG_OBJECT(m_pipeline.get(), "Adding remote %s stream to %" GST_PTR_FORMAT " %u pending incoming streams, caps: %" GST_PTR_FORMAT, mediaType, pad, m_pendingIncomingStreams, caps.get());
698
699 GRefPtr<GstWebRTCRTPTransceiver> rtcTransceiver;
700 g_object_get(pad, "transceiver", &rtcTransceiver.outPtr(), nullptr);
701
702 auto* transceiver = m_peerConnectionBackend.existingTransceiver([&](auto& transceiverBackend) {
703 return rtcTransceiver.get() == transceiverBackend.rtcTransceiver();
704 });
705 if (!transceiver) {
706 auto type = doCapsHaveType(caps.get(), "audio") ? RealtimeMediaSource::Type::Audio : RealtimeMediaSource::Type::Video;
707 transceiver = &m_peerConnectionBackend.newRemoteTransceiver(makeUnique<GStreamerRtpTransceiverBackend>(WTFMove(rtcTransceiver)), type);
708 }
709
710 GUniqueOutPtr<GstWebRTCSessionDescription> description;
711 g_object_get(m_webrtcBin.get(), "remote-description", &description.outPtr(), nullptr);
712
713 unsigned mLineIndex;
714 g_object_get(rtcTransceiver.get(), "mlineindex", &mLineIndex, nullptr);
715 const auto* media = gst_sdp_message_get_media(description->sdp, mLineIndex);
716 GUniquePtr<char> sdp(gst_sdp_media_as_text(media));
717 auto sdpString = makeString(sdp.get());
718
719 GUniquePtr<gchar> name(gst_pad_get_name(pad));
720 String label(name.get());
721 auto key = makeString("msid:");
722 auto lines = sdpString.split('\n');
723 for (auto& line : lines) {
724 auto i = line.find(key);
725 if (i != notFound) {
726 auto tmp = line.substring(i + key.ascii().length());
727 label = tmp.substring(0, tmp.find(' '));
728 break;
729 }
730 }
731
732 GST_DEBUG_OBJECT(m_pipeline.get(), "msid: %s", label.ascii().data());
733
734 GstElement* bin = nullptr;
735 auto& track = transceiver->receiver().track();
736 auto& source = track.privateTrack().source();
737 if (source.isIncomingAudioSource())
738 bin = static_cast<RealtimeIncomingAudioSourceGStreamer&>(source).bin();
739 else if (source.isIncomingVideoSource())
740 bin = static_cast<RealtimeIncomingVideoSourceGStreamer&>(source).bin();
741 else
742 RELEASE_ASSERT_NOT_REACHED();
743
744 gst_bin_add(GST_BIN_CAST(m_pipeline.get()), bin);
745 auto sinkPad = adoptGRef(gst_element_get_static_pad(bin, "sink"));
746 gst_pad_link(pad, sinkPad.get());
747 gst_element_sync_state_with_parent(bin);
748
749 track.setEnabled(true);
750 source.setMuted(false);
751
752 auto& mediaStream = mediaStreamFromRTCStream(label);
753 mediaStream.addTrackFromPlatform(track);
754 m_pendingStreams.append(&mediaStream);
755
756 auto dotFileName = makeString(GST_OBJECT_NAME(m_pipeline.get()), ".incoming-", mediaType, '-', GST_OBJECT_NAME(pad));
757 GST_DEBUG_BIN_TO_DOT_FILE_WITH_TS(GST_BIN(m_pipeline.get()), GST_DEBUG_GRAPH_SHOW_ALL, dotFileName.utf8().data());
758}
759
760void GStreamerMediaEndpoint::removeRemoteStream(GstPad*)
761{
762 GST_FIXME_OBJECT(m_pipeline.get(), "removeRemoteStream");
763 notImplemented();
764}
765
766std::optional<GStreamerMediaEndpoint::Backends> GStreamerMediaEndpoint::createTransceiverBackends(const String& kind, const RTCRtpTransceiverInit& init, GStreamerRtpSenderBackend::Source&& source)
767{
768 if (!m_webrtcBin)
769 return std::nullopt;
770
771 GST_DEBUG_OBJECT(m_pipeline.get(), "%zu streams in init data", init.streams.size());
772
773 // FIXME: Should we build the caps from the init.sendEncodings? Problem is there is no
774 // encodingName or clockRate in RTCRtpEncodingParameters.
775 Vector<std::pair<const char*, int>> encodings;
776 const char* media = kind.utf8().data();
777 if (kind == "video"_s) {
778 encodings.reserveInitialCapacity(3);
779 encodings.uncheckedAppend({ "VP8", 90000 });
780 encodings.uncheckedAppend({ "VP9", 90000 });
781 encodings.uncheckedAppend({ "H264", 90000 });
782 } else {
783 encodings.reserveInitialCapacity(1);
784 encodings.uncheckedAppend({ "OPUS", 48000 });
785 }
786
787 auto caps = adoptGRef(gst_caps_new_empty());
788 for (auto& [encodingName, clockRate] : encodings) {
789 gst_caps_append(caps.get(), gst_caps_new_simple("application/x-rtp", "media", G_TYPE_STRING, media, "encoding-name", G_TYPE_STRING, encodingName,
790 "payload", G_TYPE_INT, m_ptCounter++, "clock-rate", G_TYPE_INT, clockRate, nullptr));
791 }
792
793 auto direction = fromRTCRtpTransceiverDirection(init.direction);
794#ifndef GST_DISABLE_GST_DEBUG
795 GUniquePtr<char> desc(g_enum_to_string(GST_TYPE_WEBRTC_RTP_TRANSCEIVER_DIRECTION, direction));
796 GST_DEBUG_OBJECT(m_pipeline.get(), "Adding %s transceiver for payload %" GST_PTR_FORMAT, desc.get(), caps.get());
797#endif
798
799 // FIXME: None of this (excepted direction) is passed to webrtcbin yet.
800 GUniquePtr<GstStructure> initData(gst_structure_new("transceiver-init-data", "direction", GST_TYPE_WEBRTC_RTP_TRANSCEIVER_DIRECTION, direction, nullptr));
801
802 GValue streamIdsValue = G_VALUE_INIT;
803 g_value_init(&streamIdsValue, GST_TYPE_LIST);
804 for (auto& stream : init.streams) {
805 GValue value = G_VALUE_INIT;
806 g_value_init(&value, G_TYPE_STRING);
807 g_value_set_string(&value, stream->id().utf8().data());
808 gst_value_list_append_value(&streamIdsValue, &value);
809 g_value_unset(&value);
810 }
811 gst_structure_take_value(initData.get(), "stream-ids", &streamIdsValue);
812
813 GValue encodingsValue = G_VALUE_INIT;
814 g_value_init(&encodingsValue, GST_TYPE_LIST);
815 if (kind == "audio") {
816 if (!init.sendEncodings.isEmpty()) {
817 auto encodingData = fromRTCEncodingParameters(init.sendEncodings[0]);
818 GValue value = G_VALUE_INIT;
819 g_value_init(&value, GST_TYPE_STRUCTURE);
820 gst_value_set_structure(&value, encodingData.get());
821 gst_value_list_append_value(&encodingsValue, &value);
822 g_value_unset(&value);
823 }
824 } else {
825 for (auto& encoding : init.sendEncodings) {
826 auto encodingData = fromRTCEncodingParameters(encoding);
827 GValue value = G_VALUE_INIT;
828 g_value_init(&value, GST_TYPE_STRUCTURE);
829 gst_value_set_structure(&value, encodingData.get());
830 gst_value_list_append_value(&encodingsValue, &value);
831 g_value_unset(&value);
832 }
833 }
834 gst_structure_take_value(initData.get(), "encodings", &encodingsValue);
835
836 GRefPtr<GstWebRTCRTPTransceiver> rtcTransceiver;
837 g_signal_emit_by_name(m_webrtcBin.get(), "add-transceiver", direction, caps.get(), &rtcTransceiver.outPtr());
838 if (!rtcTransceiver)
839 return std::nullopt;
840
841 auto transceiver = makeUnique<GStreamerRtpTransceiverBackend>(WTFMove(rtcTransceiver));
842 return GStreamerMediaEndpoint::Backends { transceiver->createSenderBackend(m_peerConnectionBackend, WTFMove(source), WTFMove(initData)), transceiver->createReceiverBackend(), WTFMove(transceiver) };
843}
844
845std::optional<GStreamerMediaEndpoint::Backends> GStreamerMediaEndpoint::addTransceiver(const String& trackKind, const RTCRtpTransceiverInit& init)
846{
847 return createTransceiverBackends(trackKind, init, nullptr);
848}
849
850GStreamerRtpSenderBackend::Source GStreamerMediaEndpoint::createSourceForTrack(MediaStreamTrack& track)
851{
852 if (track.privateTrack().hasAudio())
853 return RealtimeOutgoingAudioSourceGStreamer::create(track.privateTrack());
854
855 RELEASE_ASSERT(track.privateTrack().hasVideo());
856 return RealtimeOutgoingVideoSourceGStreamer::create(track.privateTrack());
857}
858
859std::optional<GStreamerMediaEndpoint::Backends> GStreamerMediaEndpoint::addTransceiver(MediaStreamTrack& track, const RTCRtpTransceiverInit& init)
860{
861 return createTransceiverBackends(track.kind(), init, createSourceForTrack(track));
862}
863
864void GStreamerMediaEndpoint::setSenderSourceFromTrack(GStreamerRtpSenderBackend&, MediaStreamTrack&)
865{
866 GST_FIXME_OBJECT(m_pipeline.get(), "setSenderSourceFromTrack");
867 notImplemented();
868}
869
870std::unique_ptr<GStreamerRtpTransceiverBackend> GStreamerMediaEndpoint::transceiverBackendFromSender(GStreamerRtpSenderBackend& backend)
871{
872 GRefPtr<GArray> transceivers;
873 g_signal_emit_by_name(m_webrtcBin.get(), "get-transceivers", &transceivers.outPtr());
874
875 GST_DEBUG_OBJECT(m_pipeline.get(), "Looking for sender %p in %u existing transceivers", backend.rtcSender(), transceivers->len);
876 for (unsigned i = 0; i < transceivers->len; i++) {
877 GstWebRTCRTPTransceiver* current = g_array_index(transceivers.get(), GstWebRTCRTPTransceiver*, i);
878 GRefPtr<GstWebRTCRTPSender> sender;
879 g_object_get(current, "sender", &sender.outPtr(), nullptr);
880
881 if (!sender)
882 continue;
883 if (sender.get() == backend.rtcSender())
884 return WTF::makeUnique<GStreamerRtpTransceiverBackend>(current);
885 }
886
887 return nullptr;
888}
889
890void GStreamerMediaEndpoint::addIceCandidate(GStreamerIceCandidate& candidate, PeerConnectionBackend::AddIceCandidateCallback&& callback)
891{
892 GST_DEBUG_OBJECT(m_pipeline.get(), "Adding ICE candidate %s", candidate.candidate.utf8().data());
893
894 if (!candidate.candidate.startsWith("candidate:")) {
895 callOnMainThread([task = createSharedTask<PeerConnectionBackend::AddIceCandidateCallbackFunction>(WTFMove(callback))]() mutable {
896 task->run(Exception { OperationError, "Expect line: candidate:<candidate-str>" });
897 });
898 return;
899 }
900
901 auto parsedCandidate = parseIceCandidateSDP(candidate.candidate);
902 if (!parsedCandidate) {
903 callOnMainThread([task = createSharedTask<PeerConnectionBackend::AddIceCandidateCallbackFunction>(WTFMove(callback))]() mutable {
904 task->run(Exception { OperationError, "Error processing ICE candidate" });
905 });
906 return;
907 }
908
909 // FIXME: invalid sdpMLineIndex exception not relayed from webrtcbin.
910 // FIXME: Ideally this should pass the result/error to a GstPromise object.
911 g_signal_emit_by_name(m_webrtcBin.get(), "add-ice-candidate", candidate.sdpMLineIndex, candidate.candidate.utf8().data());
912 callOnMainThread([task = createSharedTask<PeerConnectionBackend::AddIceCandidateCallbackFunction>(WTFMove(callback)), descriptions = descriptionsFromWebRTCBin(m_webrtcBin.get())]() mutable {
913 task->run(WTFMove(descriptions));
914 });
915}
916
917std::unique_ptr<RTCDataChannelHandler> GStreamerMediaEndpoint::createDataChannel(const String& label, const RTCDataChannelInit& options)
918{
919 if (!m_webrtcBin)
920 return nullptr;
921
922 auto init = GStreamerDataChannelHandler::fromRTCDataChannelInit(options);
923 GST_DEBUG_OBJECT(m_pipeline.get(), "Creating data channel for init options %" GST_PTR_FORMAT, init.get());
924 GRefPtr<GstWebRTCDataChannel> channel;
925 g_signal_emit_by_name(m_webrtcBin.get(), "create-data-channel", label.utf8().data(), init.get(), &channel.outPtr());
926 if (!channel)
927 return nullptr;
928
929 return WTF::makeUnique<GStreamerDataChannelHandler>(WTFMove(channel));
930}
931
932void GStreamerMediaEndpoint::onDataChannel(GstWebRTCDataChannel* dataChannel)
933{
934 GST_DEBUG_OBJECT(m_pipeline.get(), "Incoming data channel");
935 GRefPtr<GstWebRTCDataChannel> channel = dataChannel;
936 callOnMainThread([protectedThis = Ref(*this), dataChannel = WTFMove(channel)]() mutable {
937 GST_DEBUG_OBJECT(protectedThis->m_pipeline.get(), "Incoming data channel 1");
938 if (protectedThis->isStopped())
939 return;
940 GST_DEBUG_OBJECT(protectedThis->m_pipeline.get(), "Incoming data channel 2");
941 auto& connection = protectedThis->m_peerConnectionBackend.connection();
942 connection.dispatchEvent(GStreamerDataChannelHandler::createDataChannelEvent(*connection.document(), WTFMove(dataChannel)));
943 });
944}
945
946void GStreamerMediaEndpoint::close()
947{
948 // https://gitlab.freedesktop.org/gstreamer/gst-plugins-bad/-/issues/1181
949 GST_DEBUG_OBJECT(m_pipeline.get(), "Closing");
950 if (m_pipeline)
951 gst_element_set_state(m_pipeline.get(), GST_STATE_READY);
952
953#if !RELEASE_LOG_DISABLED
954 stopLoggingStats();
955#endif
956}
957
958void GStreamerMediaEndpoint::stop()
959{
960#if !RELEASE_LOG_DISABLED
961 stopLoggingStats();
962#endif
963
964 if (!m_pipeline)
965 return;
966
967 GST_DEBUG_OBJECT(m_pipeline.get(), "Stopping");
968 teardownPipeline();
969}
970
971void GStreamerMediaEndpoint::suspend()
972{
973 if (!m_pipeline)
974 return;
975
976 GST_DEBUG_OBJECT(m_pipeline.get(), "Suspending");
977 gst_element_set_state(m_pipeline.get(), GST_STATE_PAUSED);
978}
979
980void GStreamerMediaEndpoint::resume()
981{
982 if (!m_pipeline)
983 return;
984
985 GST_DEBUG_OBJECT(m_pipeline.get(), "Resuming");
986 gst_element_set_state(m_pipeline.get(), GST_STATE_PLAYING);
987}
988
989void GStreamerMediaEndpoint::onNegotiationNeeded()
990{
991 m_isNegotiationNeeded = true;
992
993 GST_DEBUG_OBJECT(m_pipeline.get(), "Checking negotiation status");
994
995 callOnMainThread([protectedThis = Ref(*this)] {
996 if (protectedThis->isStopped())
997 return;
998 GST_DEBUG_OBJECT(protectedThis->m_pipeline.get(), "Negotiation needed!");
999 protectedThis->m_peerConnectionBackend.markAsNeedingNegotiation(0);
1000 });
1001}
1002
1003void GStreamerMediaEndpoint::onIceConnectionChange()
1004{
1005 GstWebRTCICEConnectionState state;
1006 g_object_get(m_webrtcBin.get(), "ice-connection-state", &state, nullptr);
1007 callOnMainThread([protectedThis = Ref(*this), connectionState = toRTCIceConnectionState(state)] {
1008 if (protectedThis->isStopped())
1009 return;
1010 auto& connection = protectedThis->m_peerConnectionBackend.connection();
1011 if (connection.iceConnectionState() != connectionState)
1012 connection.updateIceConnectionState(connectionState);
1013 });
1014}
1015
1016void GStreamerMediaEndpoint::onIceGatheringChange()
1017{
1018 GstWebRTCICEGatheringState state;
1019 g_object_get(m_webrtcBin.get(), "ice-gathering-state", &state, nullptr);
1020 callOnMainThread([protectedThis = Ref(*this), state] {
1021 if (protectedThis->isStopped())
1022 return;
1023 auto& connection = protectedThis->m_peerConnectionBackend.connection();
1024 if (state == GST_WEBRTC_ICE_GATHERING_STATE_COMPLETE)
1025 protectedThis->m_peerConnectionBackend.doneGatheringCandidates();
1026 else if (state == GST_WEBRTC_ICE_GATHERING_STATE_GATHERING)
1027 connection.updateIceGatheringState(RTCIceGatheringState::Gathering);
1028 else if (state == GST_WEBRTC_ICE_GATHERING_STATE_NEW)
1029 connection.updateIceGatheringState(RTCIceGatheringState::New);
1030 });
1031}
1032
1033void GStreamerMediaEndpoint::onIceCandidate(guint sdpMLineIndex, gchararray candidate)
1034{
1035 // FIXME: Get mid from candidate?
1036 GUniqueOutPtr<GstWebRTCSessionDescription> description;
1037 g_object_get(m_webrtcBin.get(), "local-description", &description.outPtr(), nullptr);
1038 if (!description)
1039 return;
1040
1041 const GstSDPMedia* media = gst_sdp_message_get_media(description->sdp, sdpMLineIndex);
1042 String candidateMid(gst_sdp_media_get_attribute_val(media, "mid"));
1043 String candidateSDP(candidate);
1044 callOnMainThread([protectedThis = Ref(*this), mid = WTFMove(candidateMid), sdp = WTFMove(candidateSDP), sdpMLineIndex, descriptions = descriptionsFromWebRTCBin(m_webrtcBin.get())]() mutable {
1045 if (protectedThis->isStopped())
1046 return;
1047 protectedThis->m_peerConnectionBackend.newICECandidate(WTFMove(sdp), WTFMove(mid), sdpMLineIndex, { }, WTFMove(descriptions));
1048 });
1049}
1050
1051void GStreamerMediaEndpoint::createSessionDescriptionSucceeded(GUniquePtr<GstWebRTCSessionDescription>&& description)
1052{
1053 callOnMainThread([protectedThis = Ref(*this), description = WTFMove(description)] {
1054 if (protectedThis->isStopped())
1055 return;
1056
1057 GUniquePtr<char> sdp(gst_sdp_message_as_text(description->sdp));
1058 if (protectedThis->m_isInitiator)
1059 protectedThis->m_peerConnectionBackend.createOfferSucceeded(sdp.get());
1060 else
1061 protectedThis->m_peerConnectionBackend.createAnswerSucceeded(sdp.get());
1062 });
1063}
1064
1065void GStreamerMediaEndpoint::createSessionDescriptionFailed(GUniquePtr<GError>&& error)
1066{
1067 callOnMainThread([protectedThis = Ref(*this), error = WTFMove(error)] {
1068 if (protectedThis->isStopped())
1069 return;
1070
1071 auto exc = Exception { OperationError, error ? error->message : "Unknown Error" };
1072 if (protectedThis->m_isInitiator)
1073 protectedThis->m_peerConnectionBackend.createOfferFailed(WTFMove(exc));
1074 else
1075 protectedThis->m_peerConnectionBackend.createAnswerFailed(WTFMove(exc));
1076 });
1077}
1078
1079void GStreamerMediaEndpoint::collectTransceivers()
1080{
1081 GRefPtr<GArray> transceivers;
1082 g_signal_emit_by_name(m_webrtcBin.get(), "get-transceivers", &transceivers.outPtr());
1083 for (unsigned i = 0; i < transceivers->len; i++) {
1084 GstWebRTCRTPTransceiver* current = g_array_index(transceivers.get(), GstWebRTCRTPTransceiver*, i);
1085
1086 auto* existingTransceiver = m_peerConnectionBackend.existingTransceiver([&](auto& transceiverBackend) {
1087 return current == transceiverBackend.rtcTransceiver();
1088 });
1089 if (existingTransceiver)
1090 continue;
1091
1092 GRefPtr<GstWebRTCRTPReceiver> receiver;
1093 GUniqueOutPtr<char> mid;
1094 g_object_get(current, "receiver", &receiver.outPtr(), "mid", &mid.outPtr(), nullptr);
1095
1096 if (!receiver)
1097 continue;
1098
1099 if (!mid)
1100 continue;
1101
1102 m_peerConnectionBackend.newRemoteTransceiver(WTF::makeUnique<GStreamerRtpTransceiverBackend>(WTFMove(current)), m_mediaForMid.get(mid.get()));
1103 }
1104}
1105
1106#if !RELEASE_LOG_DISABLED
1107void GStreamerMediaEndpoint::gatherStatsForLogging()
1108{
1109 g_signal_emit_by_name(m_webrtcBin.get(), "get-stats", nullptr, gst_promise_new_with_change_func([](GstPromise* rawPromise, gpointer userData) {
1110 auto promise = adoptGRef(rawPromise);
1111 auto result = gst_promise_wait(promise.get());
1112 if (result != GST_PROMISE_RESULT_REPLIED)
1113 return;
1114
1115 const auto* reply = gst_promise_get_reply(promise.get());
1116 RELEASE_ASSERT(reply);
1117 if (gst_structure_has_field(reply, "error"))
1118 return;
1119
1120 auto* endPoint = reinterpret_cast<GStreamerMediaEndpoint*>(userData);
1121 endPoint->onStatsDelivered(reply);
1122 }, this, nullptr));
1123}
1124
1125class RTCStatsLogger {
1126public:
1127 explicit RTCStatsLogger(const GstStructure* stats)
1128 : m_stats(stats)
1129 { }
1130
1131 String toJSONString() const { return gstStructureToJSONString(m_stats); }
1132
1133private:
1134 const GstStructure* m_stats;
1135};
1136
1137void GStreamerMediaEndpoint::processStats(const GValue* value)
1138{
1139 if (!GST_VALUE_HOLDS_STRUCTURE(value))
1140 return;
1141
1142 const GstStructure* structure = gst_value_get_structure(value);
1143 GstWebRTCStatsType statsType;
1144 if (!gst_structure_get(structure, "type", GST_TYPE_WEBRTC_STATS_TYPE, &statsType, nullptr))
1145 return;
1146
1147 // Just check a single timestamp, inbound RTP for instance.
1148 if (!m_statsFirstDeliveredTimestamp && statsType == GST_WEBRTC_STATS_INBOUND_RTP) {
1149 double timestamp;
1150 if (gst_structure_get_double(structure, "timestamp", ×tamp)) {
1151 auto ts = Seconds::fromMilliseconds(timestamp);
1152 m_statsFirstDeliveredTimestamp = ts;
1153
1154 if (!isStopped() && m_statsLogTimer.repeatInterval() != statsLogInterval(ts)) {
1155 m_statsLogTimer.stop();
1156 m_statsLogTimer.startRepeating(statsLogInterval(ts));
1157 }
1158 }
1159 }
1160
1161 if (logger().willLog(logChannel(), WTFLogLevel::Debug)) {
1162 // Stats are very verbose, let's only display them in inspector console in verbose mode.
1163 logger().debug(LogWebRTC,
1164 Logger::LogSiteIdentifier("GStreamerMediaEndpoint", "onStatsDelivered", logIdentifier()),
1165 RTCStatsLogger { structure });
1166 } else {
1167 logger().logAlways(LogWebRTCStats,
1168 Logger::LogSiteIdentifier("GStreamerMediaEndpoint", "onStatsDelivered", logIdentifier()),
1169 RTCStatsLogger { structure });
1170 }
1171}
1172
1173void GStreamerMediaEndpoint::onStatsDelivered(const GstStructure* stats)
1174{
1175 GUniquePtr<GstStructure> statsCopy(gst_structure_copy(stats));
1176 callOnMainThread([protectedThis = Ref(*this), this, stats = WTFMove(statsCopy)] {
1177 gst_structure_foreach(stats.get(), static_cast<GstStructureForeachFunc>([](GQuark, const GValue* value, gpointer userData) -> gboolean {
1178 auto* endPoint = reinterpret_cast<GStreamerMediaEndpoint*>(userData);
1179 endPoint->processStats(value);
1180 return TRUE;
1181 }), this);
1182 });
1183}
1184#endif
1185
1186#if !RELEASE_LOG_DISABLED
1187void GStreamerMediaEndpoint::startLoggingStats()
1188{
1189 if (m_statsLogTimer.isActive())
1190 m_statsLogTimer.stop();
1191 m_statsLogTimer.startRepeating(statsLogInterval(Seconds::nan()));
1192}
1193
1194void GStreamerMediaEndpoint::stopLoggingStats()
1195{
1196 m_statsLogTimer.stop();
1197}
1198
1199WTFLogChannel& GStreamerMediaEndpoint::logChannel() const
1200{
1201 return LogWebRTC;
1202}
1203
1204Seconds GStreamerMediaEndpoint::statsLogInterval(Seconds reportTimestamp) const
1205{
1206 if (logger().willLog(logChannel(), WTFLogLevel::Info))
1207 return 2_s;
1208
1209 if (reportTimestamp - m_statsFirstDeliveredTimestamp > 15_s)
1210 return 10_s;
1211
1212 return 4_s;
1213}
1214#endif
1215
1216void GStreamerMediaEndpoint::gatherDecoderImplementationName(Function<void(String&&)>&& callback)
1217{
1218 // TODO: collect stats and lookup InboundRtp "decoder_implementation" field.
1219 callback({ });
1220}
1221
1222} // namespace WebCore
1223
1224#if !RELEASE_LOG_DISABLED
1225namespace WTF {
1226
1227template<typename Type>
1228struct LogArgument;
1229
1230template <>
1231struct LogArgument<WebCore::RTCStatsLogger> {
1232 static String toString(const WebCore::RTCStatsLogger& logger)
1233 {
1234 return String(logger.toJSONString());
1235 }
1236};
1237
1238}; // namespace WTF
1239#endif // !RELEASE_LOG_DISABLED
1240
1241#endif // USE(GSTREAMER_WEBRTC)