From 43083d6c113bcc462f09f66a5eff83a43cb9c444 Mon Sep 17 00:00:00 2001
From: Tim Blechmann <tim.blechmann@qt.io>
Date: Thu, 12 Sep 2024 12:36:39 +0800
Subject: [PATCH] GStreamer: introduce qiodevice:/ uri scheme

Using `QGstAppSource` to inject data into GStreamer has a few problems:

* it is a bit cumbersome to use, as it requires going through
signal/slot connections
* it is not possible to use the same stream in multiple locations (e.g.
player and discoverer)
* it is not thread-safe (gst_play sends on_need_data after the stream is
closed by another thread)
* the ownership of QGstAppSource is rather complicated, as it's owned by
both

So we replace the QGstAppSource by a qiodevice:/ uri scheme:
* multiple threads may access the same QIODevice in a thread-safe manner
* gstreamer handles the element construction under the hood

Task-number: QTBUG-129467
Pick-to: 6.5 6.7 6.8
Change-Id: I041d1e42368d19fb2ee64b6f3730c204458a6e85
Reviewed-by: Artem Dyomin <artem.dyomin@qt.io>
---
 .../multimedia/gstreamer/CMakeLists.txt       |   3 +-
 .../audio/qgstreameraudiodecoder.cpp          |  20 +-
 .../audio/qgstreameraudiodecoder_p.h          |   3 -
 .../gstreamer/common/qgstappsource.cpp        | 233 --------
 .../gstreamer/common/qgstappsource_p.h        |  72 ---
 .../common/qgstreamermediaplayer.cpp          |  24 +-
 .../gstreamer/qgstreamerintegration.cpp       |   2 +
 .../qgstreamer_qiodevice_handler.cpp          | 498 ++++++++++++++++++
 .../qgstreamer_qiodevice_handler_p.h          |  34 ++
 .../tst_qmediaplayerbackend.cpp               |   2 -
 10 files changed, 550 insertions(+), 341 deletions(-)
 delete mode 100644 src/plugins/multimedia/gstreamer/common/qgstappsource.cpp
 delete mode 100644 src/plugins/multimedia/gstreamer/common/qgstappsource_p.h
 create mode 100644 src/plugins/multimedia/gstreamer/uri_handler/qgstreamer_qiodevice_handler.cpp
 create mode 100644 src/plugins/multimedia/gstreamer/uri_handler/qgstreamer_qiodevice_handler_p.h

