AIS and Radiosonde: implemented new message pipes. Part of #1154

plugins/channelrx/demodais/aisdemod.cpp

@@ -163,18 +163,15 @@ bool AISDemod::handleMessage(const Message& cmd)
             getMessageQueueToGUI()->push(msg);
         }
 
-        MessagePipesLegacy& messagePipes = MainCore::instance()->getMessagePipesLegacy();
+        QList<ObjectPipe*> aisPipes;
+        MainCore::instance()->getMessagePipes().getMessagePipes(this, "ais", aisPipes);
 
         // Forward to AIS feature
-        QList<MessageQueue*> *aisMessageQueues = messagePipes.getMessageQueues(this, "ais");
+        for (const auto& pipe : aisPipes)
-        if (aisMessageQueues)
-        {
-            QList<MessageQueue*>::iterator it = aisMessageQueues->begin();
-            for (; it != aisMessageQueues->end(); ++it)
         {
+            MessageQueue *messageQueue = qobject_cast<MessageQueue*>(pipe->m_element);
             MainCore::MsgPacket *msg = MainCore::MsgPacket::create(this, report.getMessage(), report.getDateTime());
-                (*it)->push(msg);
+            messageQueue->push(msg);
-            }
         }
 
         // Forward via UDP

plugins/channelrx/demodais/aisdemodgui.cpp

@@ -709,8 +709,8 @@ void AISDemodGUI::on_logOpen_clicked()
                     bool cancelled = false;
                     QStringList cols;
 
-                    MessagePipesLegacy& messagePipes = MainCore::instance()->getMessagePipesLegacy();
+                    QList<ObjectPipe*> aisPipes;
-                    QList<MessageQueue*> *aisMessageQueues = messagePipes.getMessageQueues(m_aisDemod, "ais");
+                    MainCore::instance()->getMessagePipes().getMessagePipes(this, "ais", aisPipes);
 
                     while (!cancelled && CSV::readRow(in, &cols))
                     {
@@ -725,14 +725,11 @@ void AISDemodGUI::on_logOpen_clicked()
                             messageReceived(bytes, dateTime);
 
                             // Forward to AIS feature
-                            if (aisMessageQueues)
+                            for (const auto& pipe : aisPipes)
-                            {
-                                QList<MessageQueue*>::iterator it = aisMessageQueues->begin();
-                                for (; it != aisMessageQueues->end(); ++it)
                             {
+                                MessageQueue *messageQueue = qobject_cast<MessageQueue*>(pipe->m_element);
                                 MainCore::MsgPacket *msg = MainCore::MsgPacket::create(m_aisDemod, bytes, dateTime);
-                                    (*it)->push(msg);
+                                messageQueue->push(msg);
-                                }
                             }
 
                             if (count % 1000 == 0)

plugins/channelrx/demodradiosonde/radiosondedemod.cpp

@@ -178,18 +178,15 @@ bool RadiosondeDemod::handleMessage(const Message& cmd)
             getMessageQueueToGUI()->push(msg);
         }
 
-        MessagePipesLegacy& messagePipes = MainCore::instance()->getMessagePipesLegacy();
-
         // Forward to Radiosonde feature
-        QList<MessageQueue*> *radiosondeMessageQueues = messagePipes.getMessageQueues(this, "radiosonde");
+        QList<ObjectPipe*> radiosondePipes;
-        if (radiosondeMessageQueues)
+        MainCore::instance()->getMessagePipes().getMessagePipes(this, "radiosonde", radiosondePipes);
-        {
+
-            QList<MessageQueue*>::iterator it = radiosondeMessageQueues->begin();
+        for (const auto& pipe : radiosondePipes)
-            for (; it != radiosondeMessageQueues->end(); ++it)
         {
+            MessageQueue *messageQueue = qobject_cast<MessageQueue*>(pipe->m_element);
             MainCore::MsgPacket *msg = MainCore::MsgPacket::create(this, report.getMessage(), report.getDateTime());
-                (*it)->push(msg);
+            messageQueue->push(msg);
-            }
         }
 
         // Forward via UDP

plugins/channelrx/demodradiosonde/radiosondedemodgui.cpp

