/*
 *  Copyright (C) 2009, 2010 Sebastian Dröge <sebastian.droege@collabora.co.uk>
 *  Copyright (C) 2013 Collabora Ltd.
 *  Copyright (C) 2013 Orange
 *  Copyright (C) 2014, 2015 Sebastian Dröge <sebastian@centricular.com>
 *  Copyright (C) 2015, 2016, 2018, 2019, 2020, 2021 Metrological Group B.V.
 *  Copyright (C) 2015, 2016, 2018, 2019, 2020, 2021 Igalia, S.L
 *
 *  This library is free software; you can redistribute it and/or
 *  modify it under the terms of the GNU Lesser General Public
 *  License as published by the Free Software Foundation; either
 *  version 2 of the License, or (at your option) any later version.
 *
 *  This library is distributed in the hope that it will be useful,
 *  but WITHOUT ANY WARRANTY; without even the implied warranty of
 *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 *  Lesser General Public License for more details.
 *
 *  You should have received a copy of the GNU Lesser General Public
 *  License along with this library; if not, write to the Free Software
 *  Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301  USA
 */

#include "config.h"
#include "WebKitMediaSourceGStreamer.h"

#if ENABLE(VIDEO) && ENABLE(MEDIA_SOURCE) && USE(GSTREAMER)

#include "GStreamerCommon.h"
#include "MediaSourceTrackGStreamer.h"
#include "VideoTrackPrivateGStreamer.h"
#include <cassert>
#include <gst/gst.h>
#include <wtf/Condition.h>
#include <wtf/DataMutex.h>
#include <wtf/HashMap.h>
#include <wtf/MainThread.h>
#include <wtf/MainThreadData.h>
#include <wtf/RefPtr.h>
#include <wtf/glib/WTFGType.h>
#include <wtf/text/AtomString.h>
#include <wtf/text/AtomStringHash.h>
#include <wtf/text/CString.h>

using namespace WTF;
using namespace WebCore;

GST_DEBUG_CATEGORY_STATIC(webkit_media_src_debug);
#define GST_CAT_DEFAULT webkit_media_src_debug

#define webkit_media_src_parent_class parent_class
#define WEBKIT_MEDIA_SRC_CATEGORY_INIT GST_DEBUG_CATEGORY_INIT(webkit_media_src_debug, "webkitmediasrc", 0, "websrc element");

static GstStaticPadTemplate srcTemplate = GST_STATIC_PAD_TEMPLATE("src_%s", GST_PAD_SRC,
    GST_PAD_SOMETIMES, GST_STATIC_CAPS_ANY);

enum {
    PROP_0,
    PROP_N_AUDIO,
    PROP_N_VIDEO,
    PROP_N_TEXT,
    PROP_LAST
};

struct Stream;

struct WebKitMediaSrcPrivate {
    HashMap<AtomString, RefPtr<Stream>> streams;
    Stream* streamByName(const AtomString& name)
    {
        ASSERT(isMainThread());
        Stream* stream = streams.get(name);
        ASSERT(stream);
        return stream;
    }

    // Used for stream-start events, shared by all streams.
    const unsigned groupId { gst_util_group_id_next() };

    // Set once when the source is started. Not changed after.
    GRefPtr<GstStreamCollection> collection;
    bool isStarted() { return collection; }

    // Changed on seeks.
    GstClockTime startTime { 0 };
    double rate { 1.0 };

    // Only used by URI Handler API implementation.
    GUniquePtr<char> uri;
};

static void webKitMediaSrcUriHandlerInit(gpointer, gpointer);
static void webKitMediaSrcConstructed(GObject*);
static GstStateChangeReturn webKitMediaSrcChangeState(GstElement*, GstStateChange);
static gboolean webKitMediaSrcActivateMode(GstPad*, GstObject*, GstPadMode, gboolean activate);
static void webKitMediaSrcLoop(void*);
static void webKitMediaSrcTearDownStream(WebKitMediaSrc* source, const AtomString& name);
static void webKitMediaSrcGetProperty(GObject*, unsigned propId, GValue*, GParamSpec*);
static void webKitMediaSrcStreamFlush(Stream*, bool isSeekingFlush);
static gboolean webKitMediaSrcSendEvent(GstElement*, GstEvent*);

#define webkit_media_src_parent_class parent_class

struct WebKitMediaSrcPadPrivate {
    RefPtr<Stream> stream;
};

struct WebKitMediaSrcPad {
    GstPad parent;
    WebKitMediaSrcPadPrivate* priv;
};

struct WebKitMediaSrcPadClass {
    GstPadClass parent;
};

namespace WTF {

template<> GRefPtr<WebKitMediaSrcPad> adoptGRef(WebKitMediaSrcPad* ptr)
{
    ASSERT(!ptr || !g_object_is_floating(ptr));
    return GRefPtr<WebKitMediaSrcPad>(ptr, GRefPtrAdopt);
}

template<> WebKitMediaSrcPad* refGPtr<WebKitMediaSrcPad>(WebKitMediaSrcPad* ptr)
{
    if (ptr)
        gst_object_ref_sink(GST_OBJECT(ptr));

    return ptr;
}

template<> void derefGPtr<WebKitMediaSrcPad>(WebKitMediaSrcPad* ptr)
{
    if (ptr)
        gst_object_unref(ptr);
}

} // namespace WTF

