qgst_bus_observer.cpp

Go to the documentation of this file.

// 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 "qgst_bus_observer_p.h"
 
QT_BEGIN_NAMESPACE
 
QGstBusObserver::QGstBusObserver(QGstBusHandle bus)
    : QGstBusHandle{
          std::move(bus),
      }
{
    if (!get())
        return;
 
    GPollFD pollFd{};
    gst_bus_get_pollfd(get(), &pollFd);
    Q_ASSERT(pollFd.fd);
 
#ifndef Q_OS_WIN
    m_socketNotifier.setSocket(pollFd.fd);
 
    QObject::connect(&m_socketNotifier, &QSocketNotifier::activated, &m_socketNotifier,
                     [this](QSocketDescriptor, QSocketNotifier::Type) {
        this->processAllPendingMessages();
    });
 
    m_socketNotifier.setEnabled(true);
#else
    m_socketNotifier.setHandle(reinterpret_cast<Qt::HANDLE>(pollFd.fd));
 
    QObject::connect(&m_socketNotifier, &QWinEventNotifier::activated, &m_socketNotifier,
                     [this](QWinEventNotifier::HANDLE) {
        this->processAllPendingMessages();
    });
    m_socketNotifier.setEnabled(true);
#endif
 
}
 
QGstBusObserver::~QGstBusObserver()
{
    close();
}
 
void QGstBusObserver::close()
{
    if (!get())
        return;
 
    QGstBusHandle::reset();
}
 
void QGstBusObserver::installMessageFilter(QGstreamerBusMessageFilter *filter)
{
    Q_ASSERT(filter);
    if (!busFilters.contains(filter))
        busFilters.append(filter);
}
 
void QGstBusObserver::removeMessageFilter(QGstreamerBusMessageFilter *filter)
{
    Q_ASSERT(filter);
    busFilters.removeAll(filter);
}
 
bool QGstBusObserver::processNextPendingMessage(GstMessageType type,
                                                std::optional<std::chrono::nanoseconds> timeout)
{
    if (!get())
        return false;
 
    GstClockTime gstTimeout = [&]() -> GstClockTime {
        if (!timeout)
            return GST_CLOCK_TIME_NONE; // block forever
        return timeout->count();
    }();
 
    QGstreamerMessage message{
        gst_bus_timed_pop_filtered(get(), gstTimeout, type),
        QGstreamerMessage::HasRef,
    };
    if (!message)
        return false;
 
    for (QGstreamerBusMessageFilter *filter : std::as_const(busFilters)) {
        if (filter->processBusMessage(message))
            break;
    }
 
    return true;
}
 
bool QGstBusObserver::currentThreadIsNotifierThread() const
{
    return m_socketNotifier.thread()->isCurrentThread();
}
 
void QGstBusObserver::processAllPendingMessages()
{
    for (;;) {
        bool messageHandled = processNextPendingMessage(GST_MESSAGE_ANY, std::chrono::nanoseconds{ 0 });
 
        if (!messageHandled)
            return;
    }
}
 
QT_END_NAMESPACE