@@ -846,8 +846,8 @@ void RadiosondeDemodGUI::on_logOpen_clicked()
                     bool cancelled = false;
                     QStringList cols;
 
-                    MessagePipesLegacy& framePipes = MainCore::instance()->getMessagePipesLegacy();
+                    QList<ObjectPipe*> radiosondePipes;
-                    QList<MessageQueue*> *radiosondeMessageQueues = framePipes.getMessageQueues(m_radiosondeDemod, "radiosonde");
+                    MainCore::instance()->getMessagePipes().getMessagePipes(this, "radiosonde", radiosondePipes);
 
                     while (!cancelled && CSV::readRow(in, &cols))
                     {
@@ -862,14 +862,11 @@ void RadiosondeDemodGUI::on_logOpen_clicked()
                             frameReceived(bytes, dateTime, 0, 0);
 
                             // Forward to Radiosonde feature
-                            if (radiosondeMessageQueues)
+                            for (const auto& pipe : radiosondePipes)
-                            {
-                                QList<MessageQueue*>::iterator it = radiosondeMessageQueues->begin();
-                                for (; it != radiosondeMessageQueues->end(); ++it)
                             {
+                                MessageQueue *messageQueue = qobject_cast<MessageQueue*>(pipe->m_element);
                                 MainCore::MsgPacket *msg = MainCore::MsgPacket::create(m_radiosondeDemod, bytes, dateTime);
-                                    (*it)->push(msg);
+                                messageQueue->push(msg);
-                                }
                             }
 
                             if (count % 100 == 0)

plugins/feature/ais/ais.cpp

@@ -20,7 +20,6 @@
 #include <QNetworkAccessManager>
 #include <QNetworkReply>
 #include <QBuffer>
-#include <QTimer>
 
 #include "SWGFeatureSettings.h"
 #include "SWGFeatureReport.h"
@@ -48,8 +47,6 @@ AIS::AIS(WebAPIAdapterInterface *webAPIAdapterInterface) :
     setObjectName(m_featureId);
     m_state = StIdle;
     m_errorMessage = "AIS error";
-    connect(&m_updatePipesTimer, SIGNAL(timeout()), this, SLOT(updatePipes()));
-    m_updatePipesTimer.start(1000);
     m_networkManager = new QNetworkAccessManager();
     QObject::connect(
         m_networkManager,
@@ -57,10 +54,13 @@ AIS::AIS(WebAPIAdapterInterface *webAPIAdapterInterface) :
         this,
         &AIS::networkManagerFinished
     );
+    scanAvailableChannels();
+    connect(MainCore::instance(), SIGNAL(channelAdded(int, ChannelAPI*)), this, SLOT(handleChannelAdded(int, ChannelAPI*)));
 }
 
 AIS::~AIS()
 {
+    disconnect(MainCore::instance(), SIGNAL(channelAdded(int, ChannelAPI*)), this, SLOT(handleChannelAdded(int, ChannelAPI*)));
     QObject::disconnect(
         m_networkManager,
         &QNetworkAccessManager::finished,
@@ -108,15 +108,6 @@ bool AIS::handleMessage(const Message& cmd)
     }
 }
 
-void AIS::updatePipes()
-{
-    QList<AvailablePipeSource> availablePipes = updateAvailablePipeSources("ais", AISSettings::m_pipeTypes, AISSettings::m_pipeURIs, this);
-
-    if (availablePipes != m_availablePipes) {
-        m_availablePipes = availablePipes;
-    }
-}
-
 QByteArray AIS::serialize() const
 {
     return m_settings.serialize();
@@ -335,3 +326,92 @@ void AIS::networkManagerFinished(QNetworkReply *reply)
 
     reply->deleteLater();
 }
