/*
 *  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 Metrological Group B.V.
 *  Copyright (C) 2015, 2016, 2018, 2019 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 "VideoTrackPrivateGStreamer.h"

#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

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)
    {
        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() };

    // Every time a track is added or removed this collection is swapped by an updated one and a STREAM_COLLECTION
    // message is posted in the bus.
    GRefPtr<GstStreamCollection> collection { adoptGRef(gst_stream_collection_new("WebKitMediaSrc")) };

    // 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 webKitMediaSrcFinalize(GObject*);
static GstStateChangeReturn webKitMediaSrcChangeState(GstElement*, GstStateChange);
static gboolean webKitMediaSrcActivateMode(GstPad*, GstObject*, GstPadMode, gboolean activate);
static void webKitMediaSrcLoop(void*);
static void webKitMediaSrcStreamFlushStart(const RefPtr<Stream>&);
static void webKitMediaSrcStreamFlushStop(const RefPtr<Stream>&, bool resetTime);
static void webKitMediaSrcGetProperty(GObject*, unsigned propId, GValue*, GParamSpec*);

#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*)
{
}

G_DEFINE_TYPE_WITH_CODE(WebKitMediaSrc, webkit_media_src, GST_TYPE_ELEMENT,
    G_IMPLEMENT_INTERFACE(GST_TYPE_URI_HANDLER, webKitMediaSrcUriHandlerInit);
    G_ADD_PRIVATE(WebKitMediaSrc);
    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, const AtomString& name, WebCore::MediaSourceStreamTypeGStreamer type, GRefPtr<GstCaps>&& initialCaps, GRefPtr<GstStream>&& streamInfo)
        : source(source)
        , pad(WTFMove(pad))
        , name(name)
        , type(type)
        , streamInfo(WTFMove(streamInfo))
        , streamingMembersDataMutex(WTFMove(initialCaps), source->priv->startTime, source->priv->rate, adoptGRef(gst_event_new_stream_collection(source->priv->collection.get())))
    { }

    WebKitMediaSrc* const source;
    GRefPtr<GstPad> const pad;
    AtomString const name;
    WebCore::MediaSourceStreamTypeGStreamer type;
    GRefPtr<GstStream> streamInfo;

    // The point of having a queue in WebKitMediaSource is to limit the number of context switches per second.
    // If we had no queue, the main thread would have to be awaken for every frame. On the other hand, if the
    // queue had unlimited size WebKit would end up requesting flushes more often than necessary when frames
    // in the future are re-appended. As a sweet spot between these extremes we choose to allow enqueueing a
    // few seconds worth of samples.

    // `isReadyForMoreSamples` follows the classical two water levels strategy: initially it's true until the
    // high water level is reached, then it becomes false until the queue drains down to the low water level
    // and the cycle repeats. This way we avoid stalls and minimize context switches.

    static const uint64_t durationEnqueuedHighWaterLevel = 5 * GST_SECOND;
    static const uint64_t durationEnqueuedLowWaterLevel = 2 * GST_SECOND;

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

            GstStreamCollection* collection;
            gst_event_parse_stream_collection(this->pendingStreamCollectionEvent.get(), &collection);
            ASSERT(collection);
        }

        bool hasPushedFirstBuffer { false };
        bool wasStreamStartSent { false };
        bool doesNeedSegmentEvent { true };
        GstSegment segment;
        GRefPtr<GstEvent> pendingStreamCollectionEvent;
        GRefPtr<GstCaps> pendingInitialCaps;
        GRefPtr<GstCaps> previousCaps;

        Condition padLinkedOrFlushedCondition;
        Condition queueChangedOrFlushedCondition;
        Deque<GRefPtr<GstMiniObject>> queue;
        bool isFlushing { false };
        bool doesNeedToNotifyOnLowWaterLevel { false };

        uint64_t durationEnqueued() const
        {
            // Find the first and last GstSample in the queue and subtract their DTS.
            auto frontIter = std::find_if(queue.begin(), queue.end(), [](const GRefPtr<GstMiniObject>& object) {
                return GST_IS_SAMPLE(object.get());
            });

            // If there are no samples in the queue, that makes total duration of enqueued frames of zero.
            if (frontIter == queue.end())
                return 0;

            auto backIter = std::find_if(queue.rbegin(), queue.rend(), [](const GRefPtr<GstMiniObject>& object) {
                return GST_IS_SAMPLE(object.get());
            });

            const GstBuffer* front = gst_sample_get_buffer(GST_SAMPLE(frontIter->get()));
            const GstBuffer* back = gst_sample_get_buffer(GST_SAMPLE(backIter->get()));
            return GST_BUFFER_DTS_OR_PTS(back) - GST_BUFFER_DTS_OR_PTS(front);
        }
    };
    DataMutex<StreamingMembers> streamingMembersDataMutex;

    struct ReportedStatus {
        // Set to true when the pad is removed. In the case where a reference to the Stream object is alive because of
        // a posted task to notify isReadyForMoreSamples, the notification must not be delivered if this flag is true.
        bool wasRemoved { false };

        bool isReadyForMoreSamples { true };
        SourceBufferPrivateClient* sourceBufferPrivateToNotify { nullptr };
    };
    MainThreadData<ReportedStatus> reportedStatus;
};

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 void webkit_media_src_class_init(WebKitMediaSrcClass* klass)
{
    GObjectClass* oklass = G_OBJECT_CLASS(klass);
    GstElementClass* eklass = GST_ELEMENT_CLASS(klass);

    oklass->finalize = webKitMediaSrcFinalize;
    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;

    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 webkit_media_src_init(WebKitMediaSrc* source)
{
    ASSERT(isMainThread());

    GST_OBJECT_FLAG_SET(source, GST_ELEMENT_FLAG_SOURCE);
    source->priv = G_TYPE_INSTANCE_GET_PRIVATE((source), WEBKIT_TYPE_MEDIA_SRC, WebKitMediaSrcPrivate);
    new (source->priv) WebKitMediaSrcPrivate();
}

static void webKitMediaSrcFinalize(GObject* object)
{
    ASSERT(isMainThread());

    WebKitMediaSrc* source = WEBKIT_MEDIA_SRC(object);
    source->priv->~WebKitMediaSrcPrivate();
    GST_CALL_PARENT(G_OBJECT_CLASS, finalize, (object));
}

static GstPadProbeReturn debugProbe(GstPad* pad, GstPadProbeInfo* info, void*)
{
    RefPtr<Stream>& stream = WEBKIT_MEDIA_SRC_PAD(pad)->priv->stream;
    GST_TRACE_OBJECT(stream->source, "track %s: %" GST_PTR_FORMAT, stream->name.string().utf8().data(), info->data);
    return GST_PAD_PROBE_OK;
}

// GstStreamCollection are immutable objects once posted. THEY MUST NOT BE MODIFIED once they have been posted.
// Instead, when stream changes occur a new collection must be made. The following functions help to create
// such new collections:

static GRefPtr<GstStreamCollection> copyCollectionAndAddStream(GstStreamCollection* collection, GRefPtr<GstStream>&& stream)
{
    GRefPtr<GstStreamCollection> newCollection = adoptGRef(gst_stream_collection_new(collection->upstream_id));

    unsigned n = gst_stream_collection_get_size(collection);
    for (unsigned i = 0; i < n; i++)
        gst_stream_collection_add_stream(newCollection.get(), static_cast<GstStream*>(gst_object_ref(gst_stream_collection_get_stream(collection, i))));
    gst_stream_collection_add_stream(newCollection.get(), stream.leakRef());

    return newCollection;
}

static GRefPtr<GstStreamCollection> copyCollectionWithoutStream(GstStreamCollection* collection, const GstStream* stream)
{
    GRefPtr<GstStreamCollection> newCollection = adoptGRef(gst_stream_collection_new(collection->upstream_id));

    unsigned n = gst_stream_collection_get_size(collection);
    for (unsigned i = 0; i < n; i++) {
        GRefPtr<GstStream> oldStream = gst_stream_collection_get_stream(collection, i);
        if (oldStream.get() != stream)
            gst_stream_collection_add_stream(newCollection.get(), oldStream.leakRef());
    }

    return newCollection;
}

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

void webKitMediaSrcAddStream(WebKitMediaSrc* source, const AtomString& name, WebCore::MediaSourceStreamTypeGStreamer type, GRefPtr<GstCaps>&& initialCaps)
{
    ASSERT(isMainThread());
    ASSERT(!source->priv->streams.contains(name));

    GRefPtr<GstStream> streamInfo = adoptGRef(gst_stream_new(name.string().utf8().data(), initialCaps.get(), gstStreamType(type), GST_STREAM_FLAG_SELECT));
    source->priv->collection = copyCollectionAndAddStream(source->priv->collection.get(), GRefPtr<GstStream>(streamInfo));
    gst_element_post_message(GST_ELEMENT(source), gst_message_new_stream_collection(GST_OBJECT(source), source->priv->collection.get()));

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

    {
        RefPtr<Stream> stream = adoptRef(new Stream(source, GRefPtr<GstPad>(GST_PAD(pad.get())), name, type, WTFMove(initialCaps), WTFMove(streamInfo)));
        pad->priv->stream = stream;
        source->priv->streams.set(name, WTFMove(stream));
    }

    if (gst_debug_category_get_threshold(webkit_media_src_debug) >= GST_LEVEL_TRACE)
        gst_pad_add_probe(GST_PAD(pad.get()), static_cast<GstPadProbeType>(GST_PAD_PROBE_TYPE_DATA_DOWNSTREAM | GST_PAD_PROBE_TYPE_EVENT_FLUSH), debugProbe, nullptr, nullptr);

    // Workaround: gst_element_add_pad() should already call gst_pad_set_active() if the element is PAUSED or
    // PLAYING. Unfortunately, as of GStreamer 1.14.4 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.
    GstState state;
    gst_element_get_state(GST_ELEMENT(source), &state, nullptr, 0);
    if (state > GST_STATE_READY)
        gst_pad_set_active(GST_PAD(pad.get()), true);

    gst_element_add_pad(GST_ELEMENT(source), GST_PAD(pad.get()));
}

void webKitMediaSrcRemoveStream(WebKitMediaSrc* source, const AtomString& name)
{
    ASSERT(isMainThread());
    Stream* stream = source->priv->streamByName(name);

    source->priv->collection = copyCollectionWithoutStream(source->priv->collection.get(), stream->streamInfo.get());
    gst_element_post_message(GST_ELEMENT(source), gst_message_new_stream_collection(GST_OBJECT(source), source->priv->collection.get()));

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

    stream->reportedStatus->wasRemoved = true;
    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;
        {
            DataMutex<Stream::StreamingMembers>::LockedWrapper 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);
        {
            DataMutex<Stream::StreamingMembers>::LockedWrapper 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;
    DataMutex<Stream::StreamingMembers>::LockedWrapper streamingMembers(stream->streamingMembersDataMutex);
    streamingMembers->padLinkedOrFlushedCondition.notifyOne();
}

static void webKitMediaSrcStreamNotifyLowWaterLevel(const RefPtr<Stream>& stream)
{
    RunLoop::main().dispatch([stream]() {
        if (stream->reportedStatus->wasRemoved)
            return;

        stream->reportedStatus->isReadyForMoreSamples = true;
        if (stream->reportedStatus->sourceBufferPrivateToNotify) {
            // We need to set sourceBufferPrivateToNotify BEFORE calling sourceBufferPrivateDidBecomeReadyForMoreSamples(),
            // not after, since otherwise it would destroy a notification request should the callback request one.
            SourceBufferPrivateClient* sourceBuffer = stream->reportedStatus->sourceBufferPrivateToNotify;
            stream->reportedStatus->sourceBufferPrivateToNotify = nullptr;
            sourceBuffer->sourceBufferPrivateDidBecomeReadyForMoreSamples(stream->name);
        }
    });
}

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

    DataMutex<Stream::StreamingMembers>::LockedWrapper 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.
    GST_OBJECT_LOCK(pad);
    if (!GST_PAD_IS_LINKED(pad)) {
        g_signal_connect(pad, "linked", G_CALLBACK(webKitMediaSrcPadLinked), nullptr);
        GST_OBJECT_UNLOCK(pad);

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

        g_signal_handlers_disconnect_by_func(pad, reinterpret_cast<void*>(webKitMediaSrcPadLinked), nullptr);
        if (streamingMembers->isFlushing)
            return;
    } else
        GST_OBJECT_UNLOCK(pad);
    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->pendingStreamCollectionEvent)
        gst_pad_push_event(stream->pad.get(), streamingMembers->pendingStreamCollectionEvent.leakRef());

    if (!streamingMembers->wasStreamStartSent) {
        GUniquePtr<char> streamId(g_strdup_printf("mse/%s", stream->name.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());

        bool wasStreamStartSent = gst_pad_push_event(pad, event.leakRef());
        streamingMembers->wasStreamStartSent = wasStreamStartSent;
    }

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

        gst_pad_push_event(pad, event.leakRef());

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

    streamingMembers->queueChangedOrFlushedCondition.wait(streamingMembers.mutex(), [&]() {
        return !streamingMembers->queue.isEmpty() || streamingMembers->isFlushing;
    });
    if (streamingMembers->isFlushing)
        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_pad_push_event(pad, gst_event_new_segment(&streamingMembers->segment));
        streamingMembers->doesNeedSegmentEvent = false;
    }

    GRefPtr<GstMiniObject> object = streamingMembers->queue.takeFirst();
    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_pad_push_event(stream->pad.get(), gst_event_new_caps(gst_sample_get_caps(sample.get())));
            streamingMembers->previousCaps = gst_sample_get_caps(sample.get());
        }

        if (streamingMembers->doesNeedToNotifyOnLowWaterLevel && streamingMembers->durationEnqueued() <= Stream::durationEnqueuedLowWaterLevel) {
            streamingMembers->doesNeedToNotifyOnLowWaterLevel = false;
            webKitMediaSrcStreamNotifyLowWaterLevel(RefPtr<Stream>(stream));
        }

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

        if (!streamingMembers->hasPushedFirstBuffer) {
            GUniquePtr<char> fileName { g_strdup_printf("playback-pipeline-before-playback-%s", stream->name.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.lockHolder().unlockEarly();

        ASSERT(GST_BUFFER_PTS_IS_VALID(buffer.get()));
        GstFlowReturn ret = gst_pad_push(pad, buffer.leakRef());
        if (ret != GST_FLOW_OK && ret != GST_FLOW_FLUSHING) {
            GST_ERROR_OBJECT(pad, "Pushing buffer returned %s", gst_flow_get_name(ret));
            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.lockHolder().unlockEarly();
        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 webKitMediaSrcEnqueueObject(WebKitMediaSrc* source, const AtomString& streamName, GRefPtr<GstMiniObject>&& object)
{
    ASSERT(isMainThread());
    ASSERT(object);

    Stream* stream = source->priv->streamByName(streamName);
    DataMutex<Stream::StreamingMembers>::LockedWrapper streamingMembers(stream->streamingMembersDataMutex);
    streamingMembers->queue.append(WTFMove(object));
    if (stream->reportedStatus->isReadyForMoreSamples && streamingMembers->durationEnqueued() > Stream::durationEnqueuedHighWaterLevel) {
        stream->reportedStatus->isReadyForMoreSamples = false;
        streamingMembers->doesNeedToNotifyOnLowWaterLevel = true;
    }
    streamingMembers->queueChangedOrFlushedCondition.notifyOne();
}

void webKitMediaSrcEnqueueSample(WebKitMediaSrc* source, const AtomString& streamName, GRefPtr<GstSample>&& sample)
{
    ASSERT(GST_BUFFER_PTS_IS_VALID(gst_sample_get_buffer(sample.get())));
    webKitMediaSrcEnqueueObject(source, streamName, adoptGRef(GST_MINI_OBJECT(sample.leakRef())));
}

static void webKitMediaSrcEnqueueEvent(WebKitMediaSrc* source, const AtomString& streamName, GRefPtr<GstEvent>&& event)
{
    webKitMediaSrcEnqueueObject(source, streamName, adoptGRef(GST_MINI_OBJECT(event.leakRef())));
}

void webKitMediaSrcEndOfStream(WebKitMediaSrc* source, const AtomString& streamName)
{
    webKitMediaSrcEnqueueEvent(source, streamName, adoptGRef(gst_event_new_eos()));
}

bool webKitMediaSrcIsReadyForMoreSamples(WebKitMediaSrc* source, const AtomString& streamName)
{
    ASSERT(isMainThread());
    Stream* stream = source->priv->streamByName(streamName);
    return stream->reportedStatus->isReadyForMoreSamples;
}

void webKitMediaSrcNotifyWhenReadyForMoreSamples(WebKitMediaSrc* source, const AtomString& streamName, WebCore::SourceBufferPrivateClient* sourceBufferPrivate)
{
    ASSERT(isMainThread());
    Stream* stream = source->priv->streamByName(streamName);
    ASSERT(!stream->reportedStatus->isReadyForMoreSamples);
    stream->reportedStatus->sourceBufferPrivateToNotify = sourceBufferPrivate;
}

static GstStateChangeReturn webKitMediaSrcChangeState(GstElement* element, GstStateChange transition)
{
    WebKitMediaSrc* source = WEBKIT_MEDIA_SRC(element);
    if (transition == GST_STATE_CHANGE_PAUSED_TO_READY) {
        while (!source->priv->streams.isEmpty())
            webKitMediaSrcRemoveStream(source, source->priv->streams.begin()->key);
    }
    return GST_ELEMENT_CLASS(webkit_media_src_parent_class)->change_state(element, transition);
}

static void webKitMediaSrcStreamFlushStart(const RefPtr<Stream>& stream)
{
    ASSERT(isMainThread());
    {
        DataMutex<Stream::StreamingMembers>::LockedWrapper streamingMembers(stream->streamingMembersDataMutex);

        streamingMembers->isFlushing = true;
        streamingMembers->queueChangedOrFlushedCondition.notifyOne();
        streamingMembers->padLinkedOrFlushedCondition.notifyOne();
    }

    gst_pad_push_event(stream->pad.get(), gst_event_new_flush_start());
}

static void webKitMediaSrcStreamFlushStop(const RefPtr<Stream>& stream, bool resetTime)
{
    ASSERT(isMainThread());

    // By taking the stream lock we are waiting for the streaming thread task to stop if it hadn't yet.
    GST_PAD_STREAM_LOCK(stream->pad.get());
    {
        DataMutex<Stream::StreamingMembers>::LockedWrapper streamingMembers(stream->streamingMembersDataMutex);

        streamingMembers->isFlushing = false;
        streamingMembers->doesNeedSegmentEvent = true;
        streamingMembers->queue.clear();
        if (streamingMembers->doesNeedToNotifyOnLowWaterLevel) {
            streamingMembers->doesNeedToNotifyOnLowWaterLevel = false;
            webKitMediaSrcStreamNotifyLowWaterLevel(stream);
        }
    }

    // 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(resetTime));

    gst_pad_start_task(stream->pad.get(), webKitMediaSrcLoop, stream->pad.get(), nullptr);
    GST_PAD_STREAM_UNLOCK(stream->pad.get());
}

void webKitMediaSrcFlush(WebKitMediaSrc* source, const AtomString& streamName)
{
    ASSERT(isMainThread());
    Stream* stream = source->priv->streamByName(streamName);

    bool hasPushedFirstBuffer;
    {
        DataMutex<Stream::StreamingMembers>::LockedWrapper streamingMembers(stream->streamingMembersDataMutex);
        hasPushedFirstBuffer = streamingMembers->hasPushedFirstBuffer;
    }

    if (hasPushedFirstBuffer) {
        // If no buffer has been pushed there is no need for flush... and flushing at that point could
        // expose bugs in downstream which may have not completely initialized (e.g. decodebin3 not
        // having linked the chain so far and forgetting to do it after the flush).
        webKitMediaSrcStreamFlushStart(stream);
    }

    GstClockTime pipelineStreamTime;
    gst_element_query_position(findPipeline(GRefPtr<GstElement>(GST_ELEMENT(source))).get(), GST_FORMAT_TIME,
        reinterpret_cast<gint64*>(&pipelineStreamTime));
    // -1 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)) {
        DataMutex<Stream::StreamingMembers>::LockedWrapper 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));
        streamingMembers->segment.base = pipelineRunningTime;

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

    if (hasPushedFirstBuffer)
        webKitMediaSrcStreamFlushStop(stream, false);
}

void webKitMediaSrcSeek(WebKitMediaSrc* source, uint64_t startTime, double rate)
{
    ASSERT(isMainThread());
    source->priv->startTime = startTime;
    source->priv->rate = rate;

    for (auto& pair : source->priv->streams) {
        const RefPtr<Stream>& stream = pair.value;
        bool hasPushedFirstBuffer;
        {
            DataMutex<Stream::StreamingMembers>::LockedWrapper streamingMembers(stream->streamingMembersDataMutex);
            hasPushedFirstBuffer = streamingMembers->hasPushedFirstBuffer;
        }

        if (hasPushedFirstBuffer) {
            // If no buffer has been pushed there is no need for flush... and flushing at that point could
            // expose bugs in downstream which may have not completely initialized (e.g. decodebin3 not
            // having linked the chain so far and forgetting to do it after the flush).
            webKitMediaSrcStreamFlushStart(stream);
        }

        {
            DataMutex<Stream::StreamingMembers>::LockedWrapper streamingMembers(stream->streamingMembersDataMutex);
            streamingMembers->segment.base = 0;
            streamingMembers->segment.rate = rate;
            streamingMembers->segment.start = streamingMembers->segment.time = startTime;
        }

        if (hasPushedFirstBuffer)
            webKitMediaSrcStreamFlushStop(stream, true);
    }
}

static int countStreamsOfType(WebKitMediaSrc* source, WebCore::MediaSourceStreamTypeGStreamer 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->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::MediaSourceStreamTypeGStreamer::Audio));
        break;
    case PROP_N_VIDEO:
        g_value_set_int(value, countStreamsOfType(source, WebCore::MediaSourceStreamTypeGStreamer::Video));
        break;
    case PROP_N_TEXT:
        g_value_set_int(value, countStreamsOfType(source, WebCore::MediaSourceStreamTypeGStreamer::Text));
        break;
    default:
        G_OBJECT_WARN_INVALID_PROPERTY_ID(object, propId, pspec);
    }
}

// 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);
    gchar* result;

    GST_OBJECT_LOCK(source);
    result = g_strdup(source->priv->uri.get());
    GST_OBJECT_UNLOCK(source);
    return result;
}

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;
    }

    GST_OBJECT_LOCK(source);
    source->priv->uri = GUniquePtr<char>(g_strdup(uri));
    GST_OBJECT_UNLOCK(source);
    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)