[GStreamer][WebRTC] Emit 'track' event before resolving setRemoteDescription's promise

https://bugs.webkit.org/show_bug.cgi?id=275685

Reviewed by Philippe Normand.

The track event should be emitted after sucessfully applying a remote
description and before the promise from setRemoteDescription is resolved.
However, up until now the GstWebRTC backend emitted the event based on
the 'pad-added' signal. The timing of the signal doesn't match what the
spec expects.

Fix the issue by querying the transceiver states after applying a remote
description sucessfully. By providing the transceiver states to
PeerConnectionBackend::setRemoteDescriptionSucceeded(), it's able to
emit the track events correctly acccording to the spec.

A test is added to check that the track event is emitted before
setRemoteDescription resolves, which would fail before this patch in
the GstWebRTC backend.

* LayoutTests/platform/glib/TestExpectations:
* LayoutTests/platform/glib/fast/mediastream/RTCPeerConnection-inspect-answer-expected.txt:
* LayoutTests/platform/glib/fast/mediastream/RTCPeerConnection-media-setup-single-dialog-expected.txt:
* LayoutTests/platform/glib/fast/mediastream/RTCPeerConnection-setRemoteDescription-offer-expected.txt:
* LayoutTests/platform/glib/imported/w3c/web-platform-tests/webrtc/RTCPeerConnection-setLocalDescription-parameterless.https-expected.txt: Removed.
* LayoutTests/platform/glib/imported/w3c/web-platform-tests/webrtc/protocol/split.https-expected.txt: Added.
* LayoutTests/webrtc/setRemoteDescription-track-expected.txt: Added.
* LayoutTests/webrtc/setRemoteDescription-track.html: Added.
* Source/WebCore/Modules/mediastream/PeerConnectionBackend.cpp:
(WebCore::PeerConnectionBackend::setRemoteDescriptionSucceeded):
(WebCore::PeerConnectionBackend::setRemoteDescriptionFailed):
(WebCore::PeerConnectionBackend::stop):
(WebCore::PeerConnectionBackend::dispatchTrackEvent): Deleted.
(WebCore::PeerConnectionBackend::addPendingTrackEvent): Deleted.
* Source/WebCore/Modules/mediastream/PeerConnectionBackend.h:
* Source/WebCore/Modules/mediastream/gstreamer/GStreamerMediaEndpoint.cpp:
(WebCore::GStreamerMediaEndpoint::initializePipeline):
(WebCore::GStreamerMediaEndpointTransceiverState::isolatedCopy):
(WebCore::getMediaStreamIdsFromSDPMedia):
(WebCore::isRecvDirection):
(WebCore::toGStreamerMediaEndpointTransceiverState):
(WebCore::transceiverStatesFromWebRTCBin):
(WebCore::GStreamerMediaEndpoint::doSetLocalDescription):
(WebCore::GStreamerMediaEndpoint::setTransceiverCodecPreferences):
(WebCore::GStreamerMediaEndpoint::doSetRemoteDescription):
(WebCore::GStreamerMediaEndpoint::setDescription):
(WebCore::GStreamerMediaEndpoint::connectIncomingTrack):
* Source/WebCore/Modules/mediastream/gstreamer/GStreamerMediaEndpoint.h:
* Source/WebCore/Modules/mediastream/gstreamer/GStreamerPeerConnectionBackend.cpp:
(WebCore::GStreamerPeerConnectionBackend::addPendingTrackEvent): Deleted.
(WebCore::GStreamerPeerConnectionBackend::dispatchPendingTrackEvents): Deleted.
* Source/WebCore/Modules/mediastream/gstreamer/GStreamerPeerConnectionBackend.h:
* Source/WebCore/platform/mediastream/MediaStreamTrackPrivate.cpp:
(WebCore::MediaStreamTrackPrivate::dataFlowStarted):
* Source/WebCore/platform/mediastream/MediaStreamTrackPrivate.h:
(WebCore::MediaStreamTrackPrivateObserver::dataFlowStarted):
* Source/WebCore/platform/mediastream/gstreamer/GStreamerMediaStreamSource.cpp:
(webkitMediaStreamSrcCharacteristicsChanged):