+
+void AIS::scanAvailableChannels()
+{
+    MainCore *mainCore = MainCore::instance();
+    MessagePipes& messagePipes = mainCore->getMessagePipes();
+    std::vector<DeviceSet*>& deviceSets = mainCore->getDeviceSets();
+    m_availableChannels.clear();
+
+    for (const auto& deviceSet : deviceSets)
+    {
+        DSPDeviceSourceEngine *deviceSourceEngine =  deviceSet->m_deviceSourceEngine;
+
+        if (deviceSourceEngine)
+        {
+            for (int chi = 0; chi < deviceSet->getNumberOfChannels(); chi++)
+            {
+                ChannelAPI *channel = deviceSet->getChannelAt(chi);
+
+                if ((channel->getURI() == "sdrangel.channel.aisdemod") && !m_availableChannels.contains(channel))
+                {
+                    qDebug("AIS::scanAvailableChannels: register %d:%d (%p)", deviceSet->getIndex(), chi, channel);
+                    ObjectPipe *pipe = messagePipes.registerProducerToConsumer(channel, this, "ais");
+                    MessageQueue *messageQueue = qobject_cast<MessageQueue*>(pipe->m_element);
+                    QObject::connect(
+                        messageQueue,
+                        &MessageQueue::messageEnqueued,
+                        this,
+                        [=](){ this->handleChannelMessageQueue(messageQueue); },
+                        Qt::QueuedConnection
+                    );
+                    connect(pipe, SIGNAL(toBeDeleted(int, QObject*)), this, SLOT(handleMessagePipeToBeDeleted(int, QObject*)));
+                    m_availableChannels.insert(channel);
+                }
+            }
+        }
+    }
+}
+
+void AIS::handleChannelAdded(int deviceSetIndex, ChannelAPI *channel)
+{
+    qDebug("AIS::handleChannelAdded: deviceSetIndex: %d:%d channel: %s (%p)",
+        deviceSetIndex, channel->getIndexInDeviceSet(), qPrintable(channel->getURI()), channel);
+    std::vector<DeviceSet*>& deviceSets = MainCore::instance()->getDeviceSets();
+    DeviceSet *deviceSet = deviceSets[deviceSetIndex];
+    DSPDeviceSourceEngine *deviceSourceEngine =  deviceSet->m_deviceSourceEngine;
+
+    if (deviceSourceEngine && (channel->getURI() == "sdrangel.channel.aisdemod"))
+    {
+        if (!m_availableChannels.contains(channel))
+        {
+            MessagePipes& messagePipes = MainCore::instance()->getMessagePipes();
+            ObjectPipe *pipe = messagePipes.registerProducerToConsumer(channel, this, "ais");
+            MessageQueue *messageQueue = qobject_cast<MessageQueue*>(pipe->m_element);
+            QObject::connect(
+                messageQueue,
+                &MessageQueue::messageEnqueued,
+                this,
+                [=](){ this->handleChannelMessageQueue(messageQueue); },
+                Qt::QueuedConnection
+            );
+            connect(pipe, SIGNAL(toBeDeleted(int, QObject*)), this, SLOT(handleMessagePipeToBeDeleted(int, QObject*)));
+            m_availableChannels.insert(channel);
+        }
+    }
+}
+
+void AIS::handleMessagePipeToBeDeleted(int reason, QObject* object)
+{
+    if (reason == 0) // producer (channel)
+    {
+        if (m_availableChannels.contains((ChannelAPI*) object))
+        {
+            qDebug("AIS::handleMessagePipeToBeDeleted: removing channel at (%p)", object);
+            m_availableChannels.remove((ChannelAPI*) object);
+        }
+    }
+}
+
+void AIS::handleChannelMessageQueue(MessageQueue* messageQueue)
+{
+    Message* message;
+
+    while ((message = messageQueue->pop()) != nullptr)
+    {
+        if (handleMessage(*message)) {
+            delete message;
+        }
+    }
+}

plugins/feature/ais/ais.h

@@ -21,7 +21,7 @@
 
 #include <QThread>
 #include <QNetworkRequest>
-#include <QTimer>
+#include <QSet>
 
 #include "feature/feature.h"
 #include "util/message.h"
@@ -97,8 +97,7 @@ public:
 
 private:
     AISSettings m_settings;
-    QList<PipeEndPoint::AvailablePipeSource> m_availablePipes;
+    QSet<ChannelAPI*> m_availableChannels;
-    QTimer m_updatePipesTimer;
 
     QNetworkAccessManager *m_networkManager;
     QNetworkRequest m_networkRequest;