static GType webkit_media_src_pad_get_type();
WEBKIT_DEFINE_TYPE(WebKitMediaSrcPad, webkit_media_src_pad, GST_TYPE_PAD);
#define WEBKIT_TYPE_MEDIA_SRC_PAD (webkit_media_src_pad_get_type())
#define WEBKIT_MEDIA_SRC_PAD(obj) (G_TYPE_CHECK_INSTANCE_CAST((obj), WEBKIT_TYPE_MEDIA_SRC_PAD, WebKitMediaSrcPad))

static void webkit_media_src_pad_class_init(WebKitMediaSrcPadClass*)
{
}

WEBKIT_DEFINE_TYPE_WITH_CODE(WebKitMediaSrc, webkit_media_src, GST_TYPE_ELEMENT,
    G_IMPLEMENT_INTERFACE(GST_TYPE_URI_HANDLER, webKitMediaSrcUriHandlerInit);
    GST_DEBUG_CATEGORY_INIT(webkit_media_src_debug, "webkitmediasrc", 0, "WebKit MSE source element"));

struct Stream : public ThreadSafeRefCounted<Stream> {
    Stream(WebKitMediaSrc* source, GRefPtr<GstPad>&& pad, Ref<MediaSourceTrackGStreamer>&& track, GRefPtr<GstStream>&& streamInfo)
        : source(source)
        , pad(WTFMove(pad))
        , track(WTFMove(track))
        , streamInfo(WTFMove(streamInfo))
        , streamingMembersDataMutex(GRefPtr(this->track->initialCaps()), source->priv->startTime, source->priv->rate)
    {
        ASSERT(this->track->initialCaps());
    }

    WebKitMediaSrc* const source;
    GRefPtr<GstPad> const pad;
    Ref<MediaSourceTrackGStreamer> track;
    GRefPtr<GstStream> streamInfo;

    struct StreamingMembers {
        StreamingMembers(GRefPtr<GstCaps>&& initialCaps, GstClockTime startTime, double rate)
            : pendingInitialCaps(WTFMove(initialCaps))
        {
            gst_segment_init(&segment, GST_FORMAT_TIME);
            segment.start = segment.time = startTime;
            segment.rate = rate;
            ASSERT(pendingInitialCaps);
        }

        bool hasPushedStreamCollectionEvent { false };
        bool wasStreamStartSent { false };
        bool doesNeedSegmentEvent { true };
        bool hasPushedFirstBuffer { false }; // Used to get a pipeline dump of the pipeline before buffers are flowing.
        GstSegment segment;
        GRefPtr<GstCaps> pendingInitialCaps;
        GRefPtr<GstCaps> previousCaps; // Caps from enqueued samples are compared to these to push CAPS events as needed.
        Condition padLinkedOrFlushedCondition;
        Condition queueChangedOrFlushedCondition;
        bool isFlushing { false };

        // Flushes before any buffer has been popped from the queue and sent downstream can be avoided just
        // by clearing the queue.
        bool hasPoppedFirstObject { false };
    };
    DataMutex<StreamingMembers> streamingMembersDataMutex;
};

static GRefPtr<GstElement> findPipeline(GRefPtr<GstElement> element)
{
    while (true) {
        GRefPtr<GstElement> parentElement = adoptGRef(GST_ELEMENT(gst_element_get_parent(element.get())));
        if (!parentElement)
            return element;
        element = parentElement;
    }
}

static GstStreamType gstStreamType(TrackPrivateBaseGStreamer::TrackType type)
{
    switch (type) {
    case TrackPrivateBaseGStreamer::TrackType::Video:
        return GST_STREAM_TYPE_VIDEO;
    case TrackPrivateBaseGStreamer::TrackType::Audio:
        return GST_STREAM_TYPE_AUDIO;
    case TrackPrivateBaseGStreamer::TrackType::Text:
        return GST_STREAM_TYPE_TEXT;
    default:
        GST_ERROR("Received unexpected stream type");
        return GST_STREAM_TYPE_UNKNOWN;
    }
}

#ifndef GST_DISABLE_GST_DEBUG
static const char* streamTypeToString(TrackPrivateBaseGStreamer::TrackType type)
{
    switch (type) {
    case TrackPrivateBaseGStreamer::TrackType::Audio:
        return "Audio";
    case TrackPrivateBaseGStreamer::TrackType::Video:
        return "Video";
    case TrackPrivateBaseGStreamer::TrackType::Text:
        return "Text";
    default:
    case TrackPrivateBaseGStreamer::TrackType::Unknown:
        return "Unknown";
    }
}
#endif // GST_DISABLE_GST_DEBUG

