1 files changed, 160 insertions, 114 deletions

diff --git a/Source/WebCore/platform/graphics/gstreamer/WebKitWebSourceGStreamer.cpp b/Source/WebCore/platform/graphics/gstreamer/WebKitWebSourceGStreamer.cpp
index d2ae28f2c..df7c46836 100644
--- a/Source/WebCore/platform/graphics/gstreamer/WebKitWebSourceGStreamer.cpp
+++ b/Source/WebCore/platform/graphics/gstreamer/WebKitWebSourceGStreamer.cpp
@@ -39,6 +39,8 @@
 #include <gst/app/gstappsrc.h>
 #include <gst/gst.h>
 #include <gst/pbutils/missing-plugins.h>
+#include <wtf/Condition.h>
+#include <wtf/Lock.h>
 #include <wtf/MainThread.h>
 #include <wtf/Noncopyable.h>
 #include <wtf/glib/GMutexLocker.h>
@@ -83,7 +85,7 @@ class CachedResourceStreamingClient final : public PlatformMediaResourceClient,
 class ResourceHandleStreamingClient : public ResourceHandleClient, public StreamingClient {
     WTF_MAKE_NONCOPYABLE(ResourceHandleStreamingClient); WTF_MAKE_FAST_ALLOCATED;
     public:
-        ResourceHandleStreamingClient(WebKitWebSrc*, const ResourceRequest&);
+        ResourceHandleStreamingClient(WebKitWebSrc*, ResourceRequest&&);
         virtual ~ResourceHandleStreamingClient();
 
         // StreamingClient virtual methods.
@@ -102,6 +104,12 @@ class ResourceHandleStreamingClient : public ResourceHandleClient, public Stream
         virtual void wasBlocked(ResourceHandle*);
         virtual void cannotShowURL(ResourceHandle*);
 
+        ThreadIdentifier m_thread { 0 };
+        Lock m_initializeRunLoopConditionMutex;
+        Condition m_initializeRunLoopCondition;
+        RunLoop* m_runLoop { nullptr };
+        Lock m_terminateRunLoopConditionMutex;
+        Condition m_terminateRunLoopCondition;
         RefPtr<ResourceHandle> m_resource;
 };
 
@@ -127,18 +135,19 @@ struct _WebKitWebSrcPrivate {
 
     RefPtr<PlatformMediaResourceLoader> loader;
     RefPtr<PlatformMediaResource> resource;
-    ResourceHandleStreamingClient* client;
+    std::unique_ptr<ResourceHandleStreamingClient> client;
 
     bool didPassAccessControlCheck;
 
     guint64 offset;
     guint64 size;
     gboolean seekable;
-    gboolean paused;
+    bool paused;
     bool isSeeking;
 
     guint64 requestedOffset;
 
+    bool createdInMainThread;
     MainThreadNotifier<MainThreadSourceNotification> notifier;
     GRefPtr<GstBuffer> buffer;
 };
@@ -172,57 +181,20 @@ static gboolean webKitWebSrcQueryWithParent(GstPad*, GstObject*, GstQuery*);
 
 static void webKitWebSrcNeedData(WebKitWebSrc*);
 static void webKitWebSrcEnoughData(WebKitWebSrc*);