Canonical link: https://commits.webkit.org/281892@main

Author: cadubentzen

Source/WebCore/Modules/mediastream/PeerConnectionBackend.cpp

@@ -328,7 +328,7 @@ void PeerConnectionBackend::setRemoteDescriptionSucceeded(std::optional<Descript
         DEBUG_LOG(LOGIDENTIFIER, "Transceiver states: ", *transceiverStates);
     ASSERT(m_setDescriptionCallback);
 
-    m_peerConnection.queueTaskKeepingObjectAlive(m_peerConnection, TaskSource::Networking, [this, callback = WTFMove(m_setDescriptionCallback), descriptionStates = WTFMove(descriptionStates), transceiverStates = WTFMove(transceiverStates), sctpBackend = WTFMove(sctpBackend), maxMessageSize, events = WTFMove(m_pendingTrackEvents)]() mutable {
+    m_peerConnection.queueTaskKeepingObjectAlive(m_peerConnection, TaskSource::Networking, [this, callback = WTFMove(m_setDescriptionCallback), descriptionStates = WTFMove(descriptionStates), transceiverStates = WTFMove(transceiverStates), sctpBackend = WTFMove(sctpBackend), maxMessageSize]() mutable {
         if (m_peerConnection.isClosed())
             return;
 
@@ -402,38 +402,17 @@ void PeerConnectionBackend::setRemoteDescriptionSucceeded(std::optional<Descript
 
                 track->source().setMuted(false);
             }
-        } else {
-            // FIXME: Move ports out of m_pendingTrackEvents.
-            for (auto& event : events)
-                dispatchTrackEvent(event);
         }
 
         callback({ });
     });
 }
 
-void PeerConnectionBackend::dispatchTrackEvent(PendingTrackEvent& event)
-{
-    auto& track = event.track.get();
-
-    m_peerConnection.dispatchEvent(RTCTrackEvent::create(eventNames().trackEvent, Event::CanBubble::No, Event::IsCancelable::No, WTFMove(event.receiver), WTFMove(event.track), WTFMove(event.streams), WTFMove(event.transceiver)));
-    ALWAYS_LOG(LOGIDENTIFIER, "Dispatched if feasible track of type ", track.source().type());
-
-    if (m_peerConnection.isClosed())
-        return;
-
-    // FIXME: As per spec, we should set muted to 'false' when starting to receive the content from network.
-    track.source().setMuted(false);
-}
-
 void PeerConnectionBackend::setRemoteDescriptionFailed(Exception&& exception)
 {
     ASSERT(isMainThread());
     ALWAYS_LOG(LOGIDENTIFIER, "Set remote description failed:", exception.message());
 
-    ASSERT(m_pendingTrackEvents.isEmpty());
-    m_pendingTrackEvents.clear();
-
     ASSERT(m_setDescriptionCallback);
     m_peerConnection.queueTaskKeepingObjectAlive(m_peerConnection, TaskSource::Networking, [this, callback = WTFMove(m_setDescriptionCallback), exception = WTFMove(exception)]() mutable {
         if (m_peerConnection.isClosed())
@@ -454,12 +433,6 @@ void PeerConnectionBackend::iceGatheringStateChanged(RTCIceGatheringState state)
     });
 }
 
-void PeerConnectionBackend::addPendingTrackEvent(PendingTrackEvent&& event)
-{
-    ASSERT(!m_peerConnection.isStopped());
-    m_pendingTrackEvents.append(WTFMove(event));
-}
-
 static String extractIPAddress(StringView sdp)
 {
     unsigned counter = 0;
@@ -590,8 +563,6 @@ void PeerConnectionBackend::stop()
     m_offerAnswerCallback = nullptr;
     m_setDescriptionCallback = nullptr;
 
-    m_pendingTrackEvents.clear();
-
     doStop();
 }
 

Source/WebCore/Modules/mediastream/PeerConnectionBackend.h

@@ -228,16 +228,6 @@ class PeerConnectionBackend
 
     void validateSDP(const String&) const;
 
-    struct PendingTrackEvent {
-        Ref<RTCRtpReceiver> receiver;
-        Ref<MediaStreamTrack> track;
-        Vector<RefPtr<MediaStream>> streams;
-        RefPtr<RTCRtpTransceiver> transceiver;
-    };
-    void addPendingTrackEvent(PendingTrackEvent&&);
-
-    void dispatchTrackEvent(PendingTrackEvent&);
-
 private:
     virtual void doCreateOffer(RTCOfferOptions&&) = 0;
     virtual void doCreateAnswer(RTCAnswerOptions&&) = 0;
@@ -255,8 +245,6 @@ class PeerConnectionBackend
 
     bool m_shouldFilterICECandidates { true };
 
-    Vector<PendingTrackEvent> m_pendingTrackEvents;
-
 #if !RELEASE_LOG_DISABLED
     Ref<const Logger> m_logger;
     const void* m_logIdentifier;

Source/WebCore/Modules/mediastream/gstreamer/GStreamerMediaEndpoint.cpp

@@ -170,6 +170,10 @@ bool GStreamerMediaEndpoint::initializePipeline()
     }), this);
 
 #ifndef GST_DISABLE_GST_DEBUG
