Daemon channel source new plugin (3)

plugins/channeltx/daemonsrc/CMakeLists.txt

@@ -4,7 +4,7 @@ set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -std=c++11")
 
 set(daemonsrc_SOURCES
 	daemonsrc.cpp
-#	daemonsrcthread.cpp
+	daemonsrcthread.cpp
 	daemonsrcgui.cpp
 	daemonsrcplugin.cpp
 	daemonsrcsettings.cpp
@@ -12,7 +12,7 @@ set(daemonsrc_SOURCES
 
 set(daemonsrc_HEADERS
 	daemonsrc.h
-#	daemonsrcthread.h
+	daemonsrcthread.h
 	daemonsrcgui.h
 	daemonsrcplugin.h
 	daemonsrcsettings.h
@@ -25,7 +25,7 @@ set(daemonsrc_FORMS
 include_directories(
 	.
 	${CMAKE_CURRENT_BINARY_DIR}
-#	${CMAKE_SOURCE_DIR}/sdrdaemon
+	${CMAKE_SOURCE_DIR}/sdrdaemon
 	${CM256CC_INCLUDE_DIR}
 	${CMAKE_SOURCE_DIR}/swagger/sdrangel/code/qt5/client
 )
@@ -47,7 +47,7 @@ target_link_libraries(daemonsrc
 	${QT_LIBRARIES}
 	${CM256CC_LIBRARIES}
 	sdrbase
-#	sdrdaemon
+	sdrdaemon
 	sdrgui
 	swagger
 )

plugins/channeltx/daemonsrc/daemonsrc.cpp

@@ -14,22 +14,35 @@
 // along with this program. If not, see <http://www.gnu.org/licenses/>.          //
 ///////////////////////////////////////////////////////////////////////////////////
 
+#include <sys/time.h>
+#include <unistd.h>
+
 #include <QDebug>
 
+#include "SWGChannelSettings.h"
+#include "SWGChannelReport.h"
+#include "SWGSDRDaemonChannelSourceReport.h"
+
 #include "device/devicesinkapi.h"
 #include "dsp/upchannelizer.h"
 #include "dsp/threadedbasebandsamplesource.h"
 
+#include "daemonsrcthread.h"
 #include "daemonsrc.h"
 
 MESSAGE_CLASS_DEFINITION(DaemonSrc::MsgSampleRateNotification, Message)
+MESSAGE_CLASS_DEFINITION(DaemonSrc::MsgConfigureDaemonSrc, Message)
 
 const QString DaemonSrc::m_channelIdURI = "sdrangel.channeltx.daemonsrc";
 const QString DaemonSrc::m_channelId ="DaemonSrc";
 
 DaemonSrc::DaemonSrc(DeviceSinkAPI *deviceAPI) :
     ChannelSourceAPI(m_channelIdURI),
-    m_deviceAPI(deviceAPI)
+    m_deviceAPI(deviceAPI),
+    m_sourceThread(0),
+    m_running(false),
+    m_nbCorrectableErrors(0),
+    m_nbUncorrectableErrors(0)
 {
     setObjectName(m_channelId);
 
@@ -37,6 +50,10 @@ DaemonSrc::DaemonSrc(DeviceSinkAPI *deviceAPI) :
     m_threadedChannelizer = new ThreadedBasebandSampleSource(m_channelizer, this);
     m_deviceAPI->addThreadedSource(m_threadedChannelizer);
     m_deviceAPI->addChannelAPI(this);
+
+    connect(&m_dataQueue, SIGNAL(dataBlockEnqueued()), this, SLOT(handleData()), Qt::QueuedConnection);
+    m_cm256p = m_cm256.isInitialized() ? &m_cm256 : 0;
+    m_currentMeta.init();
 }
 
 DaemonSrc::~DaemonSrc()
@@ -49,8 +66,7 @@ DaemonSrc::~DaemonSrc()
 
 void DaemonSrc::pull(Sample& sample)
 {
-    sample.m_real = 0.0f;
+    m_dataReadQueue.readSample(sample);
-    sample.m_imag = 0.0f;
 }
 
 void DaemonSrc::pullAudio(int nbSamples __attribute__((unused)))
@@ -59,10 +75,40 @@ void DaemonSrc::pullAudio(int nbSamples __attribute__((unused)))
 
 void DaemonSrc::start()
 {
+    qDebug("DaemonSrc::start");
+
+    if (m_running) {
+        stop();
+    }
+
+    m_sourceThread = new DaemonSrcThread(&m_dataQueue);
+    m_sourceThread->startStop(true);
+    m_sourceThread->dataBind(m_settings.m_dataAddress, m_settings.m_dataPort);
+    m_running = true;
 }
 
 void DaemonSrc::stop()
 {
+    qDebug("DaemonSrc::stop");
+
+    if (m_sourceThread != 0)
+    {
+        m_sourceThread->startStop(false);
+        m_sourceThread->deleteLater();
+        m_sourceThread = 0;
+    }
+
+    m_running = false;
+}
+
+void DaemonSrc::setDataLink(const QString& dataAddress, uint16_t dataPort)
+{
+    DaemonSrcSettings settings = m_settings;
+    settings.m_dataAddress = dataAddress;
+    settings.m_dataPort = dataPort;
+
+    MsgConfigureDaemonSrc *msg = MsgConfigureDaemonSrc::create(settings, false);
+    m_inputMessageQueue.push(msg);
 }
 
 bool DaemonSrc::handleMessage(const Message& cmd)
@@ -83,6 +129,14 @@ bool DaemonSrc::handleMessage(const Message& cmd)
 
         return true;
     }
+    else if (MsgConfigureDaemonSrc::match(cmd))
+    {
+        MsgConfigureDaemonSrc& cfg = (MsgConfigureDaemonSrc&) cmd;
+        qDebug() << "MsgConfigureDaemonSrc::handleMessage: MsgConfigureDaemonSrc";
+        applySettings(cfg.getSettings(), cfg.getForce());
+
+        return true;
+    }
     return false;
 }
 
@@ -95,11 +149,158 @@ bool DaemonSrc::deserialize(const QByteArray& data __attribute__((unused)))
 {
     if (m_settings.deserialize(data))
     {
+        MsgConfigureDaemonSrc *msg = MsgConfigureDaemonSrc::create(m_settings, true);
+        m_inputMessageQueue.push(msg);
         return true;
     }
     else
     {
         m_settings.resetToDefaults();
+        MsgConfigureDaemonSrc *msg = MsgConfigureDaemonSrc::create(m_settings, true);
+        m_inputMessageQueue.push(msg);
         return false;
     }
 }
+
+void DaemonSrc::applySettings(const DaemonSrcSettings& settings, bool force)
+{
+    qDebug() << "DaemonSrc::applySettings:"
+            << " m_dataAddress: " << settings.m_dataAddress
+            << " m_dataPort: " << settings.m_dataPort
+            << " force: " << force;
+
+    bool change = false;
+
+    if ((m_settings.m_dataAddress != settings.m_dataAddress) || force) {
+        change = true;
+    }
+
+    if ((m_settings.m_dataPort != settings.m_dataPort) || force) {
+        change = true;
+    }
+
+    if (change && m_sourceThread) {
+        m_sourceThread->dataBind(settings.m_dataAddress, settings.m_dataPort);
+    }
+
+    m_settings = settings;
+}
+
+void DaemonSrc::handleDataBlock(SDRDaemonDataBlock* dataBlock __attribute__((unused)))
+{
+}
+
+void DaemonSrc::handleData()
+{
+    SDRDaemonDataBlock* dataBlock;
+
+    while (m_running && ((dataBlock = m_dataQueue.pop()) != 0)) {
+        handleDataBlock(dataBlock);
+    }
+}
+
+void DaemonSrc::printMeta(const QString& header, SDRDaemonMetaDataFEC *metaData)
+{
+    qDebug().noquote() << header << ": "
+            << "|" << metaData->m_centerFrequency
+            << ":" << metaData->m_sampleRate
+            << ":" << (int) (metaData->m_sampleBytes & 0xF)
+            << ":" << (int) metaData->m_sampleBits
+            << ":" << (int) metaData->m_nbOriginalBlocks
+            << ":" << (int) metaData->m_nbFECBlocks
+            << "|" << metaData->m_tv_sec
+            << ":" << metaData->m_tv_usec
+            << "|";
+}
+
+uint32_t DaemonSrc::calculateDataReadQueueSize(int sampleRate)
+{
+    // scale for 20 blocks at 48 kS/s. Take next even number.
+    uint32_t maxSize = sampleRate / 2400;
+    maxSize = (maxSize % 2 == 0) ? maxSize : maxSize + 1;
+    qDebug("SDRDaemonChannelSource::calculateDataReadQueueSize: set max queue size to %u blocks", maxSize);
+    return maxSize;
+}
+
+int DaemonSrc::webapiSettingsGet(
+        SWGSDRangel::SWGChannelSettings& response,
+        QString& errorMessage __attribute__((unused)))
+{
+    response.setSdrDaemonChannelSourceSettings(new SWGSDRangel::SWGSDRDaemonChannelSourceSettings());
+    response.getSdrDaemonChannelSourceSettings()->init();
+    webapiFormatChannelSettings(response, m_settings);
+    return 200;
+}
+
+int DaemonSrc::webapiSettingsPutPatch(
+        bool force,
+        const QStringList& channelSettingsKeys,
+        SWGSDRangel::SWGChannelSettings& response,
+        QString& errorMessage __attribute__((unused)))
+{
+    DaemonSrcSettings settings = m_settings;
+
+    if (channelSettingsKeys.contains("dataAddress")) {
+        settings.m_dataAddress = *response.getSdrDaemonChannelSourceSettings()->getDataAddress();
+    }
+
+    if (channelSettingsKeys.contains("dataPort"))
+    {
+        int dataPort = response.getSdrDaemonChannelSourceSettings()->getDataPort();
+
+        if ((dataPort < 1024) || (dataPort > 65535)) {
+            settings.m_dataPort = 9090;
+        } else {
+            settings.m_dataPort = dataPort;
+        }
+    }
+
+    MsgConfigureDaemonSrc *msg = MsgConfigureDaemonSrc::create(settings, force);
+    m_inputMessageQueue.push(msg);
+
+    qDebug("DaemonSrc::webapiSettingsPutPatch: forward to GUI: %p", m_guiMessageQueue);
+    if (m_guiMessageQueue) // forward to GUI if any
+    {
+        MsgConfigureDaemonSrc *msgToGUI = MsgConfigureDaemonSrc::create(settings, force);
+        m_guiMessageQueue->push(msgToGUI);
+    }
+
+    webapiFormatChannelSettings(response, settings);
+
+    return 200;
+}
+
+int DaemonSrc::webapiReportGet(
+        SWGSDRangel::SWGChannelReport& response,
+        QString& errorMessage __attribute__((unused)))
+{
+    response.setSdrDaemonChannelSourceReport(new SWGSDRangel::SWGSDRDaemonChannelSourceReport());
+    response.getSdrDaemonChannelSourceReport()->init();
+    webapiFormatChannelReport(response);
+    return 200;
+}
+
+void DaemonSrc::webapiFormatChannelSettings(SWGSDRangel::SWGChannelSettings& response, const DaemonSrcSettings& settings)
+{
+    if (response.getSdrDaemonChannelSourceSettings()->getDataAddress()) {
+        *response.getSdrDaemonChannelSourceSettings()->getDataAddress() = settings.m_dataAddress;
+    } else {
+        response.getSdrDaemonChannelSourceSettings()->setDataAddress(new QString(settings.m_dataAddress));
+    }
+
+    response.getSdrDaemonChannelSourceSettings()->setDataPort(settings.m_dataPort);
+}
+
+void DaemonSrc::webapiFormatChannelReport(SWGSDRangel::SWGChannelReport& response)
+{
+    struct timeval tv;
+    gettimeofday(&tv, 0);
+
+    response.getSdrDaemonChannelSourceReport()->setTvSec(tv.tv_sec);
+    response.getSdrDaemonChannelSourceReport()->setTvUSec(tv.tv_usec);
+    response.getSdrDaemonChannelSourceReport()->setQueueSize(m_dataReadQueue.size());
+    response.getSdrDaemonChannelSourceReport()->setQueueLength(m_dataReadQueue.length());
+    response.getSdrDaemonChannelSourceReport()->setSamplesCount(m_dataReadQueue.readSampleCount());
+    response.getSdrDaemonChannelSourceReport()->setCorrectableErrorsCount(m_nbCorrectableErrors);
+    response.getSdrDaemonChannelSourceReport()->setUncorrectableErrorsCount(m_nbUncorrectableErrors);
+}

plugins/channeltx/daemonsrc/daemonsrc.h

@@ -14,20 +14,50 @@
 // along with this program. If not, see <http://www.gnu.org/licenses/>.          //
 ///////////////////////////////////////////////////////////////////////////////////
 
+#include "cm256.h"
+
 #include "dsp/basebandsamplesource.h"
 #include "channel/channelsourceapi.h"
 #include "util/message.h"
 
 #include "daemonsrcsettings.h"
+#include "channel/sdrdaemondataqueue.h"
+#include "channel/sdrdaemondatablock.h"
+#include "channel/sdrdaemondatareadqueue.h"
 
-class DeviceSinkAPI;
 class ThreadedBasebandSampleSource;
 class UpChannelizer;
+class DeviceSinkAPI;
+class DaemonSrcThread;
+class SDRDaemonDataBlock;
 
 class DaemonSrc : public BasebandSampleSource, public ChannelSourceAPI {
     Q_OBJECT
 
 public:
+    class MsgConfigureDaemonSrc : public Message {
+        MESSAGE_CLASS_DECLARATION
+
+    public:
+        const DaemonSrcSettings& getSettings() const { return m_settings; }
+        bool getForce() const { return m_force; }
+
+        static MsgConfigureDaemonSrc* create(const DaemonSrcSettings& settings, bool force)
+        {
+            return new MsgConfigureDaemonSrc(settings, force);
+        }
+
+    private:
+        DaemonSrcSettings m_settings;
+        bool m_force;
+
+        MsgConfigureDaemonSrc(const DaemonSrcSettings& settings, bool force) :
+            Message(),
+            m_settings(settings),
+            m_force(force)
+        { }
+    };
+
     class MsgSampleRateNotification : public Message {
         MESSAGE_CLASS_DECLARATION
 
@@ -66,6 +96,22 @@ public:
     virtual QByteArray serialize() const;
     virtual bool deserialize(const QByteArray& data);
 
+    virtual int webapiSettingsGet(
+            SWGSDRangel::SWGChannelSettings& response,
+            QString& errorMessage);
+
+    virtual int webapiSettingsPutPatch(
+            bool force,
+            const QStringList& channelSettingsKeys,
+            SWGSDRangel::SWGChannelSettings& response,
+            QString& errorMessage);
+
+    virtual int webapiReportGet(
+            SWGSDRangel::SWGChannelReport& response,
+            QString& errorMessage);
+
+    void setDataLink(const QString& dataAddress, uint16_t dataPort);
+
     static const QString m_channelIdURI;
     static const QString m_channelId;
 
@@ -73,6 +119,29 @@ private:
     DeviceSinkAPI* m_deviceAPI;
     ThreadedBasebandSampleSource* m_threadedChannelizer;
     UpChannelizer* m_channelizer;
+    SDRDaemonDataQueue m_dataQueue;
+    DaemonSrcThread *m_sourceThread;
+    CM256 m_cm256;
+    CM256 *m_cm256p;
+    bool m_running;
 
     DaemonSrcSettings m_settings;
+
+    CM256::cm256_block   m_cm256DescriptorBlocks[2*SDRDaemonNbOrginalBlocks]; //!< CM256 decoder descriptors (block addresses and block indexes)
+    SDRDaemonMetaDataFEC m_currentMeta;
+
+    SDRDaemonDataReadQueue m_dataReadQueue;
+
+    uint32_t m_nbCorrectableErrors;   //!< count of correctable errors in number of blocks
+    uint32_t m_nbUncorrectableErrors; //!< count of uncorrectable errors in number of blocks
+
+    void applySettings(const DaemonSrcSettings& settings, bool force = false);
+    void handleDataBlock(SDRDaemonDataBlock *dataBlock);
+    void printMeta(const QString& header, SDRDaemonMetaDataFEC *metaData);
+    uint32_t calculateDataReadQueueSize(int sampleRate);
+    void webapiFormatChannelSettings(SWGSDRangel::SWGChannelSettings& response, const DaemonSrcSettings& settings);
+    void webapiFormatChannelReport(SWGSDRangel::SWGChannelReport& response);
+
+private slots:
+    void handleData();
 };

plugins/channeltx/daemonsrc/daemonsrcgui.cpp

@@ -130,8 +130,15 @@ void DaemonSrcGUI::blockApplySettings(bool block)
     m_doApplySettings = !block;
 }
 
-void DaemonSrcGUI::applySettings(bool force __attribute((unused)))
+void DaemonSrcGUI::applySettings(bool force)
 {
+    if (m_doApplySettings)
+    {
+        setTitleColor(m_channelMarker.getColor());
+
+        DaemonSrc::MsgConfigureDaemonSrc* message = DaemonSrc::MsgConfigureDaemonSrc::create(m_settings, force);
+        m_daemonSrc->getInputMessageQueue()->push(message);
+    }
 }
 
 void DaemonSrcGUI::displaySettings()
@@ -193,3 +200,41 @@ void DaemonSrcGUI::onMenuDialogCalled(const QPoint &p)
 
     applySettings();
 }
+
+void DaemonSrcGUI::on_dataAddress_returnPressed()
+{
+    m_settings.m_dataAddress = ui->dataAddress->text();
+    applySettings();
+}
+
+void DaemonSrcGUI::on_dataPort_returnPressed()
+{
+    bool dataOk;
+    int dataPort = ui->dataPort->text().toInt(&dataOk);
+
+    if((!dataOk) || (dataPort < 1024) || (dataPort > 65535))
+    {
+        return;
+    }
+    else
+    {
+        m_settings.m_dataPort = dataPort;
+    }
+
+    applySettings();
+}
+
+void DaemonSrcGUI::on_dataApplyButton_clicked(bool checked __attribute__((unused)))
+{
+    m_settings.m_dataAddress = ui->dataAddress->text();
+
+    bool dataOk;
+    int udpDataPort = ui->dataPort->text().toInt(&dataOk);
+
+    if((dataOk) && (udpDataPort >= 1024) && (udpDataPort < 65535))
+    {
+        m_settings.m_dataPort = udpDataPort;
+    }
+
+    applySettings();
+}

plugins/channeltx/daemonsrc/daemonsrcgui.h

@@ -74,6 +74,9 @@ private:
 
 private slots:
     void handleSourceMessages();
+    void on_dataAddress_returnPressed();
+    void on_dataPort_returnPressed();
+    void on_dataApplyButton_clicked(bool checked);
     void onWidgetRolled(QWidget* widget, bool rollDown);
     void onMenuDialogCalled(const QPoint& p);
 };