@@ -107,10 +106,13 @@ private:
     void stop();
     void applySettings(const AISSettings& settings, bool force = false);
     void webapiReverseSendSettings(QList<QString>& featureSettingsKeys, const AISSettings& settings, bool force);
+    void scanAvailableChannels();
 
 private slots:
-    void updatePipes();
     void networkManagerFinished(QNetworkReply *reply);
+    void handleChannelAdded(int deviceSetIndex, ChannelAPI *channel);
+    void handleMessagePipeToBeDeleted(int reason, QObject* object);
+    void handleChannelMessageQueue(MessageQueue* messageQueue);
 };
 
 #endif // INCLUDE_FEATURE_AIS_H_

plugins/feature/radiosonde/radiosonde.cpp

@@ -20,7 +20,6 @@
 #include <QNetworkAccessManager>
 #include <QNetworkReply>
 #include <QBuffer>
-#include <QTimer>
 
 #include "SWGFeatureSettings.h"
 #include "SWGFeatureReport.h"
@@ -48,8 +47,6 @@ Radiosonde::Radiosonde(WebAPIAdapterInterface *webAPIAdapterInterface) :
     setObjectName(m_featureId);
     m_state = StIdle;
     m_errorMessage = "Radiosonde error";
-    connect(&m_updatePipesTimer, SIGNAL(timeout()), this, SLOT(updatePipes()));
-    m_updatePipesTimer.start(1000);
     m_networkManager = new QNetworkAccessManager();
     QObject::connect(
         m_networkManager,
@@ -57,10 +54,13 @@ Radiosonde::Radiosonde(WebAPIAdapterInterface *webAPIAdapterInterface) :
         this,
         &Radiosonde::networkManagerFinished
     );
+    scanAvailableChannels();
+    connect(MainCore::instance(), SIGNAL(channelAdded(int, ChannelAPI*)), this, SLOT(handleChannelAdded(int, ChannelAPI*)));
 }
 
 Radiosonde::~Radiosonde()
 {
+    disconnect(MainCore::instance(), SIGNAL(channelAdded(int, ChannelAPI*)), this, SLOT(handleChannelAdded(int, ChannelAPI*)));
     QObject::disconnect(
         m_networkManager,
         &QNetworkAccessManager::finished,
@@ -108,15 +108,6 @@ bool Radiosonde::handleMessage(const Message& cmd)
     }
 }
 
-void Radiosonde::updatePipes()
-{
-    QList<AvailablePipeSource> availablePipes = updateAvailablePipeSources("radiosonde", RadiosondeSettings::m_pipeTypes, RadiosondeSettings::m_pipeURIs, this);
-
-    if (availablePipes != m_availablePipes) {
-        m_availablePipes = availablePipes;
-    }
-}
-
 QByteArray Radiosonde::serialize() const
 {
     return m_settings.serialize();
@@ -335,3 +326,92 @@ void Radiosonde::networkManagerFinished(QNetworkReply *reply)
 
     reply->deleteLater();
 }