-static void webKitWebSrcSeek(WebKitWebSrc*);
+static gboolean webKitWebSrcSeek(WebKitWebSrc*, guint64);
 
 static GstAppSrcCallbacks appsrcCallbacks = {
     // need_data
     [](GstAppSrc*, guint, gpointer userData) {
-        WebKitWebSrc* src = WEBKIT_WEB_SRC(userData);
-        WebKitWebSrcPrivate* priv = src->priv;
-
-        {
-            WTF::GMutexLocker<GMutex> locker(*GST_OBJECT_GET_LOCK(src));
-            if (!priv->paused)
-                return;
-        }
-
-        GRefPtr<WebKitWebSrc> protector = WTF::ensureGRef(src);
-        priv->notifier.notify(MainThreadSourceNotification::NeedData, [protector] { webKitWebSrcNeedData(protector.get()); });
+        webKitWebSrcNeedData(WEBKIT_WEB_SRC(userData));
     },
     // enough_data
     [](GstAppSrc*, gpointer userData) {
-        WebKitWebSrc* src = WEBKIT_WEB_SRC(userData);
-        WebKitWebSrcPrivate* priv = src->priv;
-
-        {
-            WTF::GMutexLocker<GMutex> locker(*GST_OBJECT_GET_LOCK(src));
-            if (priv->paused)
-                return;
-        }
-
-        GRefPtr<WebKitWebSrc> protector = WTF::ensureGRef(src);
-        priv->notifier.notify(MainThreadSourceNotification::EnoughData, [protector] { webKitWebSrcEnoughData(protector.get()); });
+        webKitWebSrcEnoughData(WEBKIT_WEB_SRC(userData));
     },
     // seek_data
     [](GstAppSrc*, guint64 offset, gpointer userData) -> gboolean {
-        WebKitWebSrc* src = WEBKIT_WEB_SRC(userData);
-        WebKitWebSrcPrivate* priv = src->priv;
-
-        {
-            WTF::GMutexLocker<GMutex> locker(*GST_OBJECT_GET_LOCK(src));
-            if (offset == priv->offset && priv->requestedOffset == priv->offset)
-                return TRUE;
-
-            if (!priv->seekable)
-                return FALSE;
-
-            priv->isSeeking = true;
-            priv->requestedOffset = offset;
-        }
-
-        GRefPtr<WebKitWebSrc> protector = WTF::ensureGRef(src);
-        priv->notifier.notify(MainThreadSourceNotification::Seek, [protector] { webKitWebSrcSeek(protector.get()); });
-        return TRUE;
+        return webKitWebSrcSeek(WEBKIT_WEB_SRC(userData), offset);
     },
     { nullptr }
 };
@@ -287,6 +259,8 @@ static void webkit_web_src_init(WebKitWebSrc* src)
     src->priv = priv;
     new (priv) WebKitWebSrcPrivate();
 