plugins/channeltx/daemonsrc/daemonsrcthread.cpp

@@ -0,0 +1,189 @@
+///////////////////////////////////////////////////////////////////////////////////
+// Copyright (C) 2018 Edouard Griffiths, F4EXB                                   //
+//                                                                               //
+// This program is free software; you can redistribute it and/or modify          //
+// it under the terms of the GNU General Public License as published by          //
+// the Free Software Foundation as version 3 of the License, or                  //
+//                                                                               //
+// This program is distributed in the hope that it will be useful,               //
+// but WITHOUT ANY WARRANTY; without even the implied warranty of                //
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the                  //
+// GNU General Public License V3 for more details.                               //
+//                                                                               //
+// You should have received a copy of the GNU General Public License             //
+// along with this program. If not, see <http://www.gnu.org/licenses/>.          //
+///////////////////////////////////////////////////////////////////////////////////
+
+#include <algorithm>
+
+#include <QUdpSocket>
+#include "cm256.h"
+
+#include "channel/sdrdaemondataqueue.h"
+#include "channel/sdrdaemondatablock.h"
+#include "channel/sdrdaemonchannelsourcethread.h"
+
+#include "daemonsrcthread.h"
+
+MESSAGE_CLASS_DEFINITION(DaemonSrcThread::MsgStartStop, Message)
+MESSAGE_CLASS_DEFINITION(DaemonSrcThread::MsgDataBind, Message)
+
+DaemonSrcThread::DaemonSrcThread(SDRDaemonDataQueue *dataQueue, QObject* parent) :
+    QThread(parent),
+    m_running(false),
+    m_dataQueue(dataQueue),
+    m_address(QHostAddress::LocalHost),
+    m_socket(0)
+{
+    std::fill(m_dataBlocks, m_dataBlocks+4, (SDRDaemonDataBlock *) 0);
+    connect(&m_inputMessageQueue, SIGNAL(messageEnqueued()), this, SLOT(handleInputMessages()), Qt::QueuedConnection);
+}
+
+DaemonSrcThread::~DaemonSrcThread()
+{
+    qDebug("DaemonSrcThread::~SDRDaemonChannelSourceThread");
+}
+
+void DaemonSrcThread::startStop(bool start)
+{
+    MsgStartStop *msg = MsgStartStop::create(start);
+    m_inputMessageQueue.push(msg);
+}
+
+void DaemonSrcThread::dataBind(const QString& address, uint16_t port)
+{
+    MsgDataBind *msg = MsgDataBind::create(address, port);
+    m_inputMessageQueue.push(msg);
+}
+
+void SDRDaemonChannelSourceThread::startWork()
+{
+    qDebug("DaemonSrcThread::startWork");
+    m_startWaitMutex.lock();
+    m_socket = new QUdpSocket(this);
+    start();
+    while(!m_running)
+        m_startWaiter.wait(&m_startWaitMutex, 100);
+    m_startWaitMutex.unlock();
+}
+
+void DaemonSrcThread::stopWork()
+{
+    qDebug("DaemonSrcThread::stopWork");
+    delete m_socket;
+    m_socket = 0;
+    m_running = false;
+    wait();
+}
+
+void DaemonSrcThread::run()
+{
+    qDebug("DaemonSrcThread::run: begin");
+    m_running = true;
+    m_startWaiter.wakeAll();
+
+    while (m_running)
+    {
+        sleep(1); // Do nothing as everything is in the data handler (dequeuer)
+    }
+
+    m_running = false;
+    qDebug("DaemonSrcThread::run: end");
+}
+
+
+void DaemonSrcThread::handleInputMessages()
+{
+    Message* message;
+
+    while ((message = m_inputMessageQueue.pop()) != 0)
+    {
+        if (MsgStartStop::match(*message))
+        {
+            MsgStartStop* notif = (MsgStartStop*) message;
+            qDebug("DaemonSrcThread::handleInputMessages: MsgStartStop: %s", notif->getStartStop() ? "start" : "stop");
+
+            if (notif->getStartStop()) {
+                startWork();
+            } else {
+                stopWork();
+            }
+
+            delete message;
+        }
+        else if (MsgDataBind::match(*message))
+        {
+            MsgDataBind* notif = (MsgDataBind*) message;
+            qDebug("DaemonSrcThread::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()));
+            }
+        }
+    }
+}
+
+void DaemonSrcThread::readPendingDatagrams()
+{
+    SDRDaemonSuperBlock superBlock;
+    qint64 size;
+
+    while (m_socket->hasPendingDatagrams())
+    {
+        QHostAddress sender;
+        quint16 senderPort = 0;
+        //qint64 pendingDataSize = m_socket->pendingDatagramSize();
+        size = m_socket->readDatagram((char *) &superBlock, (long long int) sizeof(SDRDaemonSuperBlock), &sender, &senderPort);
+
+        if (size == sizeof(SDRDaemonSuperBlock))
+        {
+            unsigned int dataBlockIndex = superBlock.m_header.m_frameIndex % m_nbDataBlocks;
+
+            // create the first block for this index
+            if (m_dataBlocks[dataBlockIndex] == 0) {
+                m_dataBlocks[dataBlockIndex] = new SDRDaemonDataBlock();
+            }
+
+            if (m_dataBlocks[dataBlockIndex]->m_rxControlBlock.m_frameIndex < 0)
+            {
+                // initialize virgin block with the frame index
+                m_dataBlocks[dataBlockIndex]->m_rxControlBlock.m_frameIndex = superBlock.m_header.m_frameIndex;
+            }
+            else
+            {
+                // if the frame index is not the same for the same slot it means we are starting a new frame
+                uint32_t frameIndex = m_dataBlocks[dataBlockIndex]->m_rxControlBlock.m_frameIndex;
+
+                if (superBlock.m_header.m_frameIndex != frameIndex)
+                {
+                    //qDebug("SDRDaemonChannelSourceThread::readPendingDatagrams: push frame %u", frameIndex);
+                    m_dataQueue->push(m_dataBlocks[dataBlockIndex]);
+                    m_dataBlocks[dataBlockIndex] = new SDRDaemonDataBlock();
+                    m_dataBlocks[dataBlockIndex]->m_rxControlBlock.m_frameIndex = superBlock.m_header.m_frameIndex;
+                }
+            }
+
+            m_dataBlocks[dataBlockIndex]->m_superBlocks[superBlock.m_header.m_blockIndex] = superBlock;
+
+            if (superBlock.m_header.m_blockIndex == 0) {
+                m_dataBlocks[dataBlockIndex]->m_rxControlBlock.m_metaRetrieved = true;
+            }
+
+            if (superBlock.m_header.m_blockIndex < SDRDaemonNbOrginalBlocks) {
+                m_dataBlocks[dataBlockIndex]->m_rxControlBlock.m_originalCount++;
+            } else {
+                m_dataBlocks[dataBlockIndex]->m_rxControlBlock.m_recoveryCount++;
+            }
+
+            m_dataBlocks[dataBlockIndex]->m_rxControlBlock.m_blockCount++;
+        }
+        else
+        {
+            qWarning("SDRDaemonChannelSourceThread::readPendingDatagrams: wrong super block size not processing");
+        }
+    }
+}
+