static void webkit_media_src_class_init(WebKitMediaSrcClass* klass)
{
    GObjectClass* oklass = G_OBJECT_CLASS(klass);
    GstElementClass* eklass = GST_ELEMENT_CLASS(klass);

    oklass->constructed = webKitMediaSrcConstructed;
    oklass->get_property = webKitMediaSrcGetProperty;

    gst_element_class_add_static_pad_template_with_gtype(eklass, &srcTemplate, webkit_media_src_pad_get_type());

    gst_element_class_set_static_metadata(eklass, "WebKit MediaSource source element", "Source/Network", "Feeds samples coming from WebKit MediaSource object", "Igalia <aboya@igalia.com>");

    eklass->change_state = webKitMediaSrcChangeState;
    eklass->send_event = webKitMediaSrcSendEvent;
    g_object_class_install_property(oklass,
        PROP_N_AUDIO,
        g_param_spec_int("n-audio", "Number Audio", "Total number of audio streams",
        0, G_MAXINT, 0, GParamFlags(G_PARAM_READABLE | G_PARAM_STATIC_STRINGS)));
    g_object_class_install_property(oklass,
        PROP_N_VIDEO,
        g_param_spec_int("n-video", "Number Video", "Total number of video streams",
        0, G_MAXINT, 0, GParamFlags(G_PARAM_READABLE | G_PARAM_STATIC_STRINGS)));
    g_object_class_install_property(oklass,
        PROP_N_TEXT,
        g_param_spec_int("n-text", "Number Text", "Total number of text streams",
        0, G_MAXINT, 0, GParamFlags(G_PARAM_READABLE | G_PARAM_STATIC_STRINGS)));
}

static void webKitMediaSrcConstructed(GObject* object)
{
    GST_CALL_PARENT(G_OBJECT_CLASS, constructed, (object));

    ASSERT(isMainThread());
    GST_OBJECT_FLAG_SET(object, GST_ELEMENT_FLAG_SOURCE);
}

void webKitMediaSrcEmitStreams(WebKitMediaSrc* source, const Vector<RefPtr<MediaSourceTrackGStreamer>>& tracks)
{
    ASSERT(isMainThread());
    ASSERT(!source->priv->isStarted());
    GST_DEBUG_OBJECT(source, "Emitting STREAM_COLLECTION");

    source->priv->collection = adoptGRef(gst_stream_collection_new("WebKitMediaSrc"));
    for (const auto& track : tracks) {
        GST_DEBUG_OBJECT(source, "Adding stream with trackId '%s' of type %s with caps %" GST_PTR_FORMAT, track->trackId().string().utf8().data(), streamTypeToString(track->type()), track->initialCaps().get());

        GRefPtr<WebKitMediaSrcPad> pad = WEBKIT_MEDIA_SRC_PAD(g_object_new(webkit_media_src_pad_get_type(), "name", makeString("src_", track->trackId()).utf8().data(), "direction", GST_PAD_SRC, NULL));
        gst_pad_set_activatemode_function(GST_PAD(pad.get()), webKitMediaSrcActivateMode);

        ASSERT(track->initialCaps());
        auto stream = adoptRef(new Stream(source, GRefPtr<GstPad>(GST_PAD(pad.get())), *track,
            adoptGRef(gst_stream_new(track->trackId().string().utf8().data(), track->initialCaps().get(), gstStreamType(track->type()), GST_STREAM_FLAG_SELECT))));
        pad->priv->stream = stream;

        gst_stream_collection_add_stream(source->priv->collection.get(), GRefPtr<GstStream>(stream->streamInfo.get()).leakRef());
        source->priv->streams.set(track->trackId(), WTFMove(stream));
    }

    gst_element_post_message(GST_ELEMENT(source), gst_message_new_stream_collection(GST_OBJECT(source), source->priv->collection.get()));

    for (const RefPtr<Stream>& stream: source->priv->streams.values()) {
        // Workaround: gst_element_add_pad() should already call gst_pad_set_active() if the element is PAUSED or
        // PLAYING. Unfortunately, as of GStreamer 1.18.2 it does so with the element lock taken, causing a deadlock
        // in gst_pad_start_task(), who tries to post a `stream-status` message in the element, which also requires
        // the element lock. Activating the pad beforehand avoids that codepath.
        // https://gitlab.freedesktop.org/gstreamer/gstreamer/-/merge_requests/210
        // FIXME: Remove this workaround when the bug gets fixed and versions without the bug are no longer in use.
        GstState state;
        gst_element_get_state(GST_ELEMENT(source), &state, nullptr, 0);
        if (state > GST_STATE_READY)
            gst_pad_set_active(GST_PAD(stream->pad.get()), true);

        GST_DEBUG_OBJECT(source, "Adding pad '%s' for stream with name '%s'", GST_OBJECT_NAME(stream->pad.get()), stream->track->trackId().string().utf8().data());
        gst_element_add_pad(GST_ELEMENT(source), GST_PAD(stream->pad.get()));
    }
    GST_DEBUG_OBJECT(source, "All pads added");
}