+    g_signal_connect_swapped(m_webrtcBin.get(), "on-new-transceiver", G_CALLBACK(+[](GStreamerMediaEndpoint* endPoint, GstWebRTCRTPTransceiver* transceiver) {
+        GST_DEBUG_OBJECT(endPoint->m_webrtcBin.get(), "New transceiver: %" GST_PTR_FORMAT, transceiver);
+    }), this);
+
     g_signal_connect(m_webrtcBin.get(), "notify::connection-state", G_CALLBACK(+[](GstElement* webrtcBin, GParamSpec*, GStreamerMediaEndpoint* endPoint) {
         GstWebRTCPeerConnectionState state;
         g_object_get(webrtcBin, "connection-state", &state, nullptr);
@@ -370,6 +374,119 @@ static std::optional<PeerConnectionBackend::DescriptionStates> descriptionsFromW
     };
 }
 
+struct GStreamerMediaEndpointTransceiverState {
+    String mid;
+    Vector<String> receiverStreamIds;
+    std::optional<RTCRtpTransceiverDirection> firedDirection;
+
+    GStreamerMediaEndpointTransceiverState isolatedCopy() &&;
+};
+
+inline GStreamerMediaEndpointTransceiverState GStreamerMediaEndpointTransceiverState::isolatedCopy() &&
+{
+    return {
+        WTFMove(mid).isolatedCopy(),
+        crossThreadCopy(WTFMove(receiverStreamIds)),
+        firedDirection
+    };
+}
+
+Vector<String> getMediaStreamIdsFromSDPMedia(const GstSDPMedia& media)
+{
+    HashSet<String> mediaStreamIdsSet;
+    for (guint i = 0; i < gst_sdp_media_attributes_len(&media); ++i) {
+        const auto attribute = gst_sdp_media_get_attribute(&media, i);
+        if (!g_strcmp0(attribute->key, "msid")) {
+            auto components = String::fromUTF8(attribute->value).split(' ');
+            if (components.size() < 2)
+                continue;
+            mediaStreamIdsSet.add(components[0]);
+        }
+        // MSID may also come in ssrc attributes, specially if they're in an SDP answer. They look like:
+        // a=ssrc:3612593434 msid:e1019f4a-0983-4863-b923-b75903cced2c webrtctransceiver1
+        if (!g_strcmp0(attribute->key, "ssrc")) {
+            auto outerComponents = String::fromUTF8(attribute->value).split(' ');
+            for (auto& outer : outerComponents) {
+                auto innerComponents = outer.split(':');
+                if (innerComponents.size() < 2)
+                    continue;
+                if (innerComponents[0] == "msid"_s)
+                    mediaStreamIdsSet.add(innerComponents[1]);
+            }
+        }
+    }
+    Vector<String> mediaStreamIds;
+    mediaStreamIds.reserveCapacity(mediaStreamIdsSet.size());
+    for (const auto& msid : mediaStreamIdsSet)
+        mediaStreamIds.append(msid);
+    return mediaStreamIds;
+}
+
+inline bool isRecvDirection(GstWebRTCRTPTransceiverDirection direction)
+{
+    return direction == GST_WEBRTC_RTP_TRANSCEIVER_DIRECTION_SENDRECV || direction == GST_WEBRTC_RTP_TRANSCEIVER_DIRECTION_RECVONLY;
+}
+
+static std::optional<GStreamerMediaEndpointTransceiverState> toGStreamerMediaEndpointTransceiverState(GstElement* webrtcBin, GstWebRTCRTPTransceiver* transceiver)
+{
+    GRefPtr<GstWebRTCRTPReceiver> receiver;
+    GUniqueOutPtr<char> mid;
+    GstWebRTCRTPTransceiverDirection currentDirection;
+    guint mLineIndex;
+    g_object_get(transceiver, "receiver", &receiver.outPtr(), "current-direction", &currentDirection, "mlineindex", &mLineIndex, "mid", &mid.outPtr(), nullptr);
+#ifndef GST_DISABLE_GST_DEBUG
+    GUniquePtr<char> desc(g_enum_to_string(GST_TYPE_WEBRTC_RTP_TRANSCEIVER_DIRECTION, currentDirection));
+    GST_TRACE_OBJECT(webrtcBin, "Receiver = %" GST_PTR_FORMAT ", current-direction = %s, mlineindex = %u, mid = %s", receiver.get(), desc.get(), mLineIndex, GST_STR_NULL(mid.get()));
+#endif
+
+    GUniqueOutPtr<GstWebRTCSessionDescription> localDescription, remoteDescription;
+    g_object_get(webrtcBin, "local-description", &localDescription.outPtr(), "remote-description", &remoteDescription.outPtr(), nullptr);
+
+#ifndef GST_DISABLE_GST_DEBUG
+    if (localDescription) {
+        GUniquePtr<char> sdp(gst_sdp_message_as_text(localDescription->sdp));
+        GST_TRACE_OBJECT(webrtcBin, "Local-description:\n%s", sdp.get());
+    }
+    if (remoteDescription) {
+        GUniquePtr<char> sdp(gst_sdp_message_as_text(remoteDescription->sdp));
+        GST_TRACE_OBJECT(webrtcBin, "Remote-description:\n%s", sdp.get());
+    }
+#endif
+
+    Vector<String> streamIds;
+    if (remoteDescription && remoteDescription->sdp && mLineIndex < gst_sdp_message_medias_len(remoteDescription->sdp)) {
+        const GstSDPMedia* media = gst_sdp_message_get_media(remoteDescription->sdp, mLineIndex);
+        if (isRecvDirection(currentDirection))
+            streamIds = getMediaStreamIdsFromSDPMedia(*media);
+    }
+
+    if (UNLIKELY(!mid))
+        return { };
+
+    return { { String::fromUTF8(mid.get()), WTFMove(streamIds), { toRTCRtpTransceiverDirection(currentDirection) } } };
+}
+
+static Vector<GStreamerMediaEndpointTransceiverState> transceiverStatesFromWebRTCBin(GstElement* webrtcBin)
+{
+    Vector<GStreamerMediaEndpointTransceiverState> states;
+    GRefPtr<GArray> transceivers;
+    g_signal_emit_by_name(webrtcBin, "get-transceivers", &transceivers.outPtr());
+    GST_TRACE_OBJECT(webrtcBin, "Filling transceiver states for %u transceivers", transceivers ? transceivers->len : 0);
+    if (!transceivers || !transceivers->len)
+        return states;
+
+    states.reserveInitialCapacity(transceivers->len);
+    for (unsigned i = 0; i < transceivers->len; i++) {
+        GstWebRTCRTPTransceiver* transceiver = g_array_index(transceivers.get(), GstWebRTCRTPTransceiver*, i);
+        auto state = toGStreamerMediaEndpointTransceiverState(webrtcBin, transceiver);
+        if (!state)
+            continue;
+        states.append(WTFMove(*state));
+    }
+
+    return states;
+}
+
 void GStreamerMediaEndpoint::doSetLocalDescription(const RTCSessionDescription* description)
 {
     RefPtr initialDescription = description;
@@ -458,6 +575,19 @@ void GStreamerMediaEndpoint::doSetLocalDescription(const RTCSessionDescription*
             }
         }
 
+#ifndef GST_DISABLE_GST_DEBUG
+        auto dotFileName = makeString(GST_OBJECT_NAME(m_pipeline.get()), ".setLocalDescription"_s);
+        GST_DEBUG_BIN_TO_DOT_FILE_WITH_TS(GST_BIN(m_pipeline.get()), GST_DEBUG_GRAPH_SHOW_ALL, dotFileName.utf8().data());
+#endif
+
+        auto rtcTransceiverStates = transceiverStatesFromWebRTCBin(m_webrtcBin.get());
+        auto transceiverStates = WTF::map(rtcTransceiverStates, [this](auto& state) -> PeerConnectionBackend::TransceiverState {
+            auto streams = WTF::map(state.receiverStreamIds, [this](auto& id) -> RefPtr<MediaStream> {
+                return &mediaStreamFromRTCStream(id);
+            });
+            return { WTFMove(state.mid), WTFMove(streams), state.firedDirection };
+        });
+
         GRefPtr<GstWebRTCSCTPTransport> transport;
         g_object_get(m_webrtcBin.get(), "sctp-transport", &transport.outPtr(), nullptr);
 
@@ -468,7 +598,7 @@ void GStreamerMediaEndpoint::doSetLocalDescription(const RTCSessionDescription*
             maxMessageSize = static_cast<double>(maxMessageSizeValue);
         }
 
-        m_peerConnectionBackend.setLocalDescriptionSucceeded(WTFMove(descriptions), { }, transport ? makeUnique<GStreamerSctpTransportBackend>(WTFMove(transport)) : nullptr, maxMessageSize);
+        m_peerConnectionBackend.setLocalDescriptionSucceeded(WTFMove(descriptions), WTFMove(transceiverStates), transport ? makeUnique<GStreamerSctpTransportBackend>(WTFMove(transport)) : nullptr, maxMessageSize);
     }, [protectedThis = Ref(*this), this](const GError* error) {
         if (protectedThis->isStopped())
             return;
@@ -483,6 +613,22 @@ void GStreamerMediaEndpoint::doSetLocalDescription(const RTCSessionDescription*
     });
 }
 
