Revision control
Copy as Markdown
// Copyright (C) 2025 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 "qpipewire_async_support_p.h"
#include <QtMultimedia/private/qpipewire_instance_p.h>
QT_BEGIN_NAMESPACE
namespace QtPipeWire {
namespace {
template <typename Closure>
auto runWithEventLoopLock(Closure &&closure)
{
auto instance = QPipeWireInstance::instance();
if (instance)
return instance->runWithEventLoopLock(std::forward<Closure>(closure));
else
return closure();
}
} // namespace
// SpaListenerBase
static std::atomic_int s_sequenceNumberAllocator;
SpaListenerBase::SpaListenerBase()
: m_sequenceNumber(s_sequenceNumberAllocator.fetch_add(1, std::memory_order_relaxed))
{
}
void SpaListenerBase::removeHooks()
{
spa_hook_remove(&m_listenerHook);
}
///////////////////////////////////////////////////////////////////////////////////////////////////
// NodeEventListener
NodeEventListener::NodeEventListener(PwNodeHandle node, NodeHandler handler)
: m_node(std::move(node)), m_handler(std::move(handler))
{
static constexpr struct pw_node_events nodeEvents{
.version = PW_VERSION_NODE_EVENTS,
.info = NodeEventListener::onInfo,
.param = NodeEventListener::onParam,
};
int status = pw_node_add_listener(m_node.get(), &m_listenerHook, &nodeEvents, this);
if (status < 0)
qFatal() << "Failed to add listener: " << make_error_code(-status).message();
}
NodeEventListener::~NodeEventListener()
{
removeHooks();
runWithEventLoopLock([&] {
m_node = {};
});
}
void NodeEventListener::enumParams(spa_param_type type)
{
int status = pw_node_enum_params(m_node.get(), m_sequenceNumber, type, 0, 0, nullptr);
if (status < 0)
qCritical() << "pw_node_enum_params failed:" << make_error_code(-status).message();
}
void NodeEventListener::onInfo(void *data, const pw_node_info *info)
{
NodeEventListener *self = reinterpret_cast<NodeEventListener *>(data);
if (self->m_handler.infoHandler)
self->m_handler.infoHandler(info);
}
void NodeEventListener::onParam(void *data, int seq, uint32_t id, uint32_t index, uint32_t next,
const spa_pod *param)
{
NodeEventListener *self = reinterpret_cast<NodeEventListener *>(data);
if (self->m_handler.paramHandler)
self->m_handler.paramHandler(seq, id, index, next, param);
}
///////////////////////////////////////////////////////////////////////////////////////////////////
// CoreEventListener
CoreEventListener::CoreEventListener()
{
coreEvents.version = PW_VERSION_CORE_EVENTS;
}
CoreEventListener::~CoreEventListener()
{
runWithEventLoopLock([&] {
removeHooks();
});
}
///////////////////////////////////////////////////////////////////////////////////////////////////
// CoreEventDoneListener
CoreEventDoneListener::CoreEventDoneListener()
{
coreEvents.done = [](void *self, uint32_t id, int seq) {
Q_ASSERT(QPipeWireInstance::instance()->isInPwThreadLoop());
CoreEventDoneListener *listener = reinterpret_cast<CoreEventDoneListener *>(self);
if (id == PW_ID_CORE && listener->m_seqnum == seq) {
listener->m_seqnum = -1;
if (auto handler = std::exchange(listener->m_handler, {}))
handler();
}
};
}
q23::expected<void, int> CoreEventDoneListener::asyncWait(pw_core *coreConnection,
std::function<void()> handler)
{
m_handler = std::move(handler);
return runWithEventLoopLock([&]() -> q23::expected<void, int> {
int status = pw_core_add_listener(coreConnection, &m_listenerHook, &coreEvents, this);
if (status < 0) {
qFatal() << "pw_core_add_listener failed" << make_error_code(-status).message();
return q23::unexpected(status);
}
Q_ASSERT(m_seqnum == -1);
status = pw_core_sync(coreConnection, PW_ID_CORE, 0);
if (status < 0)
return q23::unexpected(status);
m_seqnum = status;
return {};
});
}
///////////////////////////////////////////////////////////////////////////////////////////////////
CoreEventSyncHelper::CoreEventSyncHelper() = default;
q23::expected<bool, int> CoreEventSyncHelper::sync(pw_core *coreConnection,
std::optional<std::chrono::nanoseconds> timeout)
{
auto voidOrError = CoreEventDoneListener::asyncWait(coreConnection, [&] {
m_semaphore.release();
});
if (voidOrError) {
if (timeout)
return m_semaphore.try_acquire_for(*timeout);
m_semaphore.acquire();
return true;
}
return voidOrError.error();
}
} // namespace QtPipeWire
QT_END_NAMESPACE