static void webKitMediaSrcTearDownStream(WebKitMediaSrc* source, const AtomString& name)
{
    ASSERT(isMainThread());
    Stream* stream = source->priv->streamByName(name);
    GST_DEBUG_OBJECT(source, "Tearing down stream '%s'", name.string().utf8().data());

    // Flush the source element **and** downstream. We want to stop the streaming thread and for that we need all elements downstream to be idle.
    webKitMediaSrcStreamFlush(stream, false);
    // Stop the thread now.
    gst_pad_set_active(stream->pad.get(), false);

    if (source->priv->isStarted())
        gst_element_remove_pad(GST_ELEMENT(source), stream->pad.get());
    source->priv->streams.remove(name);
}

static gboolean webKitMediaSrcActivateMode(GstPad* pad, GstObject* source, GstPadMode mode, gboolean active)
{
    if (mode != GST_PAD_MODE_PUSH) {
        GST_ERROR_OBJECT(source, "Unexpected pad mode in WebKitMediaSrc");
        return false;
    }

    if (active)
        gst_pad_start_task(pad, webKitMediaSrcLoop, pad, nullptr);
    else {
        // Unblock the streaming thread.
        RefPtr<Stream>& stream = WEBKIT_MEDIA_SRC_PAD(pad)->priv->stream;
        {
            DataMutexLocker streamingMembers { stream->streamingMembersDataMutex };
            streamingMembers->isFlushing = true;
            streamingMembers->padLinkedOrFlushedCondition.notifyOne();
            streamingMembers->queueChangedOrFlushedCondition.notifyOne();
        }
        // Following gstbasesrc implementation, this code is not flushing downstream.
        // If there is any possibility of the streaming thread being blocked downstream the caller MUST flush before.
        // Otherwise a deadlock would occur as the next function tries to join the thread.
        gst_pad_stop_task(pad);
        {
            DataMutexLocker streamingMembers { stream->streamingMembersDataMutex };
            streamingMembers->isFlushing = false;
        }
    }
    return true;
}

static void webKitMediaSrcPadLinked(GstPad* pad, GstPad*, void*)
{
    RefPtr<Stream>& stream = WEBKIT_MEDIA_SRC_PAD(pad)->priv->stream;
    DataMutexLocker streamingMembers { stream->streamingMembersDataMutex };
    streamingMembers->padLinkedOrFlushedCondition.notifyOne();
}

static void webKitMediaSrcWaitForPadLinkedOrFlush(GstPad* pad, DataMutexLocker<Stream::StreamingMembers>& streamingMembers)
{
    {
        auto locker = GstObjectLocker(pad);
        if (LIKELY(GST_PAD_IS_LINKED(pad)))
            return;

        GST_DEBUG_OBJECT(pad, "Waiting for the pad to be linked...");
        g_signal_connect(pad, "linked", G_CALLBACK(webKitMediaSrcPadLinked), nullptr);
    }

    assertIsHeld(streamingMembers.mutex());
    streamingMembers->padLinkedOrFlushedCondition.wait(streamingMembers.mutex());

    g_signal_handlers_disconnect_by_func(pad, reinterpret_cast<void*>(webKitMediaSrcPadLinked), nullptr);
    GST_DEBUG_OBJECT(pad, "Finished waiting for the pad to be linked.");
}