diff --git a/src/plugins/multimedia/gstreamer/CMakeLists.txt b/src/plugins/multimedia/gstreamer/CMakeLists.txt
index 22d427a55e..11f754d675 100644
--- a/src/plugins/multimedia/gstreamer/CMakeLists.txt
+++ b/src/plugins/multimedia/gstreamer/CMakeLists.txt
@@ -14,7 +14,6 @@ qt_internal_add_module(QGstreamerMediaPluginImplPrivate
         common/qgst_bus.cpp common/qgst_bus_p.h
         common/qgst_debug.cpp common/qgst_debug_p.h
         common/qgst_handle_types_p.h
-        common/qgstappsource.cpp common/qgstappsource_p.h
         common/qgstreameraudioinput.cpp common/qgstreameraudioinput_p.h
         common/qgstreameraudiooutput.cpp common/qgstreameraudiooutput_p.h
         common/qgstreamerbufferprobe.cpp common/qgstreamerbufferprobe_p.h
@@ -36,6 +35,7 @@ qt_internal_add_module(QGstreamerMediaPluginImplPrivate
         mediacapture/qgstreamerimagecapture.cpp mediacapture/qgstreamerimagecapture_p.h
         mediacapture/qgstreamermediacapturesession.cpp mediacapture/qgstreamermediacapturesession_p.h
         mediacapture/qgstreamermediarecorder.cpp mediacapture/qgstreamermediarecorder_p.h
+        uri_handler/qgstreamer_qiodevice_handler.cpp uri_handler/qgstreamer_qiodevice_handler_p.h
         uri_handler/qgstreamer_qrc_handler.cpp uri_handler/qgstreamer_qrc_handler_p.h
 
     NO_UNITY_BUILD_SOURCES
@@ -44,6 +44,7 @@ qt_internal_add_module(QGstreamerMediaPluginImplPrivate
         common/qgstreamervideosink.cpp
 
         # preprocessor / internal linkage
+        uri_handler/qgstreamer_qiodevice_handler.cpp
         uri_handler/qgstreamer_qrc_handler.cpp
     NO_GENERATE_CPP_EXPORTS
     DEFINES
diff --git a/src/plugins/multimedia/gstreamer/audio/qgstreameraudiodecoder.cpp b/src/plugins/multimedia/gstreamer/audio/qgstreameraudiodecoder.cpp
index cee226ef3c..ec292916b8 100644
--- a/src/plugins/multimedia/gstreamer/audio/qgstreameraudiodecoder.cpp
+++ b/src/plugins/multimedia/gstreamer/audio/qgstreameraudiodecoder.cpp
@@ -5,9 +5,9 @@
 #include <audio/qgstreameraudiodecoder_p.h>
 
 #include <common/qgst_debug_p.h>
-#include <common/qgstappsource_p.h>
 #include <common/qgstreamermessage_p.h>
 #include <common/qgstutils_p.h>
+#include <uri_handler/qgstreamer_qiodevice_handler_p.h>
 
 #include <gst/gstvalue.h>
 #include <gst/base/gstbasesrc.h>
@@ -71,9 +71,6 @@ QGstreamerAudioDecoder::QGstreamerAudioDecoder(QAudioDecoder *parent)
 
     g_object_set(m_playbin.object(), "audio-sink", m_outputBin.element(), NULL);
 
-    m_deepNotifySourceConnection = m_playbin.connect(
-            "deep-notify::source", (GCallback)&configureAppSrcElement, (gpointer)this);
-
     // Set volume to 100%
     gdouble volume = 1.0;
     m_playbin.set("volume", volume);
@@ -86,18 +83,6 @@ QGstreamerAudioDecoder::~QGstreamerAudioDecoder()
     m_playbin.removeMessageFilter(this);
 }
 
-void QGstreamerAudioDecoder::configureAppSrcElement([[maybe_unused]] GObject *object, GObject *orig,
-                                                    [[maybe_unused]] GParamSpec *pspec,
-                                                    QGstreamerAudioDecoder *self)
-{
-    QGstElementHandle appsrc;
-    g_object_get(orig, "source", &appsrc, NULL);
-
-    GstAppSrc *gstAppSrc = qGstSafeCast<GstAppSrc>(appsrc.release());
-    if (gstAppSrc)
-        QGstAppSource::attachQIODeviceToGstAppSrc(gstAppSrc, self->mDevice);
-}
-
 bool QGstreamerAudioDecoder::processBusMessage(const QGstreamerMessage &message)
 {
     qCDebug(qLcGstreamerAudioDecoder) << "received bus message:" << message;
@@ -310,7 +295,8 @@ void QGstreamerAudioDecoder::start()
             return;
         }
 
-        m_playbin.set("uri", "appsrc://");
+        QUrl streamURL = qGstRegisterQIODevice(mDevice);
+        m_playbin.set("uri", streamURL.toEncoded().constData());
     } else {
         return;
     }
diff --git a/src/plugins/multimedia/gstreamer/audio/qgstreameraudiodecoder_p.h b/src/plugins/multimedia/gstreamer/audio/qgstreameraudiodecoder_p.h
index f7bacd941d..60214e5bb8 100644
--- a/src/plugins/multimedia/gstreamer/audio/qgstreameraudiodecoder_p.h
+++ b/src/plugins/multimedia/gstreamer/audio/qgstreameraudiodecoder_p.h
@@ -72,9 +72,6 @@ private slots:
     static GstFlowReturn new_sample(GstAppSink *sink, gpointer user_data);
     GstFlowReturn newSample(GstAppSink *sink);
 
-    static void configureAppSrcElement(GObject *, GObject *, GParamSpec *,
-                                       QGstreamerAudioDecoder *_this);
-
     void setAudioFlags(bool wantNativeAudio);
     void addAppSink();
     void removeAppSink();
diff --git a/src/plugins/multimedia/gstreamer/common/qgstappsource.cpp b/src/plugins/multimedia/gstreamer/common/qgstappsource.cpp
deleted file mode 100644
index f2e6540368..0000000000
--- a/src/plugins/multimedia/gstreamer/common/qgstappsource.cpp
+++ /dev/null
@@ -1,233 +0,0 @@
-// Copyright (C) 2016 The Qt Company Ltd.
-// SPDX-License-Identifier: LicenseRef-Qt-Commercial OR LGPL-3.0-only OR GPL-2.0-only OR GPL-3.0-only
-
-#include "qgstappsource_p.h"
-
-#include <QtCore/qdebug.h>
-#include <QtCore/qloggingcategory.h>
-
-Q_STATIC_LOGGING_CATEGORY(qLcAppSrc, "qt.multimedia.appsrc");
-
-QT_BEGIN_NAMESPACE
-
-QGstAppSource::QGstAppSource(GstAppSrc *owner, QIODevice *stream, qint64 offset)
-    : m_owningAppSrc(owner)
-{
-    Q_ASSERT(owner);
-
-    QGstAppSrc ownerObject{
-        m_owningAppSrc,
-        QGstAppSrc::NeedsRef,
-    };
-
-    ownerObject.set("qgst-app-src", this, QGstObject::qDeleteFromVoidPointer<QGstAppSource>);
-
-    setup(stream, offset);
-}
-
-QGstAppSource::~QGstAppSource()
-{
-    streamDestroyed();
-    qCDebug(qLcAppSrc) << "~QGstAppSrc";
-}
-
-void QGstAppSource::attachQIODeviceToGstAppSrc(GstAppSrc *source, QIODevice *ioDevice)
-{
-    // NOLINTBEGIN(clang-analyzer-cplusplus.NewDeleteLeaks)
-    // GstAppSrc takes ownership of QGstAppSource
-
-    QGstAppSource *appSource = new QGstAppSource(source, ioDevice);
-    Q_ASSERT(appSource);
-
-    // NOLINTEND(clang-analyzer-cplusplus.NewDeleteLeaks)
-}
-
-bool QGstAppSource::setup(QIODevice *stream, qint64 offset)
-{
-    QMutexLocker locker(&m_mutex);
-
-    if (!setStream(stream, offset))
-        return false;
-
-    GstAppSrcCallbacks callbacks{};
-    callbacks.need_data = QGstAppSource::on_need_data;
-    callbacks.enough_data = QGstAppSource::on_enough_data;
-    callbacks.seek_data = QGstAppSource::on_seek_data;
-
-    QGstAppSrc{ m_owningAppSrc, QGstAppSrc::NeedsRef }.setCallbacks(callbacks, this, nullptr);
-
-    gst_app_src_set_max_bytes(m_owningAppSrc, maxBytes);
-
-    if (m_sequential) {
-        gst_app_src_set_stream_type(m_owningAppSrc, GST_APP_STREAM_TYPE_STREAM);
-        gst_app_src_set_size(m_owningAppSrc, -1);
-    } else {
-        gst_app_src_set_stream_type(m_owningAppSrc, GST_APP_STREAM_TYPE_RANDOM_ACCESS);
-        gst_app_src_set_size(m_owningAppSrc, m_stream->size() - m_offset);
-    }
-
-    return true;
-}
-
-bool QGstAppSource::setStream(QIODevice *stream, qint64 offset)
-{
-    if (m_stream) {
-        disconnect(m_stream, &QIODevice::readyRead, this, &QGstAppSource::onDataReady);
-        disconnect(m_stream, &QIODevice::destroyed, this, &QGstAppSource::streamDestroyed);
-        m_stream = nullptr;
-    }
-
-    m_sequential = true;
-
-    if (stream) {
-        if (!stream->isOpen() && !stream->open(QIODevice::ReadOnly))
-            return false;
-        m_stream = stream;
-        connect(m_stream, &QIODevice::destroyed, this, &QGstAppSource::streamDestroyed);
-        connect(m_stream, &QIODevice::readyRead, this, &QGstAppSource::onDataReady);
-        m_sequential = m_stream->isSequential();
-        m_offset = offset;
-    }
-    return true;
-}
-
-bool QGstAppSource::isStreamValid() const
-{
-    return m_stream != nullptr && m_stream->isOpen();
-}
-void QGstAppSource::onDataReady()
-{
-    QMutexLocker locker(&m_mutex);
-    qCDebug(qLcAppSrc) << "onDataReady" << m_stream->bytesAvailable() << m_stream->size();
-    pushData(m_stream->bytesAvailable());
-}
-
-void QGstAppSource::streamDestroyed()
-{
-    QMutexLocker locker(&m_mutex);
-    qCDebug(qLcAppSrc) << "stream destroyed";
-    m_stream = nullptr;
-    sendEOS();
-}
-
-void QGstAppSource::pushData(qint64 bytesToRead)
-{
-    if (!m_stream) {
-        qCDebug(qLcAppSrc) << "push data: return immediately - stream was destroyed";
-        return;
-    }
-
-    if (!m_dataNeeded) {
-        qCDebug(qLcAppSrc) << "push data: return immediately - no data needed";
-        return;
-    }
-
-    Q_ASSERT(m_stream);
-
-    qCDebug(qLcAppSrc) << "pushData" << m_stream;
-    if ((m_stream && m_stream->atEnd())) {
-        sendEOS();
-        qCDebug(qLcAppSrc) << "end pushData" << m_stream;
-        return;
-    }
-
-    qint64 size = m_stream->bytesAvailable();
-
-    size = qMin(size, bytesToRead);
-    qCDebug(qLcAppSrc) << "    reading" << size << "bytes of requested" << bytesToRead;
-
-    GstBuffer *buffer = gst_buffer_new_and_alloc(size);
-
-    if (m_sequential)
-        buffer->offset = bytesReadSoFar;
-    else
-        buffer->offset = m_stream->pos();
-
-    GstMapInfo mapInfo;
-    gst_buffer_map(buffer, &mapInfo, GST_MAP_WRITE);
-    void* bufferData = mapInfo.data;
-
-    qint64 bytesRead;
-    bytesRead = m_stream->read((char *)bufferData, size);
-
-    buffer->offset_end = buffer->offset + bytesRead - 1;
-    bytesReadSoFar += bytesRead;
-
-    gst_buffer_unmap(buffer, &mapInfo);
-    qCDebug(qLcAppSrc) << "pushing bytes into gstreamer" << buffer->offset << bytesRead;
-    if (bytesRead == 0) {
-        gst_buffer_unref(buffer);
-        sendEOS();
-        qCDebug(qLcAppSrc) << "end pushData" << m_stream;
-        return;
-    }
-
-    GstFlowReturn ret = gst_app_src_push_buffer(m_owningAppSrc, buffer);
-    switch (ret) {
-    case GST_FLOW_OK:
-        break;
-
-    default:
-        qWarning() << "QGstAppSrc: push buffer error -" << gst_flow_get_name(ret);
-        break;
-    }
-
-    qCDebug(qLcAppSrc) << "end pushData" << m_stream;
-}
-
-bool QGstAppSource::doSeek(qint64 streamPosition)
-{
-    if (isStreamValid())
-        return m_stream->seek(streamPosition + m_offset);
-    return false;
-}
-
-gboolean QGstAppSource::on_seek_data(GstAppSrc *, guint64 streamPosition, gpointer userdata)
-{
-    // we do get some spurious seeks to INT_MAX, ignore those
-    if (streamPosition == std::numeric_limits<quint64>::max())
-        return true;
-
-    QGstAppSource *self = reinterpret_cast<QGstAppSource *>(userdata);
-    Q_ASSERT(self);
-
-    QMutexLocker locker(&self->m_mutex);
-
-    if (self->m_sequential)
-        return false;
-
-    self->doSeek(streamPosition);
-    return true;
-}
-
-void QGstAppSource::on_enough_data(GstAppSrc *, gpointer userdata)
-{
-    qCWarning(qLcAppSrc) << "on_enough_data";
-
-    QGstAppSource *self = reinterpret_cast<QGstAppSource *>(userdata);
-    Q_ASSERT(self);
-
-    QMutexLocker locker(&self->m_mutex);
-    self->m_dataNeeded = false;
-}
-
-void QGstAppSource::on_need_data(GstAppSrc *, guint numberOfBytes, gpointer userdata)
-{
-    qCDebug(qLcAppSrc) << "on_need_data requesting bytes" << numberOfBytes;
-    QGstAppSource *self = static_cast<QGstAppSource *>(userdata);
-    Q_ASSERT(self);
-    QMutexLocker locker(&self->m_mutex);
-    self->m_dataNeeded = true;
-    self->pushData(numberOfBytes);
-    qCDebug(qLcAppSrc) << "done on_need_data";
-}
-
-void QGstAppSource::sendEOS()
-{
-    qCDebug(qLcAppSrc) << "sending EOS";
-    gst_app_src_end_of_stream(m_owningAppSrc);
-}
-
-QT_END_NAMESPACE
-
-#include "moc_qgstappsource_p.cpp"
diff --git a/src/plugins/multimedia/gstreamer/common/qgstappsource_p.h b/src/plugins/multimedia/gstreamer/common/qgstappsource_p.h
deleted file mode 100644
index f6bd94bfa7..0000000000
--- a/src/plugins/multimedia/gstreamer/common/qgstappsource_p.h
+++ /dev/null
@@ -1,72 +0,0 @@
-// Copyright (C) 2016 The Qt Company Ltd.
-// SPDX-License-Identifier: LicenseRef-Qt-Commercial OR LGPL-3.0-only OR GPL-2.0-only OR GPL-3.0-only
-
-#ifndef QGSTAPPSRC_H
-#define QGSTAPPSRC_H
-
-//
-//  W A R N I N G
-//  -------------
-//
-// This file is not part of the Qt API.  It exists purely as an
-// implementation detail.  This header file may change from version to
-// version without notice, or even be removed.
-//
-// We mean it.
-//
-
-#include <QtCore/qobject.h>
-#include <QtCore/qiodevice.h>
-#include <QtCore/qmutex.h>
-
-#include <QtMultimedia/private/qtmultimediaglobal_p.h>
-
-#include <common/qgst_p.h>
-#include <gst/app/gstappsrc.h>
-
-QT_BEGIN_NAMESPACE
-
-class QGstAppSource : public QObject
-{
-    Q_OBJECT
-public:
-    explicit QGstAppSource(GstAppSrc *owner, QIODevice *, qint64 offset = 0);
-    ~QGstAppSource();
-
-    static void attachQIODeviceToGstAppSrc(GstAppSrc *, QIODevice *);
-
-private Q_SLOTS:
-    void onDataReady();
-    void streamDestroyed();
-
-private:
-    bool setup(QIODevice *stream = nullptr, qint64 offset = 0);
-
-    bool doSeek(qint64 streamPosition);
-    void pushData(qint64 bytesToRead);
-
-    bool setStream(QIODevice *, qint64 offset);
-    bool isStreamValid() const;
-
-    static gboolean on_seek_data(GstAppSrc *element, guint64 arg0, gpointer userdata);
-    static void on_enough_data(GstAppSrc *element, gpointer userdata);
-    static void on_need_data(GstAppSrc *element, uint arg0, gpointer userdata);
-
-    void sendEOS();
-
-    mutable QMutex m_mutex;
-
-    QIODevice *m_stream = nullptr;
-
-    GstAppSrc *m_owningAppSrc = nullptr; // QGstAppSource is owned by this GstAppSrc.
-    bool m_sequential = true;
-    qint64 m_offset = 0;
-    qint64 bytesReadSoFar = 0;
-    bool m_dataNeeded = false;
-
-    static constexpr qint64 maxBytes = 64 * 1024;
-};
-
-QT_END_NAMESPACE
-
-#endif
diff --git a/src/plugins/multimedia/gstreamer/common/qgstreamermediaplayer.cpp b/src/plugins/multimedia/gstreamer/common/qgstreamermediaplayer.cpp
index 34d6d40e4c..3397c4cd15 100644
--- a/src/plugins/multimedia/gstreamer/common/qgstreamermediaplayer.cpp
+++ b/src/plugins/multimedia/gstreamer/common/qgstreamermediaplayer.cpp
@@ -5,19 +5,20 @@
 
 #include <audio/qgstreameraudiodevice_p.h>
 #include <common/qgst_debug_p.h>
-#include <common/qgstappsource_p.h>
 #include <common/qgstpipeline_p.h>
 #include <common/qgstreameraudiooutput_p.h>
 #include <common/qgstreamermessage_p.h>
 #include <common/qgstreamermetadata_p.h>
 #include <common/qgstreamervideooutput_p.h>
 #include <common/qgstreamervideosink_p.h>
+#include <uri_handler/qgstreamer_qiodevice_handler_p.h>
 #include <qgstreamerformatinfo_p.h>
 
 #include <QtMultimedia/qaudiodevice.h>
-#include <QtCore/qurl.h>
 #include <QtCore/qdebug.h>
+#include <QtCore/qiodevice.h>
 #include <QtCore/qloggingcategory.h>
+#include <QtCore/qurl.h>
 #include <QtCore/private/quniquehandle_p.h>
 
 #if QT_CONFIG(gstreamer_gl)
@@ -357,7 +358,8 @@ const QGstPipeline &QGstreamerMediaPlayer::pipeline() const
 
 bool QGstreamerMediaPlayer::canPlayQrc() const
 {
-    return true;
+    return false;
+    // return true;
 }
 
 void QGstreamerMediaPlayer::stopOrEOS(bool eos)
@@ -879,12 +881,6 @@ void QGstreamerMediaPlayer::sourceSetupCallback(GstElement *uridecodebin, GstEle
     const gchar *typeName = g_type_name_from_instance((GTypeInstance *)source);
     qCDebug(qLcMediaPlayer) << "Setting up source:" << typeName;
 
-    if (typeName == std::string_view("GstAppSrc")) {
-        QGstAppSource::attachQIODeviceToGstAppSrc(qGstCheckedCast<GstAppSrc>(source),
-                                                  self->m_stream);
-        return;
-    }
-
     if (typeName == std::string_view("GstRTSPSrc")) {
         QGstElement s(source, QGstElement::NeedsRef);
         int latency{40};
@@ -981,6 +977,9 @@ void QGstreamerMediaPlayer::setMedia(const QUrl &content, QIODevice *stream)
 
     m_url = content;
     m_stream = stream;
+    QUrl streamURL;
+    if (stream)
+        streamURL = qGstRegisterQIODevice(stream);
 
     if (decoder) {
         playerPipeline.stopAndRemoveElements(decoder);
@@ -1044,13 +1043,12 @@ void QGstreamerMediaPlayer::setMedia(const QUrl &content, QIODevice *stream)
     padAdded = decoder.onPadAdded<&QGstreamerMediaPlayer::decoderPadAdded>(this);
     padRemoved = decoder.onPadRemoved<&QGstreamerMediaPlayer::decoderPadRemoved>(this);
 
+    QUrl uri = m_stream ? streamURL : content;
+    decoder.set("uri", uri.toEncoded().constData());
     if (m_stream) {
-        decoder.set("uri", "appsrc://");
         seekableChanged(!m_stream->isSequential());
     } else {
-        QByteArray contentUri = content.toEncoded();
-        decoder.set("uri", contentUri.constData());
-        if (contentUri.startsWith("qrc:"))
+        if (content.toEncoded().startsWith("qrc:"))
             seekableChanged(true); // qrc resources are seekable
     }
 
diff --git a/src/plugins/multimedia/gstreamer/qgstreamerintegration.cpp b/src/plugins/multimedia/gstreamer/qgstreamerintegration.cpp
index e72541efdd..1da5759c83 100644
--- a/src/plugins/multimedia/gstreamer/qgstreamerintegration.cpp
+++ b/src/plugins/multimedia/gstreamer/qgstreamerintegration.cpp
@@ -14,6 +14,7 @@
 #include <mediacapture/qgstreamerimagecapture_p.h>
 #include <mediacapture/qgstreamermediacapturesession_p.h>
 #include <mediacapture/qgstreamermediarecorder_p.h>
+#include <uri_handler/qgstreamer_qiodevice_handler_p.h>
 #include <uri_handler/qgstreamer_qrc_handler_p.h>
 
 #include <QtCore/qloggingcategory.h>
@@ -163,6 +164,7 @@ QGstreamerIntegration::QGstreamerIntegration()
     }
 
     qGstRegisterQRCHandler(nullptr);
+    qGstRegisterQIODeviceHandler(nullptr);
 }
 
 QPlatformMediaFormatInfo *QGstreamerIntegration::createFormatInfo()
diff --git a/src/plugins/multimedia/gstreamer/uri_handler/qgstreamer_qiodevice_handler.cpp b/src/plugins/multimedia/gstreamer/uri_handler/qgstreamer_qiodevice_handler.cpp
new file mode 100644
index 0000000000..9aa617e3be
--- /dev/null
+++ b/src/plugins/multimedia/gstreamer/uri_handler/qgstreamer_qiodevice_handler.cpp
@@ -0,0 +1,498 @@
+// Copyright (C) 2024 The Qt Company Ltd.
+// SPDX-License-Identifier: LicenseRef-Qt-Commercial OR LGPL-3.0-only OR GPL-2.0-only OR GPL-3.0-only
+
+#include "qgstreamer_qiodevice_handler_p.h"
+
+#include <QtCore/qdebug.h>
+#include <QtCore/qglobal.h>
+#include <QtCore/qiodevice.h>
+#include <QtCore/qmap.h>
+#include <QtCore/qmutex.h>
+#include <QtCore/qobject.h>
+#include <QtCore/qspan.h>
+#include <QtCore/qurl.h>
+#include <QtCore/quuid.h>
+
+#include <gst/base/gstbasesrc.h>
+#include <map>
+#include <memory>
+#include <utility>
+
+QT_BEGIN_NAMESPACE
+
+namespace {
+
+using namespace Qt::Literals;
+
+// QIODeviceRegistry
+
+class QIODeviceRegistry : public QObject
+{
+public:
+    struct Record
+    {
+        explicit Record(QByteArray, QIODevice *);
+
+        void unsetDevice();
+        bool isValid() const;
+
+        const QByteArray id;
+
+        template <typename Functor>
+        auto runWhileLocked(Functor &&f)
+        {
+            QMutexLocker guard(&mutex);
+            return f(device);
+        }
+
+    private:
+        QIODevice *device;
+        mutable QMutex mutex;
+    };
+    using SharedRecord = std::shared_ptr<Record>;
+
+    QByteArray registerQIODevice(QIODevice *);
+    SharedRecord findRecord(QByteArrayView);
+
+private:
+    void unregisterDevice(QIODevice *);
+    void deviceDestroyed(QIODevice *);
+
+    QMutex m_registryMutex;
+    std::map<QByteArray, SharedRecord, std::less<>> m_registry;
+    QMap<QIODevice *, QByteArray> m_reverseLookupTable;
+};
+
+QByteArray QIODeviceRegistry::registerQIODevice(QIODevice *device)
+{
+    Q_ASSERT(device);
+
+    if (device->isSequential())
+        qWarning() << "GStreamer: sequential QIODevices are not fully supported";
+
+    QMutexLocker lock(&m_registryMutex);
+
+    auto it = m_reverseLookupTable.find(device);
+    if (it != m_reverseLookupTable.end())
+        return it.value();
+
+    QByteArray identifier =
+            "qiodevice:/"_ba + QUuid::createUuid().toByteArray(QUuid::StringFormat::Id128);
+
+    m_registry.emplace(identifier, std::make_shared<Record>(identifier, device));
+
+    QMetaObject::Connection destroyedConnection = QObject::connect(
+            device, &QObject::destroyed, this,
+            [this, device] {
+        // Caveat: if the QIODevice has not been closed, we unregister the device, however gstreamer
+        // worker threads have a chance to briefly read from a partially destroyed QIODevice.
+        // There's nothing we can do about it
+        unregisterDevice(device);
+    },
+            Qt::DirectConnection);
+
+    QObject::connect(
+            device, &QIODevice::aboutToClose, this,
+            [this, device, destroyedConnection = std::move(destroyedConnection)] {
+        unregisterDevice(device);
+        disconnect(destroyedConnection);
+    },
+            Qt::DirectConnection);
+
+    m_reverseLookupTable.insert(device, identifier);
+    return identifier;
+}
+
+QIODeviceRegistry::SharedRecord QIODeviceRegistry::findRecord(QByteArrayView id)
+{
+    QMutexLocker registryLock(&m_registryMutex);
+
+    auto it = m_registry.find(id);
+    if (it != m_registry.end())
+        return it->second;
+    return {};
+}
+
+void QIODeviceRegistry::unregisterDevice(QIODevice *device)
+{
+    QMutexLocker registryLock(&m_registryMutex);
+    auto reverseLookupIt = m_reverseLookupTable.find(device);
+    if (reverseLookupIt == m_reverseLookupTable.end())
+        return;
+
+    auto it = m_registry.find(reverseLookupIt.value());
+    Q_ASSERT(it != m_registry.end());
+
+    it->second->unsetDevice();
+    m_reverseLookupTable.erase(reverseLookupIt);
+    m_registry.erase(it);
+}
+
+QIODeviceRegistry::Record::Record(QByteArray id, QIODevice *device)
+    : id {
+          std::move(id),
+      },
+      device {
+          device,
+      }
+{
+    if (!device->isOpen())
+        device->open(QIODevice::ReadOnly);
+}
+
+void QIODeviceRegistry::Record::unsetDevice()
+{
+    QMutexLocker lock(&mutex);
+    device = nullptr;
+}
+
+bool QIODeviceRegistry::Record::isValid() const
+{
+    QMutexLocker lock(&mutex);
+    return device;
+}
+
+Q_GLOBAL_STATIC(QIODeviceRegistry, gQIODeviceRegistry);
+
+// qt helpers
+
+// glib / gstreamer object
+enum PropertyId : uint8_t {
+    PROP_NONE,
+    PROP_URI,
+};
+
+struct QGstQIODeviceSrc
+{
+    void getProperty(guint propId, GValue *value, const GParamSpec *pspec) const;
+    void setProperty(guint propId, const GValue *value, const GParamSpec *pspec);
+    auto lockObject() const;
+
+    bool start();
+    bool stop();
+
+    bool isSeekable();
+    std::optional<guint64> size();
+    GstFlowReturn fill(guint64 offset, guint length, GstBuffer *buf);
+    void getURI(GValue *value) const;
+    bool setURI(const char *location, GError **err = nullptr);
+
+    GstBaseSrc baseSrc;
+    QIODeviceRegistry::SharedRecord record;
+};
+
+void QGstQIODeviceSrc::getProperty(guint propId, GValue *value, const GParamSpec *pspec) const
+{
+    switch (propId) {
+    case PROP_URI:
+        return getURI(value);
+
+    default:
+        G_OBJECT_WARN_INVALID_PROPERTY_ID(this, propId, pspec);
+        break;
+    }
+}
+
+void QGstQIODeviceSrc::setProperty(guint propId, const GValue *value, const GParamSpec *pspec)
+{
+    switch (propId) {
+    case PROP_URI:
+        setURI(g_value_get_string(value));
+        break;
+    default:
+        G_OBJECT_WARN_INVALID_PROPERTY_ID(this, propId, pspec);
+        break;
+    }
+}
+
+auto QGstQIODeviceSrc::lockObject() const
+{
+    GST_OBJECT_LOCK(this);
+    return qScopeGuard([this] {
+        GST_OBJECT_UNLOCK(this);
+    });
+}
+
+bool QGstQIODeviceSrc::start()
+{
+    auto lock = lockObject();
+
+    return record && record->isValid();
+}
+
+bool QGstQIODeviceSrc::stop()
+{
+    return true;
+}
+
+bool QGstQIODeviceSrc::isSeekable()
+{
+    auto lock = lockObject();
+    return record->runWhileLocked([&](QIODevice *device) {
+        return !device->isSequential();
+    });
+}
+
+std::optional<guint64> QGstQIODeviceSrc::size()
+{
+    auto lock = lockObject();
+    if (!record)
+        return std::nullopt;
+
+    qint64 size = record->runWhileLocked([&](QIODevice *device) {
+        return device->size();
+    });
+
+    if (size == -1)
+        return std::nullopt;
+    return size;
+}
+
+GstFlowReturn QGstQIODeviceSrc::fill(guint64 offset, guint length, GstBuffer *buf)
+{
+    auto lock = lockObject();
+
+    if (!record)
+        return GST_FLOW_ERROR;
+
+    GstMapInfo info;
+    if (!gst_buffer_map(buf, &info, GST_MAP_WRITE)) {
+        GST_ELEMENT_ERROR(this, RESOURCE, WRITE, (nullptr), ("Can't map buffer for writing"));
+        return GST_FLOW_ERROR;
+    }
+
+    int64_t totalRead = 0;
+    GstFlowReturn ret = record->runWhileLocked([&](QIODevice *device) -> GstFlowReturn {
+        if (device->pos() != qint64(offset)) {
+            bool success = device->seek(offset);
+            if (!success) {
+                qWarning() << "seek on iodevice failed";
+                return GST_FLOW_ERROR;
+            }
+        }
+
+        int64_t remain = length;
+        while (remain) {
+            int64_t bytesRead =
+                    device->read(reinterpret_cast<char *>(info.data + totalRead), remain);
+            if (bytesRead == -1) {
+                if (device->atEnd()) {
+                    return GST_FLOW_EOS;
+                }
+                GST_ELEMENT_ERROR(this, RESOURCE, READ, (nullptr), GST_ERROR_SYSTEM);
+                return GST_FLOW_ERROR;
+            }
+
+            remain -= bytesRead;
+            totalRead += bytesRead;
+        }
+
+        return GST_FLOW_OK;
+    });
+
+    if (ret != GST_FLOW_OK) {
+        gst_buffer_unmap(buf, &info);
+        gst_buffer_resize(buf, 0, 0);
+        return ret;
+    }
+
+    gst_buffer_unmap(buf, &info);
+    if (totalRead != length)
+        gst_buffer_resize(buf, 0, totalRead);
+
+    GST_BUFFER_OFFSET(buf) = offset;
+    GST_BUFFER_OFFSET_END(buf) = offset + totalRead;
+
+    return GST_FLOW_OK;
+}
+
+void QGstQIODeviceSrc::getURI(GValue *value) const
+{
+    auto lock = lockObject();
+    if (record)
+        g_value_set_string(value, record->id);
+    else
+        g_value_set_string(value, nullptr);
+}
+
+bool QGstQIODeviceSrc::setURI(const char *location, GError **err)
+{
+    Q_ASSERT(QLatin1StringView(location).startsWith("qiodevice:/"_L1));
+
+    {
+        auto lock = lockObject();
+        GstState state = GST_STATE(this);
+        if (state != GST_STATE_READY && state != GST_STATE_NULL) {
+            g_warning(
+                    "Changing the `uri' property on qiodevicesrc when the resource is open is not "
+                    "supported.");
+            if (err)
+                g_set_error(err, GST_URI_ERROR, GST_URI_ERROR_BAD_STATE,
+                            "Changing the `uri' property on qiodevicesrc when the resource is open "
+                            "is not supported.");
+            return false;
+        }
+    }
+
+    auto newRecord = gQIODeviceRegistry->findRecord(QByteArrayView{ location });
+
+    {
+        auto lock = lockObject();
+        record = std::move(newRecord);
+    }
+
+    g_object_notify(G_OBJECT(this), "uri");
+
+    return true;
+}
+
+struct QGstQIODeviceSrcClass
+{
+    GstBaseSrcClass parent_class;
+};
+
+// GObject
+static void gst_qiodevice_src_class_init(QGstQIODeviceSrcClass *klass);
+static void gst_qiodevice_src_init(QGstQIODeviceSrc *self);
+
+GType gst_qiodevice_src_get_type();
+
+template <typename T>
+QGstQIODeviceSrc *asQGstQIODeviceSrc(T *obj)
+{
+    return (G_TYPE_CHECK_INSTANCE_CAST((obj), gst_qiodevice_src_get_type(), QGstQIODeviceSrc));
+}
+
+// URI handler
+void qGstInitQIODeviceURIHandler(gpointer, gpointer);
+
+#define gst_qiodevice_src_parent_class parent_class
+G_DEFINE_TYPE_WITH_CODE(QGstQIODeviceSrc, gst_qiodevice_src, GST_TYPE_BASE_SRC,
+                        G_IMPLEMENT_INTERFACE(GST_TYPE_URI_HANDLER, qGstInitQIODeviceURIHandler));
+
+// implementations
+
+void gst_qiodevice_src_init(QGstQIODeviceSrc *self)
+{
+    using SharedRecord = QIODeviceRegistry::SharedRecord;
+
+    new (reinterpret_cast<void *>(&self->record)) SharedRecord;
+
+    static constexpr guint defaultBlockSize = 16384;
+    gst_base_src_set_blocksize(&self->baseSrc, defaultBlockSize);
+}
+
+void gst_qiodevice_src_class_init(QGstQIODeviceSrcClass *klass)
+{
+    // GObject
+    GObjectClass *gobjectClass = G_OBJECT_CLASS(klass);
+    gobjectClass->set_property = [](GObject *instance, guint propId, const GValue *value,
+                                    GParamSpec *pspec) {
+        QGstQIODeviceSrc *src = asQGstQIODeviceSrc(instance);
+        return src->setProperty(propId, value, pspec);
+    };
+
+    gobjectClass->get_property = [](GObject *instance, guint propId, GValue *value,
+                                    GParamSpec *pspec) {
+        QGstQIODeviceSrc *src = asQGstQIODeviceSrc(instance);
+        return src->getProperty(propId, value, pspec);
+    };
+
+    g_object_class_install_property(
+            gobjectClass, PROP_URI,
+            g_param_spec_string("uri", "QRC Location", "Path of the qrc to read", nullptr,
+                                GParamFlags(G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS
+                                            | GST_PARAM_MUTABLE_READY)));
+
+    gobjectClass->finalize = [](GObject *instance) {
+        QGstQIODeviceSrc *src = asQGstQIODeviceSrc(instance);
+        using SharedRecord = QIODeviceRegistry::SharedRecord;
+        src->record.~SharedRecord();
+        G_OBJECT_CLASS(parent_class)->finalize(instance);
+    };
+
+    // GstElement
+    GstElementClass *gstelementClass = GST_ELEMENT_CLASS(klass);
+    gst_element_class_set_static_metadata(gstelementClass, "QRC Source", "Source/QRC",
+                                          "Read from arbitrary point in QRC resource",
+                                          "Tim Blechmann <tim.blechmann@qt.io>");
+
+    static GstStaticPadTemplate srctemplate =
+            GST_STATIC_PAD_TEMPLATE("src", GST_PAD_SRC, GST_PAD_ALWAYS, GST_STATIC_CAPS_ANY);
+
+    gst_element_class_add_static_pad_template(gstelementClass, &srctemplate);
+
+    // GstBaseSrc
+    GstBaseSrcClass *gstbasesrcClass = GST_BASE_SRC_CLASS(klass);
+    gstbasesrcClass->start = [](GstBaseSrc *basesrc) -> gboolean {
+        QGstQIODeviceSrc *src = asQGstQIODeviceSrc(basesrc);
+        return src->start();
+    };
+    gstbasesrcClass->stop = [](GstBaseSrc *basesrc) -> gboolean {
+        QGstQIODeviceSrc *src = asQGstQIODeviceSrc(basesrc);
+        return src->stop();
+    };
+
+    gstbasesrcClass->is_seekable = [](GstBaseSrc *basesrc) -> gboolean {
+        QGstQIODeviceSrc *src = asQGstQIODeviceSrc(basesrc);
+        return src->isSeekable();
+    };
+    gstbasesrcClass->get_size = [](GstBaseSrc *basesrc, guint64 *size) -> gboolean {
+        QGstQIODeviceSrc *src = asQGstQIODeviceSrc(basesrc);
+        auto optionalSize = src->size();
+        if (!optionalSize)
+            return false;
+        *size = optionalSize.value();
+        return true;
+    };
+    gstbasesrcClass->fill = [](GstBaseSrc *basesrc, guint64 offset, guint length,
+                               GstBuffer *buf) -> GstFlowReturn {
+        QGstQIODeviceSrc *src = asQGstQIODeviceSrc(basesrc);
+        return src->fill(offset, length, buf);
+    };
+}
+
+void qGstInitQIODeviceURIHandler(gpointer g_handlerInterface, gpointer)
+{
+    GstURIHandlerInterface *iface = (GstURIHandlerInterface *)g_handlerInterface;
+
+    iface->get_type = [](GType) {
+        return GST_URI_SRC;
+    };
+    iface->get_protocols = [](GType) {
+        static constexpr const gchar *protocols[] = {
+            "qiodevice",
+            nullptr,
+        };
+        return protocols;
+    };
+    iface->get_uri = [](GstURIHandler *handler) -> gchar * {
+        QGstQIODeviceSrc *src = asQGstQIODeviceSrc(handler);
+        auto lock = src->lockObject();
+        if (src->record)
+            return g_strdup(src->record->id.constData());
+        return nullptr;
+    };
+    iface->set_uri = [](GstURIHandler *handler, const gchar *uri, GError **err) -> gboolean {
+        QGstQIODeviceSrc *src = asQGstQIODeviceSrc(handler);
+        return src->setURI(uri, err);
+    };
+}
+
+} // namespace
+
+// plugin registration
+
+void qGstRegisterQIODeviceHandler(GstPlugin *plugin)
+{
+    gst_element_register(plugin, "qiodevicesrc", GST_RANK_PRIMARY, gst_qiodevice_src_get_type());
+}
+
+QUrl qGstRegisterQIODevice(QIODevice *device)
+{
+    return QUrl{
+        QString::fromLatin1(gQIODeviceRegistry->registerQIODevice(device)),
+    };
+}
+
+QT_END_NAMESPACE
diff --git a/src/plugins/multimedia/gstreamer/uri_handler/qgstreamer_qiodevice_handler_p.h b/src/plugins/multimedia/gstreamer/uri_handler/qgstreamer_qiodevice_handler_p.h
new file mode 100644
index 0000000000..ec248aaafc
--- /dev/null
+++ b/src/plugins/multimedia/gstreamer/uri_handler/qgstreamer_qiodevice_handler_p.h
@@ -0,0 +1,34 @@
+// Copyright (C) 2024 The Qt Company Ltd.
+// SPDX-License-Identifier: LicenseRef-Qt-Commercial OR LGPL-3.0-only OR GPL-2.0-only OR GPL-3.0-only
+
+#ifndef QGSTREAMER_QIODEVICE_HANDLER_H
+#define QGSTREAMER_QIODEVICE_HANDLER_H
+
+#include <QtCore/qglobal.h>
+#include <gst/gstplugin.h>
+
+//
+//  W A R N I N G
+//  -------------
+//
+// This file is not part of the Qt API. It exists purely as an
+// implementation detail. This header file may change from version to
+// version without notice, or even be removed.
+//
+// We mean it.
+//
+
+QT_BEGIN_NAMESPACE
+
+class QIODevice;
+class QUrl;
+
+void qGstRegisterQIODeviceHandler(GstPlugin *plugin);
+
+// Note: if the QIODevice is sequential, the QUrl is only allowed to be used in one location.
+// if the QUrl is not sequential, it can be passed to multiple destinations
+QUrl qGstRegisterQIODevice(QIODevice *);
+
+QT_END_NAMESPACE
+
+#endif
diff --git a/tests/auto/integration/qmediaplayerbackend/tst_qmediaplayerbackend.cpp b/tests/auto/integration/qmediaplayerbackend/tst_qmediaplayerbackend.cpp
index 07208ca063..b5dd9f3e11 100644
--- a/tests/auto/integration/qmediaplayerbackend/tst_qmediaplayerbackend.cpp
+++ b/tests/auto/integration/qmediaplayerbackend/tst_qmediaplayerbackend.cpp
@@ -3147,8 +3147,6 @@ void tst_QMediaPlayerBackend::playFromSequentialStream()
     player.setVideoOutput(&surface);
     QSequentialFileAdaptor file(u":"_s + m_localVideoFile->toEncoded(QUrl::RemoveScheme));
 
-    QSKIP_GSTREAMER("QGstAppSource misbehaves on sequential streams");
-
     QObject::connect(&player, &QMediaPlayer::errorOccurred, &player, [] {
         QFAIL("playFromSequentialStream: error occurred");
     });