Revision control
++ // 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/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 <mutex>
++ #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;
++ std::map<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->second;
++
++ 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.emplace(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->second);
++ 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);
++
++ 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);
++
++ // lockable
++ void lock() const { GST_OBJECT_LOCK(&baseSrc); }
++ void unlock() const { GST_OBJECT_UNLOCK(&baseSrc); }
++
++ 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;
++ }
++ }
++
++ bool QGstQIODeviceSrc::start()
++ {
++ std::lock_guard guard{ *this };
++ return record && record->isValid();
++ }
++
++ bool QGstQIODeviceSrc::stop()
++ {
++ return true;
++ }
++
++ bool QGstQIODeviceSrc::isSeekable()
++ {
++ std::lock_guard guard{ *this };
++ return record->runWhileLocked([&](QIODevice *device) {
++ return !device->isSequential();
++ });
++ }
++
++ std::optional<guint64> QGstQIODeviceSrc::size()
++ {
++ std::lock_guard guard{ *this };
++ 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)
++ {
++ std::unique_lock guard{ *this };
++ if (!record)
++ return GST_FLOW_ERROR;
++
++ GstMapInfo info;
++ if (!gst_buffer_map(buf, &info, GST_MAP_WRITE)) {
++ guard.unlock();
++ GST_ELEMENT_ERROR(&baseSrc, 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;
++ }
++ guard.unlock();
++ GST_ELEMENT_ERROR(&baseSrc, 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
++ {
++ std::lock_guard guard{ *this };
++ if (record)
++ g_value_set_string(value, record->id.constData());
++ else
++ g_value_set_string(value, nullptr);
++ }
++
++ bool QGstQIODeviceSrc::setURI(const char *location, GError **err)
++ {
++ Q_ASSERT(QLatin1StringView(location).startsWith("qiodevice:/"_L1));
++
++ {
++ std::lock_guard guard{ *this };
++ 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 });
++
++ {
++ std::lock_guard guard{ *this };
++ 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);
++ std::lock_guard guard{ *src };
++ 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