// Called with STREAM_LOCK.
static void webKitMediaSrcLoop(void* userData)
{
    GstPad* pad = GST_PAD(userData);
    RefPtr<Stream>& stream = WEBKIT_MEDIA_SRC_PAD(pad)->priv->stream;

    DataMutexLocker streamingMembers { stream->streamingMembersDataMutex };
    if (streamingMembers->isFlushing) {
        gst_pad_pause_task(pad);
        return;
    }

    // Since the pad can and will be added when the element is in PLAYING state, this task can start running
    // before the pad is linked. Wait for the pad to be linked to avoid buffers being lost to not-linked errors.
    webKitMediaSrcWaitForPadLinkedOrFlush(pad, streamingMembers);
    if (streamingMembers->isFlushing) {
        gst_pad_pause_task(pad);
        return;
    }
    ASSERT(gst_pad_is_linked(pad));

    // By keeping the lock we are guaranteed that a flush will not happen while we send essential events.
    // These events should never block downstream, so the lock should be released in little time in every
    // case.

    if (!streamingMembers->hasPushedStreamCollectionEvent) {
        GST_DEBUG_OBJECT(pad, "Pushing STREAM_COLLECTION event.");
        bool wasStreamCollectionSent = gst_pad_push_event(stream->pad.get(), gst_event_new_stream_collection(stream->source->priv->collection.get()));
        streamingMembers->hasPushedStreamCollectionEvent = true;
        GST_DEBUG_OBJECT(pad, "STREAM_COLLECTION event has been pushed, %s was returned.", boolForPrinting(wasStreamCollectionSent));
        // Initial events like this must go through, flushes (including tearing down the element) is not allowed until
        // `hasPushedFirstBuffer` has been set to true.
        ASSERT(wasStreamCollectionSent);
    }

    if (!streamingMembers->wasStreamStartSent) {
        GUniquePtr<char> streamId { g_strdup_printf("mse/%s", stream->track->trackId().string().utf8().data()) };
        GRefPtr<GstEvent> event = adoptGRef(gst_event_new_stream_start(streamId.get()));
        gst_event_set_group_id(event.get(), stream->source->priv->groupId);
        gst_event_set_stream(event.get(), stream->streamInfo.get());

        GST_DEBUG_OBJECT(pad, "Pushing STREAM_START event.");
        bool wasStreamStartSent = gst_pad_push_event(pad, event.leakRef());
        streamingMembers->wasStreamStartSent = wasStreamStartSent;
        GST_DEBUG_OBJECT(pad, "STREAM_START event pushed, %s was returned.", boolForPrinting(wasStreamStartSent));
        ASSERT(wasStreamStartSent);
    }

    if (streamingMembers->pendingInitialCaps) {
        GRefPtr<GstEvent> event = adoptGRef(gst_event_new_caps(streamingMembers->pendingInitialCaps.get()));

        GST_DEBUG_OBJECT(pad, "Pushing initial CAPS event: %" GST_PTR_FORMAT, streamingMembers->pendingInitialCaps.get());
        bool wasCapsEventSent = gst_pad_push_event(pad, event.leakRef());
        GST_DEBUG_OBJECT(pad, "Pushed initial CAPS event, %s was returned.", boolForPrinting(wasCapsEventSent));

        streamingMembers->previousCaps = WTFMove(streamingMembers->pendingInitialCaps);
        ASSERT(!streamingMembers->pendingInitialCaps);
    }

    GRefPtr<GstMiniObject> object;
    {
        DataMutexLocker queue { stream->track->queueDataMutex() };
        if (!queue->isEmpty()) {
            object = queue->pop();
            streamingMembers->hasPoppedFirstObject = true;
            GST_TRACE_OBJECT(pad, "Queue not empty, popped %" GST_PTR_FORMAT, object.get());
        } else {
            queue->notifyWhenNotEmpty([&object, stream](GRefPtr<GstMiniObject>&& receivedObject) {
                ASSERT(isMainThread());
                DataMutexLocker streamingMembers { stream->streamingMembersDataMutex };
                ASSERT(!streamingMembers->isFlushing);

                object = WTFMove(receivedObject);
                streamingMembers->hasPoppedFirstObject = true;
                streamingMembers->queueChangedOrFlushedCondition.notifyAll();
            });
            GST_TRACE_OBJECT(pad, "Waiting for objects to be pushed to the track queue.");
        }
    }

    // Wait to receive an object from the queue (if we didn't get one already) or flush.
    streamingMembers->queueChangedOrFlushedCondition.wait(streamingMembers.mutex(), [&]() {
        return streamingMembers->isFlushing || object;
    });
    {
        // Ensure that notifyWhenNotEmpty()'s callback (if any) is cleared after this point.
        DataMutexLocker queue { stream->track->queueDataMutex() };
        queue->resetNotEmptyHandler();
    }
    if (streamingMembers->isFlushing) {
        gst_pad_pause_task(pad);
        return;
    }

    // We wait to get a sample before emitting the first segment. This way, if we get a seek before any
    // enqueue, we're sending only one segment. This also ensures that when such a seek is made, where we also
    // omit the flush (see webKitMediaSrcFlush) we actually emit the updated, correct segment.
    if (streamingMembers->doesNeedSegmentEvent) {
        GST_DEBUG_OBJECT(pad, "Need new SEGMENT event, pushing it: %" GST_SEGMENT_FORMAT, &streamingMembers->segment);
        bool result = gst_pad_push_event(pad, gst_event_new_segment(&streamingMembers->segment));
        GST_DEBUG_OBJECT(pad, "SEGMENT event pushed, result = %s.", boolForPrinting(result));
        ASSERT(result);
        streamingMembers->doesNeedSegmentEvent = false;
    }

    if (GST_IS_SAMPLE(object.get())) {
        GRefPtr<GstSample> sample = adoptGRef(GST_SAMPLE(object.leakRef()));

        if (!gst_caps_is_equal(gst_sample_get_caps(sample.get()), streamingMembers->previousCaps.get())) {
            // This sample needs new caps (typically because of a quality change).
            GST_DEBUG_OBJECT(pad, "Pushing new CAPS event: %" GST_PTR_FORMAT, gst_sample_get_caps(sample.get()));
            bool result = gst_pad_push_event(stream->pad.get(), gst_event_new_caps(gst_sample_get_caps(sample.get())));
            GST_DEBUG_OBJECT(pad, "CAPS event pushed, result = %s.", boolForPrinting(result));
            ASSERT(result);
            streamingMembers->previousCaps = gst_sample_get_caps(sample.get());
        }

        GRefPtr<GstBuffer> buffer = gst_sample_get_buffer(sample.get());
        sample.clear();

        bool pushingFirstBuffer = !streamingMembers->hasPushedFirstBuffer;
        if (pushingFirstBuffer) {
            GST_DEBUG_OBJECT(pad, "Sending first buffer on this pad.");
            GUniquePtr<char> fileName { g_strdup_printf("playback-pipeline-before-playback-%s", stream->track->trackId().string().utf8().data()) };
            GST_DEBUG_BIN_TO_DOT_FILE_WITH_TS(GST_BIN(findPipeline(GRefPtr<GstElement>(GST_ELEMENT(stream->source))).get()),
                GST_DEBUG_GRAPH_SHOW_ALL, fileName.get());
            streamingMembers->hasPushedFirstBuffer = true;
        }

        // Push the buffer without the streamingMembers lock so that flushes can happen while it travels downstream.
        streamingMembers.unlockEarly();

        ASSERT(GST_BUFFER_PTS_IS_VALID(buffer.get()));
        GST_TRACE_OBJECT(pad, "Pushing buffer downstream: %" GST_PTR_FORMAT, buffer.get());
        GstFlowReturn result = gst_pad_push(pad, buffer.leakRef());
        if (result != GST_FLOW_OK && result != GST_FLOW_FLUSHING) {
            GST_ERROR_OBJECT(pad, "Pushing buffer returned %s", gst_flow_get_name(result));
            GUniquePtr<char> fileName { g_strdup_printf("playback-pipeline-pushing-buffer-failed-%s", stream->track->trackId().string().utf8().data()) };
            GST_DEBUG_BIN_TO_DOT_FILE_WITH_TS(GST_BIN(findPipeline(GRefPtr<GstElement>(GST_ELEMENT(stream->source))).get()),
                GST_DEBUG_GRAPH_SHOW_ALL, fileName.get());
            gst_pad_pause_task(pad);
        }
    } else if (GST_IS_EVENT(object.get())) {
        // EOS events and other enqueued events are also sent unlocked so they can react to flushes if necessary.
        GRefPtr<GstEvent> event = GRefPtr<GstEvent>(GST_EVENT(object.leakRef()));

        streamingMembers.unlockEarly();
        GST_DEBUG_OBJECT(pad, "Pushing event downstream: %" GST_PTR_FORMAT, event.get());
        bool eventHandled = gst_pad_push_event(pad, GRefPtr<GstEvent>(event).leakRef());
        if (!eventHandled)
            GST_DEBUG_OBJECT(pad, "Pushed event was not handled: %" GST_PTR_FORMAT, event.get());
    } else
        ASSERT_NOT_REACHED();
}