+    priv->createdInMainThread = isMainThread();
+
     priv->appsrc = GST_APP_SRC(gst_element_factory_make("appsrc", 0));
     if (!priv->appsrc) {
         GST_ERROR_OBJECT(src, "Failed to create appsrc");
@@ -412,28 +386,37 @@ static void webKitWebSrcStop(WebKitWebSrc* src)
 {
     WebKitWebSrcPrivate* priv = src->priv;
 
-    ASSERT(isMainThread());
+    if (priv->resource || (priv->loader && !priv->keepAlive)) {
+        GRefPtr<WebKitWebSrc> protector = WTF::ensureGRef(src);
+        priv->notifier.cancelPendingNotifications(MainThreadSourceNotification::NeedData | MainThreadSourceNotification::EnoughData | MainThreadSourceNotification::Seek);
+        bool keepAlive = priv->keepAlive;
+        priv->notifier.notify(MainThreadSourceNotification::Stop, [protector, keepAlive] {
+            WebKitWebSrcPrivate* priv = protector->priv;
+
+            WTF::GMutexLocker<GMutex> locker(*GST_OBJECT_GET_LOCK(protector.get()));
+            if (priv->resource) {
+                priv->resource->stop();
+                priv->resource->setClient(nullptr);
+                priv->resource = nullptr;
+            }
+
+            if (!keepAlive)
+                priv->loader = nullptr;
+        });
+    }
 
     WTF::GMutexLocker<GMutex> locker(*GST_OBJECT_GET_LOCK(src));
 
     bool wasSeeking = std::exchange(priv->isSeeking, false);
 
-    priv->notifier.cancelPendingNotifications(MainThreadSourceNotification::NeedData | MainThreadSourceNotification::EnoughData | MainThreadSourceNotification::Seek);
-
-    if (priv->client) {
-        delete priv->client;
-        priv->client = 0;
-    }
-
-    if (!priv->keepAlive)
-        priv->loader = nullptr;
+    priv->client = nullptr;
 
     if (priv->buffer) {
         unmapGstBuffer(priv->buffer.get());
         priv->buffer.clear();
     }
 
-    priv->paused = FALSE;
+    priv->paused = false;
 
     priv->offset = 0;
     priv->seekable = FALSE;
@@ -510,8 +493,6 @@ static void webKitWebSrcStart(WebKitWebSrc* src)
 {
     WebKitWebSrcPrivate* priv = src->priv;
 
-    ASSERT(isMainThread());
-
     WTF::GMutexLocker<GMutex> locker(*GST_OBJECT_GET_LOCK(src));
 
     priv->didPassAccessControlCheck = false;
@@ -576,37 +557,41 @@ static void webKitWebSrcStart(WebKitWebSrc* src)
     // We always request Icecast/Shoutcast metadata, just in case ...
     request.setHTTPHeaderField(HTTPHeaderName::IcyMetadata, "1");
 
-    bool loadFailed = true;
-    if (priv->player && !priv->loader)
-        priv->loader = priv->player->createResourceLoader();
+    if (!priv->player || !priv->createdInMainThread) {
+        priv->client = std::make_unique<ResourceHandleStreamingClient>(src, WTFMove(request));
+        if (priv->client->loadFailed()) {
+            GST_ERROR_OBJECT(src, "Failed to setup streaming client");
+            priv->client = nullptr;
+            locker.unlock();
+            webKitWebSrcStop(src);
+        } else
+            GST_DEBUG_OBJECT(src, "Started request");
+        return;
+    }
+
+    locker.unlock();
+    GRefPtr<WebKitWebSrc> protector = WTF::ensureGRef(src);
+    priv->notifier.notify(MainThreadSourceNotification::Start, [protector, request] {
+        WebKitWebSrcPrivate* priv = protector->priv;
+
+        WTF::GMutexLocker<GMutex> locker(*GST_OBJECT_GET_LOCK(protector.get()));
+        if (!priv->loader)
+            priv->loader = priv->player->createResourceLoader();
 
-    if (priv->loader) {
         PlatformMediaResourceLoader::LoadOptions loadOptions = 0;
         if (request.url().protocolIsBlob())
             loadOptions |= PlatformMediaResourceLoader::LoadOption::BufferData;
         priv->resource = priv->loader->requestResource(request, loadOptions);
-        loadFailed = !priv->resource;
-
-        if (priv->resource)
-            priv->resource->setClient(std::make_unique<CachedResourceStreamingClient>(src));
-    } else {
-        priv->client = new ResourceHandleStreamingClient(src, request);
-        loadFailed = priv->client->loadFailed();
-    }
-
-    if (loadFailed) {
-        GST_ERROR_OBJECT(src, "Failed to setup streaming client");
-        if (priv->client) {
-            delete priv->client;
-            priv->client = nullptr;
+        if (priv->resource) {
+            priv->resource->setClient(std::make_unique<CachedResourceStreamingClient>(protector.get()));
+            GST_DEBUG_OBJECT(protector.get(), "Started request");
+        } else {
+            GST_ERROR_OBJECT(protector.get(), "Failed to setup streaming client");
+            priv->loader = nullptr;
+            locker.unlock();
+            webKitWebSrcStop(protector.get());
         }
-        priv->loader = nullptr;
-        priv->resource = nullptr;
-        locker.unlock();
-        webKitWebSrcStop(src);
-        return;
-    }
-    GST_DEBUG_OBJECT(src, "Started request");
+    });
 }
 
 static GstStateChangeReturn webKitWebSrcChangeState(GstElement* element, GstStateChange transition)
@@ -638,16 +623,13 @@ static GstStateChangeReturn webKitWebSrcChangeState(GstElement* element, GstStat
     case GST_STATE_CHANGE_READY_TO_PAUSED:
     {
         GST_DEBUG_OBJECT(src, "READY->PAUSED");
-        GRefPtr<WebKitWebSrc> protector = WTF::ensureGRef(src);
-        priv->notifier.notify(MainThreadSourceNotification::Start, [protector] { webKitWebSrcStart(protector.get()); });
+        webKitWebSrcStart(src);
         break;
     }
     case GST_STATE_CHANGE_PAUSED_TO_READY:
     {
         GST_DEBUG_OBJECT(src, "PAUSED->READY");
-        priv->notifier.cancelPendingNotifications();
-        GRefPtr<WebKitWebSrc> protector = WTF::ensureGRef(src);
-        priv->notifier.notify(MainThreadSourceNotification::Stop, [protector] { webKitWebSrcStop(protector.get()); });
+        webKitWebSrcStop(src);
         break;
     }
     default:
@@ -772,59 +754,91 @@ static void webKitWebSrcUriHandlerInit(gpointer gIface, gpointer)
     iface->set_uri = webKitWebSrcSetUri;
 }
 
-// appsrc callbacks
-
 static void webKitWebSrcNeedData(WebKitWebSrc* src)
 {
     WebKitWebSrcPrivate* priv = src->priv;
 
-    ASSERT(isMainThread());
-
     GST_DEBUG_OBJECT(src, "Need more data");
 
     {
         WTF::GMutexLocker<GMutex> locker(*GST_OBJECT_GET_LOCK(src));
-        priv->paused = FALSE;
+        if (!priv->paused)
+            return;
+        priv->paused = false;
+        if (priv->client) {
+            priv->client->setDefersLoading(false);
+            return;
+        }
     }
 
-    if (priv->client)
-        priv->client->setDefersLoading(false);
-    if (priv->resource)
-        priv->resource->setDefersLoading(false);
+    GRefPtr<WebKitWebSrc> protector = WTF::ensureGRef(src);
+    priv->notifier.notify(MainThreadSourceNotification::NeedData, [protector] {
+        WebKitWebSrcPrivate* priv = protector->priv;
+        if (priv->resource)
+            priv->resource->setDefersLoading(false);
+    });
 }
 
 static void webKitWebSrcEnoughData(WebKitWebSrc* src)
 {
     WebKitWebSrcPrivate* priv = src->priv;
 
-    ASSERT(isMainThread());
-
     GST_DEBUG_OBJECT(src, "Have enough data");
 
     {
         WTF::GMutexLocker<GMutex> locker(*GST_OBJECT_GET_LOCK(src));
-        priv->paused = TRUE;
+        if (priv->paused)
+            return;
+        priv->paused = true;
+        if (priv->client) {
+            priv->client->setDefersLoading(true);
+            return;
+        }
     }
 
-    if (priv->client)
-        priv->client->setDefersLoading(true);
-    if (priv->resource)
-        priv->resource->setDefersLoading(true);
+    GRefPtr<WebKitWebSrc> protector = WTF::ensureGRef(src);
+    priv->notifier.notify(MainThreadSourceNotification::EnoughData, [protector] {
+        WebKitWebSrcPrivate* priv = protector->priv;
+        if (priv->resource)
+            priv->resource->setDefersLoading(true);
+    });
 }
 
-static void webKitWebSrcSeek(WebKitWebSrc* src)
+static gboolean webKitWebSrcSeek(WebKitWebSrc* src, guint64 offset)
 {
-    ASSERT(isMainThread());
+    WebKitWebSrcPrivate* priv = src->priv;
+
+    {
+        WTF::GMutexLocker<GMutex> locker(*GST_OBJECT_GET_LOCK(src));
+        if (offset == priv->offset && priv->requestedOffset == priv->offset)
+            return TRUE;
+
+        if (!priv->seekable)
+            return FALSE;
+
+        priv->isSeeking = true;
+        priv->requestedOffset = offset;
+    }
 
     GST_DEBUG_OBJECT(src, "Seeking to offset: %" G_GUINT64_FORMAT, src->priv->requestedOffset);
+    if (priv->client) {
+        webKitWebSrcStop(src);
+        webKitWebSrcStart(src);
+        return TRUE;
+    }
 
-    webKitWebSrcStop(src);
-    webKitWebSrcStart(src);
+    GRefPtr<WebKitWebSrc> protector = WTF::ensureGRef(src);
+    priv->notifier.notify(MainThreadSourceNotification::Seek, [protector] {
+        webKitWebSrcStop(protector.get());
+        webKitWebSrcStart(protector.get());
+    });
+    return TRUE;
 }
 
 void webKitWebSrcSetMediaPlayer(WebKitWebSrc* src, WebCore::MediaPlayer* player)
 {
     ASSERT(player);
+    ASSERT(src->priv->createdInMainThread);
     WTF::GMutexLocker<GMutex> locker(*GST_OBJECT_GET_LOCK(src));
     src->priv->player = player;
 }
@@ -1051,17 +1065,47 @@ void CachedResourceStreamingClient::loadFinished(PlatformMediaResource&)
     handleNotifyFinished();
 }
 
-ResourceHandleStreamingClient::ResourceHandleStreamingClient(WebKitWebSrc* src, const ResourceRequest& request)
+ResourceHandleStreamingClient::ResourceHandleStreamingClient(WebKitWebSrc* src, ResourceRequest&& request)
     : StreamingClient(src)
 {
-    m_resource = ResourceHandle::create(0 /*context*/, request, this, false, false);
+    LockHolder locker(m_initializeRunLoopConditionMutex);
+    m_thread = createThread("ResourceHandleStreamingClient", [this, request] {
+        {
+            LockHolder locker(m_initializeRunLoopConditionMutex);
+            m_runLoop = &RunLoop::current();
+            m_resource = ResourceHandle::create(nullptr /*context*/, request, this, true, false);
+            m_initializeRunLoopCondition.notifyOne();
+        }
+        if (!m_resource)
+            return;
+
+        m_runLoop->dispatch([this] { m_resource->setDefersLoading(false); });
+        m_runLoop->run();
+        {
+            LockHolder locker(m_terminateRunLoopConditionMutex);
+            m_runLoop = nullptr;
+            m_resource->clearClient();
+            m_resource->cancel();
+            m_resource = nullptr;
+            m_terminateRunLoopCondition.notifyOne();
+        }
+    });
+    m_initializeRunLoopCondition.wait(m_initializeRunLoopConditionMutex);
 }
 
 ResourceHandleStreamingClient::~ResourceHandleStreamingClient()
 {
-    if (m_resource) {
-        m_resource->cancel();
-        m_resource = nullptr;
+    if (m_thread) {
+        detachThread(m_thread);
+        m_thread = 0;
+    }
+
+    if (m_runLoop == &RunLoop::current())
+        m_runLoop->stop();
+    else {
+        LockHolder locker(m_terminateRunLoopConditionMutex);
+        m_runLoop->stop();
+        m_terminateRunLoopCondition.wait(m_terminateRunLoopConditionMutex);
     }
 }
 
@@ -1072,8 +1116,10 @@ bool ResourceHandleStreamingClient::loadFailed() const
 
 void ResourceHandleStreamingClient::setDefersLoading(bool defers)
 {
-    if (m_resource)
-        m_resource->setDefersLoading(defers);
+    m_runLoop->dispatch([this, defers] {
+        if (m_resource)
+            m_resource->setDefersLoading(defers);
+    });
 }
 
 char* ResourceHandleStreamingClient::getOrCreateReadBuffer(size_t requestedSize, size_t& actualSize)