+
+void Radiosonde::scanAvailableChannels()
+{
+    MainCore *mainCore = MainCore::instance();
+    MessagePipes& messagePipes = mainCore->getMessagePipes();
+    std::vector<DeviceSet*>& deviceSets = mainCore->getDeviceSets();
+    m_availableChannels.clear();
+
+    for (const auto& deviceSet : deviceSets)
+    {
+        DSPDeviceSourceEngine *deviceSourceEngine =  deviceSet->m_deviceSourceEngine;
+
+        if (deviceSourceEngine)
+        {
+            for (int chi = 0; chi < deviceSet->getNumberOfChannels(); chi++)
+            {
+                ChannelAPI *channel = deviceSet->getChannelAt(chi);
+
+                if ((channel->getURI() == "sdrangel.channel.radiosondedemod") && !m_availableChannels.contains(channel))
+                {
+                    qDebug("Radiosonde::scanAvailableChannels: register %d:%d (%p)", deviceSet->getIndex(), chi, channel);
+                    ObjectPipe *pipe = messagePipes.registerProducerToConsumer(channel, this, "radiosonde");
+                    MessageQueue *messageQueue = qobject_cast<MessageQueue*>(pipe->m_element);
+                    QObject::connect(
+                        messageQueue,
+                        &MessageQueue::messageEnqueued,
+                        this,
+                        [=](){ this->handleChannelMessageQueue(messageQueue); },
+                        Qt::QueuedConnection
+                    );
+                    connect(pipe, SIGNAL(toBeDeleted(int, QObject*)), this, SLOT(handleMessagePipeToBeDeleted(int, QObject*)));
+                    m_availableChannels.insert(channel);
+                }
+            }
+        }
+    }
+}
+
+void Radiosonde::handleChannelAdded(int deviceSetIndex, ChannelAPI *channel)
+{
+    qDebug("Radiosonde::handleChannelAdded: deviceSetIndex: %d:%d channel: %s (%p)",
+        deviceSetIndex, channel->getIndexInDeviceSet(), qPrintable(channel->getURI()), channel);
+    std::vector<DeviceSet*>& deviceSets = MainCore::instance()->getDeviceSets();
+    DeviceSet *deviceSet = deviceSets[deviceSetIndex];
+    DSPDeviceSourceEngine *deviceSourceEngine =  deviceSet->m_deviceSourceEngine;
+
+    if (deviceSourceEngine && (channel->getURI() == "sdrangel.channel.radiosondedemod"))
+    {
+        if (!m_availableChannels.contains(channel))
+        {
+            MessagePipes& messagePipes = MainCore::instance()->getMessagePipes();
+            ObjectPipe *pipe = messagePipes.registerProducerToConsumer(channel, this, "radiosonde");
+            MessageQueue *messageQueue = qobject_cast<MessageQueue*>(pipe->m_element);
+            QObject::connect(
+                messageQueue,
+                &MessageQueue::messageEnqueued,
+                this,
+                [=](){ this->handleChannelMessageQueue(messageQueue); },
+                Qt::QueuedConnection
+            );
+            connect(pipe, SIGNAL(toBeDeleted(int, QObject*)), this, SLOT(handleMessagePipeToBeDeleted(int, QObject*)));
+            m_availableChannels.insert(channel);
+        }
+    }
+}
+
+void Radiosonde::handleMessagePipeToBeDeleted(int reason, QObject* object)
+{
+    if (reason == 0) // producer (channel)
+    {
+        if (m_availableChannels.contains((ChannelAPI*) object))
+        {
+            qDebug("Radiosonde::handleMessagePipeToBeDeleted: removing channel at (%p)", object);
+            m_availableChannels.remove((ChannelAPI*) object);
+        }
+    }
+}
+
+void Radiosonde::handleChannelMessageQueue(MessageQueue* messageQueue)
+{
+    Message* message;
+
+    while ((message = messageQueue->pop()) != nullptr)
+    {
+        if (handleMessage(*message)) {
+            delete message;
+        }
+    }
+}

plugins/feature/radiosonde/radiosonde.h

@@ -21,7 +21,7 @@
 
 #include <QThread>
 #include <QNetworkRequest>
-#include <QTimer>
+#include <QSet>
 
 #include "feature/feature.h"
 #include "util/message.h"
@@ -97,8 +97,7 @@ public:
 
 private:
     RadiosondeSettings m_settings;
-    QList<PipeEndPoint::AvailablePipeSource> m_availablePipes;
+    QSet<ChannelAPI*> m_availableChannels;
-    QTimer m_updatePipesTimer;
 
     QNetworkAccessManager *m_networkManager;
     QNetworkRequest m_networkRequest;
@@ -107,10 +106,13 @@ private:
     void stop();
     void applySettings(const RadiosondeSettings& settings, bool force = false);
     void webapiReverseSendSettings(QList<QString>& featureSettingsKeys, const RadiosondeSettings& settings, bool force);
+    void scanAvailableChannels();
 
 private slots:
-    void updatePipes();
     void networkManagerFinished(QNetworkReply *reply);
+    void handleChannelAdded(int deviceSetIndex, ChannelAPI *channel);
+    void handleMessagePipeToBeDeleted(int reason, QObject* object);
+    void handleChannelMessageQueue(MessageQueue* messageQueue);
 };
 
 #endif // INCLUDE_FEATURE_RADIOSONDE_H_