static void webKitMediaSrcStreamFlush(Stream* stream, bool isSeekingFlush)
{
    ASSERT(isMainThread());
    bool skipFlush = false;

    {
        DataMutexLocker streamingMembers { stream->streamingMembersDataMutex };

        if (!streamingMembers->hasPoppedFirstObject) {
            GST_DEBUG_OBJECT(stream->source, "Flush request for stream '%s' occurred before hasPoppedFirstObject, just clearing the queue and readjusting the segment.", stream->track->trackId().string().utf8().data());
            DataMutexLocker queue { stream->track->queueDataMutex() };
            // We use clear() instead of flush() because the WebKitMediaSrc streaming thread could be waiting
            // for the queue. flush() would cancel the notEmptyCallback therefore leaving the streaming thread
            // stuck waiting forever.
            queue->clear();
            skipFlush = true;
        }
    }

    if (!skipFlush) {
        // Signal the loop() function to stop waiting for any condition variable, pause the task and return,
        // which will keeping the streaming thread idle.
        GST_DEBUG_OBJECT(stream->pad.get(), "Taking the StreamingMembers mutex and setting isFlushing = true.");
        {
            DataMutexLocker streamingMembers { stream->streamingMembersDataMutex };
            DataMutexLocker queue { stream->track->queueDataMutex() };

            streamingMembers->isFlushing = true;
            queue->flush(); // Clear the queue and cancel any waiting callback.

            streamingMembers->queueChangedOrFlushedCondition.notifyAll();
            streamingMembers->padLinkedOrFlushedCondition.notifyAll();
        }

        // Flush downstream. This will stop processing in downstream elements and if the streaming thread was in a
        // downstream chain() function, it will quickly return to the loop() function, which thanks to the
        // previous section will also quickly end.
        GST_DEBUG_OBJECT(stream->pad.get(), "Sending FLUSH_START downstream.");
        gst_pad_push_event(stream->pad.get(), gst_event_new_flush_start());
        GST_DEBUG_OBJECT(stream->pad.get(), "FLUSH_START sent.");
    }

    // Adjust segment. This is different for seeks and non-seeking flushes.
    if (isSeekingFlush) {
        // In the case of seeking flush we are resetting the timeline (see the flush stop later).
        // The resulting segment is brand new, but with a different start time.
        WebKitMediaSrcPrivate* priv = stream->source->priv;
        DataMutexLocker streamingMembers { stream->streamingMembersDataMutex };
        streamingMembers->segment.base = 0;
        streamingMembers->segment.rate = priv->rate;
        streamingMembers->segment.start = streamingMembers->segment.time = priv->startTime;
    } else {
        // In the case of non-seeking flushes we don't reset the timeline, so instead we need to increase the `base` field
        // by however running time we're starting after the flush.

        GstClockTime pipelineStreamTime;
        gst_element_query_position(findPipeline(GRefPtr<GstElement>(GST_ELEMENT(stream->source))).get(), GST_FORMAT_TIME,
            reinterpret_cast<gint64*>(&pipelineStreamTime));
        // GST_CLOCK_TIME_NONE is returned when the pipeline is not yet pre-rolled (e.g. just after a seek). In this case
        // we don't need to adjust the segment though, as running time has not advanced.
        if (GST_CLOCK_TIME_IS_VALID(pipelineStreamTime)) {
            DataMutexLocker streamingMembers { stream->streamingMembersDataMutex };
            // We need to increase the base by the running time accumulated during the previous segment.

            GstClockTime pipelineRunningTime = gst_segment_to_running_time(&streamingMembers->segment, GST_FORMAT_TIME, pipelineStreamTime);
            assert(GST_CLOCK_TIME_IS_VALID(pipelineRunningTime));
            GST_DEBUG_OBJECT(stream->source, "Resetting segment to current pipeline running time (%" GST_TIME_FORMAT") and stream time (%" GST_TIME_FORMAT ")",
                GST_TIME_ARGS(pipelineRunningTime), GST_TIME_ARGS(pipelineStreamTime));
            streamingMembers->segment.base = pipelineRunningTime;

            streamingMembers->segment.start = streamingMembers->segment.time = static_cast<GstClockTime>(pipelineStreamTime);
        }
    }

    if (!skipFlush) {
        // By taking the stream lock we are waiting for the streaming thread task to stop if it hadn't yet.
        GST_DEBUG_OBJECT(stream->pad.get(), "Taking the STREAM_LOCK.");
        auto streamLock = GstPadStreamLocker(stream->pad.get());
        {
            GST_DEBUG_OBJECT(stream->pad.get(), "Taking the StreamingMembers mutex again.");
            DataMutexLocker streamingMembers { stream->streamingMembersDataMutex };
            GST_DEBUG_OBJECT(stream->pad.get(), "StreamingMembers mutex taken, using it to set isFlushing = false.");
            streamingMembers->isFlushing = false;
            streamingMembers->doesNeedSegmentEvent = true;
        }

        GST_DEBUG_OBJECT(stream->pad.get(), "Sending FLUSH_STOP downstream (resetTime = %s).", boolForPrinting(isSeekingFlush));
        // Since FLUSH_STOP is a synchronized event, we send it while we still hold the stream lock of the pad.
        gst_pad_push_event(stream->pad.get(), gst_event_new_flush_stop(isSeekingFlush));
        GST_DEBUG_OBJECT(stream->pad.get(), "FLUSH_STOP sent.");

        GST_DEBUG_OBJECT(stream->pad.get(), "Starting webKitMediaSrcLoop task and releasing the STREAM_LOCK.");
        gst_pad_start_task(stream->pad.get(), webKitMediaSrcLoop, stream->pad.get(), nullptr);
    }
}