+void GStreamerMediaEndpoint::setTransceiverCodecPreferences(const GstSDPMedia& media, guint transceiverIdx)
+{
+    auto direction = getDirectionFromSDPMedia(&media);
+    if (direction == GST_WEBRTC_RTP_TRANSCEIVER_DIRECTION_NONE)
+        return;
+
+    GRefPtr<GstWebRTCRTPTransceiver> rtcTransceiver;
+    g_signal_emit_by_name(m_webrtcBin.get(), "get-transceiver", transceiverIdx, &rtcTransceiver.outPtr());
+    if (!rtcTransceiver)
+        return;
+
+    auto caps = capsFromSDPMedia(&media);
+    GST_TRACE_OBJECT(m_webrtcBin.get(), "Setting codec-preferences to %" GST_PTR_FORMAT, caps.get());
+    g_object_set(rtcTransceiver.get(), "codec-preferences", caps.get(), nullptr);
+}
+
 void GStreamerMediaEndpoint::doSetRemoteDescription(const RTCSessionDescription& description)
 {
     auto initialSDP = description.sdp().isolatedCopy();
@@ -494,17 +640,9 @@ void GStreamerMediaEndpoint::doSetRemoteDescription(const RTCSessionDescription&
         unsigned numberOfMedias = gst_sdp_message_medias_len(&message);
         for (unsigned i = 0; i < numberOfMedias; i++) {
             const auto* media = gst_sdp_message_get_media(&message, i);
-            auto direction = getDirectionFromSDPMedia(media);
-            if (direction == GST_WEBRTC_RTP_TRANSCEIVER_DIRECTION_NONE)
+            if (UNLIKELY(!media))
                 continue;
-
-            GRefPtr<GstWebRTCRTPTransceiver> rtcTransceiver;
-            g_signal_emit_by_name(m_webrtcBin.get(), "get-transceiver", i, &rtcTransceiver.outPtr());
-            if (!rtcTransceiver)
-                continue;
-
-            auto caps = capsFromSDPMedia(media);
-            g_object_set(rtcTransceiver.get(), "codec-preferences", caps.get(), nullptr);
+            setTransceiverCodecPreferences(*media, i);
         }
     }, [protectedThis = Ref(*this), this, initialSDP = WTFMove(initialSDP), localDescriptionSdp = WTFMove(localDescriptionSdp), localDescriptionSdpType = WTFMove(localDescriptionSdpType)](const GstSDPMessage& message) {
         if (protectedThis->isStopped())
@@ -540,6 +678,19 @@ void GStreamerMediaEndpoint::doSetRemoteDescription(const RTCSessionDescription&
             }
         }
 
+#ifndef GST_DISABLE_GST_DEBUG
+        auto dotFileName = makeString(GST_OBJECT_NAME(m_pipeline.get()), ".setRemoteDescription"_s);
+        GST_DEBUG_BIN_TO_DOT_FILE_WITH_TS(GST_BIN(m_pipeline.get()), GST_DEBUG_GRAPH_SHOW_ALL, dotFileName.utf8().data());
+#endif
+
+        auto rtcTransceiverStates = transceiverStatesFromWebRTCBin(m_webrtcBin.get());
+        auto transceiverStates = WTF::map(rtcTransceiverStates, [this](auto& state) -> PeerConnectionBackend::TransceiverState {
+            auto streams = WTF::map(state.receiverStreamIds, [this](auto& id) -> RefPtr<MediaStream> {
+                return &mediaStreamFromRTCStream(id);
+            });
+            return { WTFMove(state.mid), WTFMove(streams), state.firedDirection };
+        });
+
         GRefPtr<GstWebRTCSCTPTransport> transport;
         g_object_get(m_webrtcBin.get(), "sctp-transport", &transport.outPtr(), nullptr);
 
@@ -550,7 +701,7 @@ void GStreamerMediaEndpoint::doSetRemoteDescription(const RTCSessionDescription&
             maxMessageSize = static_cast<double>(maxMessageSizeValue);
         }
 
-        m_peerConnectionBackend.setRemoteDescriptionSucceeded(WTFMove(descriptions), { }, transport ? makeUnique<GStreamerSctpTransportBackend>(WTFMove(transport)) : nullptr, maxMessageSize);
+        m_peerConnectionBackend.setRemoteDescriptionSucceeded(WTFMove(descriptions), WTFMove(transceiverStates), transport ? makeUnique<GStreamerSctpTransportBackend>(WTFMove(transport)) : nullptr, maxMessageSize);
     }, [protectedThis = Ref(*this), this](const GError* error) {
         if (protectedThis->isStopped())
             return;
@@ -576,6 +727,8 @@ WEBKIT_DEFINE_ASYNC_DATA_STRUCT(SetDescriptionCallData)
 
 void GStreamerMediaEndpoint::setDescription(const RTCSessionDescription* description, DescriptionType descriptionType, Function<void(const GstSDPMessage&)>&& preProcessCallback, Function<void(const GstSDPMessage&)>&& successCallback, Function<void(const GError*)>&& failureCallback)
 {
+    GST_DEBUG_OBJECT(m_webrtcBin.get(), "Setting %s description", descriptionType == DescriptionType::Local ? "local" : "remote");
+
     GUniqueOutPtr<GstSDPMessage> message;
     auto sdpType = RTCSdpType::Offer;
 
@@ -990,12 +1143,9 @@ void GStreamerMediaEndpoint::connectIncomingTrack(WebRTCTrackData& data)
             GST_WARNING_OBJECT(m_pipeline.get(), "SDP media for transceiver %u not found, skipping incoming track setup", mLineIndex);
             return;
         }
-
         transceiver = &m_peerConnectionBackend.newRemoteTransceiver(makeUnique<GStreamerRtpTransceiverBackend>(WTFMove(rtcTransceiver)), data.type, trackIdFromSDPMedia(*media));
     }
 