plugins/channeltx/daemonsrc/daemonsrcthread.h

@@ -0,0 +1,109 @@
+///////////////////////////////////////////////////////////////////////////////////
+// Copyright (C) 2018 Edouard Griffiths, F4EXB                                   //
+//                                                                               //
+// This program is free software; you can redistribute it and/or modify          //
+// it under the terms of the GNU General Public License as published by          //
+// the Free Software Foundation as version 3 of the License, or                  //
+//                                                                               //
+// This program is distributed in the hope that it will be useful,               //
+// but WITHOUT ANY WARRANTY; without even the implied warranty of                //
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the                  //
+// GNU General Public License V3 for more details.                               //
+//                                                                               //
+// You should have received a copy of the GNU General Public License             //
+// along with this program. If not, see <http://www.gnu.org/licenses/>.          //
+///////////////////////////////////////////////////////////////////////////////////
+
+#ifndef PLUGINS_CHANNELTX_DAEMONSRC_DAEMONSRCTHREAD_H_
+#define PLUGINS_CHANNELTX_DAEMONSRC_DAEMONSRCTHREAD_H_
+
+#include <QThread>
+#include <QMutex>
+#include <QWaitCondition>
+#include <QHostAddress>
+
+#include "util/message.h"
+#include "util/messagequeue.h"
+
+class SDRDaemonDataQueue;
+class SDRDaemonDataBlock;
+class QUdpSocket;
+
+class DaemonSrcThread : public QThread {
+    Q_OBJECT
+public:
+    class MsgStartStop : public Message {
+        MESSAGE_CLASS_DECLARATION
+
+    public:
+        bool getStartStop() const { return m_startStop; }
+
+        static MsgStartStop* create(bool startStop) {
+            return new MsgStartStop(startStop);
+        }
+
+    protected:
+        bool m_startStop;
+
+        MsgStartStop(bool startStop) :
+            Message(),
+            m_startStop(startStop)
+        { }
+    };
+
+    class MsgDataBind : public Message {
+        MESSAGE_CLASS_DECLARATION
+
+    public:
+        QHostAddress getAddress() const { return m_address; }
+        uint16_t getPort() const { return m_port; }
+
+        static MsgDataBind* create(const QString& address, uint16_t port) {
+            return new MsgDataBind(address, port);
+        }
+
+    protected:
+        QHostAddress m_address;
+        uint16_t m_port;
+
+        MsgDataBind(const QString& address, uint16_t port) :
+            Message(),
+            m_port(port)
+        {
+            m_address.setAddress(address);
+        }
+    };
+
+    DaemonSrcThread(SDRDaemonDataQueue *dataQueue, QObject* parent = 0);
+    ~DaemonSrcThread();
+
+    void startStop(bool start);
+    void dataBind(const QString& address, uint16_t port);
+
+private:
+    QMutex m_startWaitMutex;
+    QWaitCondition m_startWaiter;
+    bool m_running;
+
+    MessageQueue m_inputMessageQueue;
+    SDRDaemonDataQueue *m_dataQueue;
+
+    QHostAddress m_address;
+    QUdpSocket *m_socket;
+
+    static const uint32_t m_nbDataBlocks = 4;          //!< number of data blocks in the ring buffer
+    SDRDaemonDataBlock *m_dataBlocks[m_nbDataBlocks];  //!< ring buffer of data blocks indexed by frame affinity
+
+    void startWork();
+    void stopWork();
+
+    void run();
+
+private slots:
+    void handleInputMessages();
+    void readPendingDatagrams();
+};
+
+
+
+#endif /* PLUGINS_CHANNELTX_DAEMONSRC_DAEMONSRCTHREAD_H_ */