void webKitMediaSrcFlush(WebKitMediaSrc* source, const AtomString& streamName)
{
    ASSERT(isMainThread());
    GST_DEBUG_OBJECT(source, "Received non-seek flush request for stream '%s'.", streamName.string().utf8().data());
    Stream* stream = source->priv->streamByName(streamName);

    webKitMediaSrcStreamFlush(stream, false);
}

static void webKitMediaSrcSeek(WebKitMediaSrc* source, uint64_t startTime, double rate)
{
    ASSERT(isMainThread());
    source->priv->startTime = startTime;
    source->priv->rate = rate;
    GST_DEBUG_OBJECT(source, "Seek requested to time %" GST_TIME_FORMAT " with rate %f.", GST_TIME_ARGS(startTime), rate);

    for (const RefPtr<Stream>& stream : source->priv->streams.values())
        webKitMediaSrcStreamFlush(stream.get(), true);
}

static int countStreamsOfType(WebKitMediaSrc* source, WebCore::TrackPrivateBaseGStreamer::TrackType type)
{
    // Barring pipeline dumps someone may add during debugging, WebKit will only read these properties (n-video etc.) from the main thread.
    return std::count_if(source->priv->streams.begin(), source->priv->streams.end(), [type](auto item) {
        return item.value->track->type() == type;
    });
}

