Connecting microphone audio and sending to speaker and udp network simultaneously

[johnco3]

I need to capture audio from the microphone QAudioSource and forward it to both the speakers QAudioSink and simultaneously over the network wrapped in RTP headers. I am not sure how to do this.

Right now I reliably play synthesized audio tones to the speaker from a custom QIODevice (std::unique_ptr<IODeviceSink> mpIODeviceSink opened in readOnly mode) containing preformatted audio tone data (sine waves). I based this work off of the audio output sample that ships with QT 6.x. Once the QAudioSink is opened (QWindowsAudioSink in my case) it regularly requests data reads of variable sized buffers from the mpIODeviceSink (contaning the tone data). The Sink is opened by default in "pull" mode using a QTimer in the QAudioSink under the covers. I presume this timer sees when the speaker runs out of data and keeps requesting more the mpIODeviceSink to keep the audio from underruns.

Also within this same worker thread, I have a QAudioSource (microphone default input device) feeding another QIODevice (***std::unique_ptr<IODeviceSource> mpIODeviceSource which is *** opened in writeOnly mode). Whenever the microphone captures audio, it writes it mpIODeviceSource (using the following connect call (this is based off of the audio input example with Qt 6.x) - I have a custom audioCaptured signal that I emit from mpIODeviceSource whenever data is captured from the microphone allowing me to send it to the network - right now I just log it).

I was thinking that I could use an intermediate QIODevice (like a QBuffer or a subclass of it) opened in read/write mode that could share its buffers between the microphone (input) and the speaker & ideally the UDPSocket as outputs. (1 input to 2 outputs). I have no idea if this is the correct approach, of if I need to subclass a QAudioSink & associate that with the UDPSocket IODevice in order to transmit fixed size UDP frames to a remote RTP listener device.

The tricky thing is that these UDP frames are fixed size datagrams and should transmitted at regular intervals to maintain good audio.

    connect(mpIODeviceSource.get(), &IODeviceSource::audioCaptured,
        this, &RtpWorker::handleAudioCaptured, Qt::UniqueConnection);

Here is the RtpWorker class in question

#pragma once

// SYSTEM INCLUDES
#include <QTimer>
#include <QThread>
#include <QUdpSocket>
#include <QAudioSink>
#include <QAudioSource>
#include <spdlog/common.h>

// APPLICATION INCLUDES
#include "rtp/rtp.h"
#include "IODeviceSink.h"
#include "IODeviceSource.h"

// DEFINES
// EXTERNAL FUNCTIONS
// EXTERNAL VARIABLES
// CONSTANTS
// STRUCTS
// TYPEDEFS
// FORWARD DECLARATIONS
class QTimer;
class QTimerEvent;
class QUdpSocket;
class QAudioSource;
class QAudioBuffer;
class QAudioSink;

const auto gHeaderLenLambda = [](const QByteArray& rRawHeader)->qsizetype {
    const auto header = reinterpret_cast<const rtp_hdr_t*>(rRawHeader.constData());
    qsizetype headerLen = sizeof(rtp_hdr_t) + header->cc * sizeof(uint32_t);
    // header extension
    if (header->x != 0u) {
        const auto ext = reinterpret_cast<const rtp_hdr_ext_t*>(
            rRawHeader.constData() + headerLen);
        const auto ext_len = ntohs(ext->len);
        headerLen += static_cast<qsizetype>(
            sizeof(uint32_t) + (ext_len * sizeof(uint32_t)));
    }
    // must have sufficient bytes in raw header
    if (headerLen > rRawHeader.size()) {
        return -1;
    }
    return headerLen;
};


/**
 * Active Object QT pattern - see tutorial https://www.youtube.com/watch?v=SncJ3D-fO7g
 * See also: https://forum.qt.io/topic/140465/qt-6-x-qaudiosink-potential-bug
 */
class RtpWorker final : public QObject {
    Q_OBJECT
public:
    //! mpThread cannot have parent, as we need to move this object
    //! Also, note most QT class members need 'this' as parent for
    //! thread affinity to work correctly.
    RtpWorker();

    //! these  deleted constructors and assignment operators are not
    //! strictly required as the QObject superclass prevents copy
    //! but left in for clarity
    RtpWorker(const RtpWorker&) = delete;
    RtpWorker(RtpWorker&&) noexcept = delete;
    RtpWorker& operator=(const RtpWorker&) = delete;
    RtpWorker& operator=(RtpWorker&&) noexcept = delete;

    //! virtual destructor overload
    ~RtpWorker() override {
        QMetaObject::invokeMethod(this, "cleanup");
        mpThread->wait();
    }
Q_SIGNALS:
    void stop();
    void start(const qint64, const QString&, const quint16,
        const qint64, const QByteArray&, const QByteArrayList&);
    void datagramsReady(const QByteArrayList&);
    void bytesWritten(qint64, bool);
    void watchdogStatusChanged(bool, const QString&);

