From 3cb22c8b8a837f41d5389d37bf9b62ee8c958d6f Mon Sep 17 00:00:00 2001 From: f4exb Date: Tue, 7 Dec 2021 05:58:17 +0100 Subject: [PATCH] Remote source: corrected worker handling --- .../remotesource/remotesourcesource.cpp | 8 ++- .../remotesource/remotesourceworker.cpp | 65 +++++++++++++------ .../remotesource/remotesourceworker.h | 10 ++- 3 files changed, 58 insertions(+), 25 deletions(-) diff --git a/plugins/channeltx/remotesource/remotesourcesource.cpp b/plugins/channeltx/remotesource/remotesourcesource.cpp index 4b43a7a8a..9c8f86f82 100644 --- a/plugins/channeltx/remotesource/remotesourcesource.cpp +++ b/plugins/channeltx/remotesource/remotesourcesource.cpp @@ -35,7 +35,9 @@ RemoteSourceSource::RemoteSourceSource() : } RemoteSourceSource::~RemoteSourceSource() -{} +{ + stop(); +} void RemoteSourceSource::pull(SampleVector::iterator begin, unsigned int nbSamples) { @@ -105,7 +107,7 @@ void RemoteSourceSource::stop() { stopWorker(); m_sourceWorker->deleteLater(); - m_sourceWorker = 0; + m_sourceWorker = nullptr; } m_running = false; @@ -276,4 +278,4 @@ void RemoteSourceSource::applyChannelSettings(int channelSampleRate, bool force) } m_channelSampleRate = channelSampleRate; -} \ No newline at end of file +} diff --git a/plugins/channeltx/remotesource/remotesourceworker.cpp b/plugins/channeltx/remotesource/remotesourceworker.cpp index 19141d85c..cdace534f 100644 --- a/plugins/channeltx/remotesource/remotesourceworker.cpp +++ b/plugins/channeltx/remotesource/remotesourceworker.cpp @@ -19,9 +19,9 @@ #include #include -#include -#include "cm256cc/cm256.h" +#include +#include "cm256cc/cm256.h" #include "remotesourceworker.h" MESSAGE_CLASS_DEFINITION(RemoteSourceWorker::MsgDataBind, Message) @@ -31,11 +31,18 @@ RemoteSourceWorker::RemoteSourceWorker(RemoteDataQueue *dataQueue, QObject* pare m_running(false), m_dataQueue(dataQueue), m_address(QHostAddress::LocalHost), - m_socket(nullptr), + m_socket(this), + m_mutex(QMutex::Recursive), m_sampleRate(0) { std::fill(m_dataBlocks, m_dataBlocks+4, (RemoteDataBlock *) 0); connect(&m_inputMessageQueue, SIGNAL(messageEnqueued()), this, SLOT(handleInputMessages()), Qt::QueuedConnection); + connect(&m_socket, SIGNAL(readyRead()),this, SLOT(recv())); +#if QT_VERSION < QT_VERSION_CHECK(5, 15, 0) + connect(&m_socket, QOverload::of(&QAbstractSocket::error), this, &APRSWorker::errorOccurred); +#else + connect(&m_socket, &QAbstractSocket::errorOccurred, this, &RemoteSourceWorker::errorOccurred); +#endif } RemoteSourceWorker::~RemoteSourceWorker() @@ -49,20 +56,43 @@ void RemoteSourceWorker::dataBind(const QString& address, uint16_t port) m_inputMessageQueue.push(msg); } -void RemoteSourceWorker::startWork() +bool RemoteSourceWorker::startWork() { qDebug("RemoteSourceWorker::startWork"); - m_socket = new QUdpSocket(this); - m_socket->setSocketOption(QAbstractSocket::ReceiveBufferSizeSocketOption, getDataSocketBufferSize(m_sampleRate)); - m_running = false; + QMutexLocker mutexLocker(&m_mutex); + m_socket.setSocketOption(QAbstractSocket::ReceiveBufferSizeSocketOption, getDataSocketBufferSize(m_sampleRate)); + connect(&m_inputMessageQueue, SIGNAL(messageEnqueued()), this, SLOT(handleInputMessages())); + connect(thread(), SIGNAL(started()), this, SLOT(started())); + connect(thread(), SIGNAL(finished()), this, SLOT(finished())); + m_running = true; + return m_running; +} + +void RemoteSourceWorker::started() +{ + disconnect(thread(), SIGNAL(started()), this, SLOT(started())); } void RemoteSourceWorker::stopWork() { qDebug("RemoteSourceWorker::stopWork"); - delete m_socket; - m_socket = nullptr; + QMutexLocker mutexLocker(&m_mutex); + disconnect(&m_inputMessageQueue, SIGNAL(messageEnqueued()), this, SLOT(handleInputMessages())); +} + +void RemoteSourceWorker::finished() +{ + // Close any existing connection + if (m_socket.isOpen()) { + m_socket.close(); + } m_running = false; + disconnect(thread(), SIGNAL(finished()), this, SLOT(finished())); +} + +void RemoteSourceWorker::errorOccurred(QAbstractSocket::SocketError socketError) +{ + qWarning() << "RemoteSourceWorker::errorOccurred: " << socketError; } void RemoteSourceWorker::handleInputMessages() @@ -73,15 +103,12 @@ void RemoteSourceWorker::handleInputMessages() { if (MsgDataBind::match(*message)) { + QMutexLocker mutexLocker(&m_mutex); MsgDataBind* notif = (MsgDataBind*) message; qDebug("RemoteSourceWorker::handleInputMessages: MsgDataBind: %s:%d", qPrintable(notif->getAddress().toString()), notif->getPort()); - - if (m_socket) - { - disconnect(m_socket, SIGNAL(readyRead()), this, SLOT(readPendingDatagrams())); - m_socket->bind(notif->getAddress(), notif->getPort()); - connect(m_socket, SIGNAL(readyRead()), this, SLOT(readPendingDatagrams())); - } + disconnect(&m_socket, SIGNAL(readyRead()), this, SLOT(readPendingDatagrams())); + m_socket.bind(notif->getAddress(), notif->getPort()); + connect(&m_socket, SIGNAL(readyRead()), this, SLOT(readPendingDatagrams())); } } } @@ -91,12 +118,12 @@ void RemoteSourceWorker::readPendingDatagrams() RemoteSuperBlock superBlock; qint64 size; - while (m_socket->hasPendingDatagrams()) + while (m_socket.hasPendingDatagrams()) { QHostAddress sender; quint16 senderPort = 0; //qint64 pendingDataSize = m_socket->pendingDatagramSize(); - size = m_socket->readDatagram((char *) &superBlock, (long long int) sizeof(RemoteSuperBlock), &sender, &senderPort); + size = m_socket.readDatagram((char *) &superBlock, (long long int) sizeof(RemoteSuperBlock), &sender, &senderPort); if (size == sizeof(RemoteSuperBlock)) { @@ -111,7 +138,7 @@ void RemoteSourceWorker::readPendingDatagrams() if (m_sampleRate != sampleRate) { qDebug("RemoteSourceWorker::readPendingDatagrams: sampleRate: %u", sampleRate); - m_socket->setSocketOption(QAbstractSocket::ReceiveBufferSizeSocketOption, getDataSocketBufferSize(sampleRate)); + m_socket.setSocketOption(QAbstractSocket::ReceiveBufferSizeSocketOption, getDataSocketBufferSize(sampleRate)); m_sampleRate = sampleRate; } } diff --git a/plugins/channeltx/remotesource/remotesourceworker.h b/plugins/channeltx/remotesource/remotesourceworker.h index b953f3d27..a5ee5375d 100644 --- a/plugins/channeltx/remotesource/remotesourceworker.h +++ b/plugins/channeltx/remotesource/remotesourceworker.h @@ -20,13 +20,13 @@ #include #include +#include #include "util/message.h" #include "util/messagequeue.h" class RemoteDataQueue; class RemoteDataBlock; -class QUdpSocket; class RemoteSourceWorker : public QObject { Q_OBJECT @@ -57,7 +57,7 @@ public: RemoteSourceWorker(RemoteDataQueue *dataQueue, QObject* parent = 0); ~RemoteSourceWorker(); - void startWork(); + bool startWork(); void stopWork(); void dataBind(const QString& address, uint16_t port); @@ -68,7 +68,8 @@ private: RemoteDataQueue *m_dataQueue; QHostAddress m_address; - QUdpSocket *m_socket; + QUdpSocket m_socket; + QMutex m_mutex; static const uint32_t m_nbDataBlocks = 4; //!< number of data blocks in the ring buffer RemoteDataBlock *m_dataBlocks[m_nbDataBlocks]; //!< ring buffer of data blocks indexed by frame affinity @@ -77,6 +78,9 @@ private: static int getDataSocketBufferSize(uint32_t inSampleRate); private slots: + void started(); + void finished(); + void errorOccurred(QAbstractSocket::SocketError socketError); void handleInputMessages(); void readPendingDatagrams(); };