Remote source: corrected worker handling

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;
-}
+}

plugins/channeltx/remotesource/remotesourceworker.cpp

@@ -19,9 +19,9 @@
 #include <channel/remotedataqueue.h>
 #include <algorithm>
 
-#include <QUdpSocket>
+#include <QThread>
-#include "cm256cc/cm256.h"
 
+#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<QAbstractSocket::SocketError>::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);
+    QMutexLocker mutexLocker(&m_mutex);
-    m_socket->setSocketOption(QAbstractSocket::ReceiveBufferSizeSocketOption, getDataSocketBufferSize(m_sampleRate));
+    m_socket.setSocketOption(QAbstractSocket::ReceiveBufferSizeSocketOption, getDataSocketBufferSize(m_sampleRate));
-    m_running = false;
+    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;
+    QMutexLocker mutexLocker(&m_mutex);
-    m_socket = nullptr;
+    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());
-
+            disconnect(&m_socket, SIGNAL(readyRead()), this, SLOT(readPendingDatagrams()));
-            if (m_socket)
+            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;
                 }
             }

plugins/channeltx/remotesource/remotesourceworker.h

@@ -20,13 +20,13 @@
 
 #include <QObject>
 #include <QHostAddress>
+#include <QUdpSocket>
 
 #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();
 };