Revision control
Copy as Markdown
// 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