static void webKitMediaSrcGetProperty(GObject* object, unsigned propId, GValue* value, GParamSpec* pspec)
{
    WebKitMediaSrc* source = WEBKIT_MEDIA_SRC(object);

    switch (propId) {
    case PROP_N_AUDIO:
        g_value_set_int(value, countStreamsOfType(source, WebCore::TrackPrivateBaseGStreamer::TrackType::Audio));
        break;
    case PROP_N_VIDEO:
        g_value_set_int(value, countStreamsOfType(source, WebCore::TrackPrivateBaseGStreamer::TrackType::Video));
        break;
    case PROP_N_TEXT:
        g_value_set_int(value, countStreamsOfType(source, WebCore::TrackPrivateBaseGStreamer::TrackType::Text));
        break;
    default:
        G_OBJECT_WARN_INVALID_PROPERTY_ID(object, propId, pspec);
    }
}

static GstStateChangeReturn webKitMediaSrcChangeState(GstElement* element, GstStateChange transition)
{
    WebKitMediaSrc* source = WEBKIT_MEDIA_SRC(element);
    switch (transition) {
    case GST_STATE_CHANGE_PAUSED_TO_READY:
        GST_DEBUG_OBJECT(source, "Downgrading to READY state, tearing down all streams...");
        while (!source->priv->streams.isEmpty())
            webKitMediaSrcTearDownStream(source, source->priv->streams.begin()->key);
        break;
    case GST_STATE_CHANGE_READY_TO_PAUSED:
        if (source->priv->isStarted()) {
            GST_FIXME_OBJECT(source, "Resuming state from READY -> PAUSED after a downgrade is not implemented. Expect failure.");
        }
        break;
    default:
        break;
    }
    return GST_ELEMENT_CLASS(webkit_media_src_parent_class)->change_state(element, transition);
}

static gboolean webKitMediaSrcSendEvent(GstElement* element, GstEvent* event)
{
    switch (GST_EVENT_TYPE(event)) {
    case GST_EVENT_SEEK: {
        double rate;
        GstFormat format;
        GstSeekType startType;
        int64_t start;
        gst_event_parse_seek(event, &rate, &format, nullptr, &startType, &start, nullptr, nullptr);
        if (format != GST_FORMAT_TIME || startType != GST_SEEK_TYPE_SET) {
            GST_ERROR_OBJECT(element, "Rejecting unsupported seek event: %" GST_PTR_FORMAT, event);
            return false;
        }
        GST_DEBUG_OBJECT(element, "Handling seek event: %" GST_PTR_FORMAT, event);
        webKitMediaSrcSeek(WEBKIT_MEDIA_SRC(element), start, rate);
        return true;
    }
    default:
        GST_DEBUG_OBJECT(element, "Rejecting unsupported event: %" GST_PTR_FORMAT, event);
        return false;
    }
}

// URI handler interface. It's only purpose is for the element to be instantiated by playbin on "mediasourceblob:"
// URIs. The actual URI does not matter.
static GstURIType webKitMediaSrcUriGetType(GType)
{
    return GST_URI_SRC;
}

static const gchar* const* webKitMediaSrcGetProtocols(GType)
{
    static const char* protocols[] = {"mediasourceblob", nullptr };
    return protocols;
}

static gchar* webKitMediaSrcGetUri(GstURIHandler* handler)
{
    WebKitMediaSrc* source = WEBKIT_MEDIA_SRC(handler);

    auto locker = GstObjectLocker(source);
    return g_strdup(source->priv->uri.get());
}

static gboolean webKitMediaSrcSetUri(GstURIHandler* handler, const gchar* uri, GError**)
{
    WebKitMediaSrc* source = WEBKIT_MEDIA_SRC(handler);

    if (GST_STATE(source) >= GST_STATE_PAUSED) {
        GST_ERROR_OBJECT(source, "URI can only be set in states < PAUSED");
        return false;
    }

    auto locker = GstObjectLocker(source);
    source->priv->uri = GUniquePtr<char>(g_strdup(uri));
    return TRUE;
}

static void webKitMediaSrcUriHandlerInit(void* gIface, void*)
{
    GstURIHandlerInterface* iface = (GstURIHandlerInterface *) gIface;

    iface->get_type = webKitMediaSrcUriGetType;
    iface->get_protocols = webKitMediaSrcGetProtocols;
    iface->get_uri = webKitMediaSrcGetUri;
    iface->set_uri = webKitMediaSrcSetUri;
}

namespace WTF {
template <> GRefPtr<WebKitMediaSrc> adoptGRef(WebKitMediaSrc* ptr)
{
    ASSERT(!ptr || !g_object_is_floating(G_OBJECT(ptr)));
    return GRefPtr<WebKitMediaSrc>(ptr, GRefPtrAdopt);
}

template <> WebKitMediaSrc* refGPtr<WebKitMediaSrc>(WebKitMediaSrc* ptr)
{
    if (ptr)
        gst_object_ref_sink(GST_OBJECT(ptr));

    return ptr;
}

template <> void derefGPtr<WebKitMediaSrc>(WebKitMediaSrc* ptr)
{
    if (ptr)
        gst_object_unref(ptr);
}
} // namespace WTF

#endif // USE(GSTREAMER)