    /**
     * Signal to asynchronously notify application thread .
     *
     * @param rStatusText [in] Status area notification text.
     * @param rStyleSheet [in] Text stylesheet.
     * @param timeout     [in] Timeout after which the text
     *                    disappears.
     */
    void workerStatusChange(const QString& rStatusText,
        const QString& rStyleSheet, int timeout = 0) const;

    /**
     * IPC Signal from worker thread used to safely update
     * mTXBufferList.
     *
     * @param aTSIncrement [in] Timestamp increment per datagram.
     * @param rAudioData   [in] raw multichannel audio.
     */
    void updateAudio(const qint64 aTSIncrement, const QByteArrayList& rAudioData);

    //! Add log entry
    void addLogEntry(const spdlog::level::level_enum, const QString&);

    // slots are all called while in the worker thread context
private Q_SLOTS:
    /**
     * Start signal handler.
     *
     * <p>This slot runs in the worker thread context.  Creates a
     * buffered list of transmit UDP datagrams.  Each datagram is prefixed
     * with an RTP header (including a sequence# & timestamp).
     *
     * <p>The first datagram will always have the marker bit set.
     *
     * @param aTSIncrement   [in] Timestamp increment per datagram.
     * @param rTargetIP      [in] Target IP address.
     * @param aTargetPort    [in] Target UDP port.
     * @param aDurationUsecs
     *                       [in] Duration (specified in microseconds) to
     *                       sleep between recurring timer callbacks.
     *                       This time represents the duration for all
     *                       <code>mTXBufferList</code> datagrams.
     * @param rHeader        [in] RTP header consisting of a QByteArray
     *                       containing an rtp_hdr_t with optional contributing
     *                       sources (up to 16) & rtp_hdr_ext_t fields & data if
     *                       the x bit of the rtp_hdr_t is set.
     * @param rAudioData     [in] raw multichannel audio.
     */
    void handleStart(
        const qint64 aTSIncrement,
        const QString& rTargetIP,
        const quint16 aTargetPort,
        const qint64 aDurationUsecs,
        const QByteArray& rHeader,
        const QByteArrayList& rAudioData);

    /**
     * Slot to handle captured audio data.
     *
     * <p>Saves the audio data to a QIODevice.
     *
     * @param rAudioBuffer [in] audio data with (containing valid
     *                     QAudioFormat).
     */
    void handleAudioCaptured(const QAudioBuffer& rAudioBuffer);

    /**
     * Audio updated slot handler.
     *
     * @param aTSIncrement [in] Timestamp increment per datagram.
     * @param rAudioData   [in] raw multichannel audio.
     */
    void handleAudioUpdated(
        const qint64 aTSIncrement,
        const QByteArrayList& rAudioData);

    /**
     * QUdpSocket slot readyRead handler.
     *
     * <p>Slot called when UDP data returned via echo to this
     * thread.  When called drain any pending datagrams and pet the
     * inactivity watchdog timer.
     */
    void readyRead();

    /**
     * Slot to handle bytes written.
     *
     * <p>Called when bytes successfully sent to target where we
     * emit the bytesWritten signal to update the stats area in the
     * GUI.
     *
     * @param aBytesWritten [in] number of bytes written to socket.
     */
    void handleBytesWritten(qint64 aBytesWritten) {
        // forward to GUI
        Q_EMIT bytesWritten(aBytesWritten, false);
    }

    /**
     * Stop slot handler.
     *
     * <p>This is called in the worker thread context.  Close open
     * timers, sockets, QAudioSources and QAudioSinks.
     */
    void handleStop();


    /**
     * Cleanup method - runs in worker thread context.
     */
    void cleanup();

    /**
     * Timer callback event handler.
     *
     * @param event [in] timer event.
     */
    void timerEvent(QTimerEvent* event) override;
private:
    std::unique_ptr<QThread> mpThread;
    std::unique_ptr<QBasicTimer> mpTimer;
    std::unique_ptr<QTimer> mpWatchdog;
    std::unique_ptr<QUdpSocket> mpSocket;
    //! Audio Source (microphone).
    std::unique_ptr<QAudioSource> mpAudioSource;
    //! IODevice connected to mpAudioSource
    std::unique_ptr<IODeviceSource> mpIODeviceSource;
    //! Audio Sink (speaker).
    std::unique_ptr<QAudioSink> mpAudioSink;
    //! IODevice connected to mpAudioSink
    std::unique_ptr<IODeviceSink> mpIODeviceSink;

    QByteArrayList mTXBufferList;
    QHostAddress mTargetIP;
    quint16 mTargetPort;
    qint64 mTSIncrement;
    qint64 mTXCounter;
};