-    m_peerConnectionBackend.addPendingTrackEvent({ Ref(transceiver->receiver()), Ref(transceiver->receiver().track()), { }, Ref(*transceiver) });
-
     auto mediaStreamBin = adoptGRef(gst_bin_get_by_name(GST_BIN_CAST(m_pipeline.get()), data.mediaStreamBinName.ascii().data()));
     auto& track = transceiver->receiver().track();
     auto& source = track.privateTrack().source();
@@ -1009,22 +1159,32 @@ void GStreamerMediaEndpoint::connectIncomingTrack(WebRTCTrackData& data)
             return;
     }
 
-    auto& mediaStream = mediaStreamFromRTCStream(data.mediaStreamId);
-    mediaStream.addTrackFromPlatform(track);
+    m_pendingIncomingMediaStreamIDs.append(data.mediaStreamId);
 
-    for (auto& processor : m_trackProcessors.values()) {
-        if (!processor->isReady())
-            return;
+    unsigned totalExpectedMediaTracks = 0;
+    for (unsigned i = 0; i < gst_sdp_message_medias_len(description->sdp); i++) {
+        const auto media = gst_sdp_message_get_media(description->sdp, i);
+        const char* mediaType = gst_sdp_media_get_media(media);
+        if (g_str_equal(mediaType, "audio") || g_str_equal(mediaType, "video"))
+            totalExpectedMediaTracks++;
     }
 
-    GST_DEBUG_OBJECT(m_pipeline.get(), "Incoming streams gathered, now dispatching track events");
-    m_peerConnectionBackend.dispatchPendingTrackEvents(mediaStream);
-    gst_element_set_state(m_pipeline.get(), GST_STATE_PLAYING);
+    GST_DEBUG_OBJECT(m_pipeline.get(), "Expecting %u media tracks", totalExpectedMediaTracks);
+    if (m_pendingIncomingMediaStreamIDs.size() < totalExpectedMediaTracks) {
+        GST_DEBUG_OBJECT(m_pipeline.get(), "Only %zu track(s) received so far", m_pendingIncomingMediaStreamIDs.size());
+        return;
+    }
 
-#ifndef GST_DISABLE_GST_DEBUG
-    auto dotFileName = makeString(GST_OBJECT_NAME(m_pipeline.get()), ".connected-"_s, data.mediaStreamId);
-    GST_DEBUG_BIN_TO_DOT_FILE_WITH_TS(GST_BIN(m_pipeline.get()), GST_DEBUG_GRAPH_SHOW_ALL, dotFileName.utf8().data());
-#endif
+    for (auto& mediaStreamID : m_pendingIncomingMediaStreamIDs) {
+        auto& mediaStream = mediaStreamFromRTCStream(mediaStreamID);
+        GST_DEBUG_OBJECT(m_pipeline.get(), "Incoming stream %s ready, notifying observers", mediaStreamID.ascii().data());
+        mediaStream.privateStream().forEachTrack([](auto& track) {
+            GST_DEBUG("Incoming stream has track %s", track.id().ascii().data());
+            track.dataFlowStarted();
+        });
+    }
+
+    m_pendingIncomingMediaStreamIDs.clear();
     gst_element_set_state(m_pipeline.get(), GST_STATE_PLAYING);
 }