diff --git a/plugins/channeltx/daemonsrc/CMakeLists.txt b/plugins/channeltx/daemonsrc/CMakeLists.txt index 6cb7e1acd..158eda183 100644 --- a/plugins/channeltx/daemonsrc/CMakeLists.txt +++ b/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 ) diff --git a/plugins/channeltx/daemonsrc/daemonsrc.cpp b/plugins/channeltx/daemonsrc/daemonsrc.cpp index 694a258dc..798b98018 100644 --- a/plugins/channeltx/daemonsrc/daemonsrc.cpp +++ b/plugins/channeltx/daemonsrc/daemonsrc.cpp @@ -14,22 +14,35 @@ // along with this program. If not, see . // /////////////////////////////////////////////////////////////////////////////////// +#include +#include + #include +#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; - sample.m_imag = 0.0f; + m_dataReadQueue.readSample(sample); } 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); +} diff --git a/plugins/channeltx/daemonsrc/daemonsrc.h b/plugins/channeltx/daemonsrc/daemonsrc.h index 743313b92..8d0dd9a1e 100644 --- a/plugins/channeltx/daemonsrc/daemonsrc.h +++ b/plugins/channeltx/daemonsrc/daemonsrc.h @@ -14,20 +14,50 @@ // along with this program. If not, see . // /////////////////////////////////////////////////////////////////////////////////// +#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(); }; diff --git a/plugins/channeltx/daemonsrc/daemonsrcgui.cpp b/plugins/channeltx/daemonsrc/daemonsrcgui.cpp index 7e5074d55..9d2212ed9 100644 --- a/plugins/channeltx/daemonsrc/daemonsrcgui.cpp +++ b/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(); +} diff --git a/plugins/channeltx/daemonsrc/daemonsrcgui.h b/plugins/channeltx/daemonsrc/daemonsrcgui.h index f8d87bcd6..1c55506bb 100644 --- a/plugins/channeltx/daemonsrc/daemonsrcgui.h +++ b/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); }; diff --git a/plugins/channeltx/daemonsrc/daemonsrcthread.cpp b/plugins/channeltx/daemonsrc/daemonsrcthread.cpp new file mode 100644 index 000000000..bf2b7c176 --- /dev/null +++ b/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 . // +/////////////////////////////////////////////////////////////////////////////////// + +#include + +#include +#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"); + } + } +} + diff --git a/plugins/channeltx/daemonsrc/daemonsrcthread.h b/plugins/channeltx/daemonsrc/daemonsrcthread.h new file mode 100644 index 000000000..dc11e1a76 --- /dev/null +++ b/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 . // +/////////////////////////////////////////////////////////////////////////////////// + +#ifndef PLUGINS_CHANNELTX_DAEMONSRC_DAEMONSRCTHREAD_H_ +#define PLUGINS_CHANNELTX_DAEMONSRC_DAEMONSRCTHREAD_H_ + +#include +#include +#include +#include + +#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_ */