Add PipeEndPoint class, that both ChannelAPI and Feature can inherit

from, to allow either to be used as pipe sources.
Add MsgMapItem, MsgPacket and MsgTargetAzimuthElevation messages that
can be passed between channels and features via pipes.

plugins/feature/vorlocalizer/vorlocalizer.cpp

@@ -244,8 +244,8 @@ bool VORLocalizer::handleMessage(const Message& cmd)
         qDebug() << "VORLocalizer::handleMessage: MsgReportChannelDeleted";
         MessagePipesCommon::MsgReportChannelDeleted& report = (MessagePipesCommon::MsgReportChannelDeleted&) cmd;
         const MessagePipesCommon::ChannelRegistrationKey& channelKey = report.getChannelRegistrationKey();
-        const ChannelAPI *channel = channelKey.m_key;
+        const PipeEndPoint *channel = channelKey.m_key;
-        m_availableChannels.remove(const_cast<ChannelAPI*>(channel));
+        m_availableChannels.remove(const_cast<ChannelAPI*>(reinterpret_cast<const ChannelAPI*>(channel)));
         updateChannels();
         MessageQueue *messageQueue = MainCore::instance()->getMessagePipes().unregisterChannelToFeature(channel, this, "report");
         disconnect(messageQueue, SIGNAL(messageEnqueued()), this, SLOT(handleChannelMessageQueue(MessageQueue*)));

sdrbase/CMakeLists.txt

@@ -167,6 +167,7 @@ set(sdrbase_SOURCES
     pipes/messagepipes.cpp
     pipes/messagepipescommon.cpp
     pipes/messagepipesgcworker.cpp
+    pipes/pipeendpoint.cpp
 
     settings/featuresetpreset.cpp
     settings/preferences.cpp
@@ -349,6 +350,7 @@ set(sdrbase_HEADERS
     pipes/messagepipes.h
     pipes/messagepipescommon.h
     pipes/messagepipesgcworker.h
+    pipes/pipeendpoint.h
 
     plugin/plugininterface.h
     plugin/pluginapi.h

sdrbase/channel/channelapi.cpp

@@ -19,6 +19,7 @@
 
 #include "util/uid.h"
 #include "channelapi.h"
+#include "maincore.h"
 
 ChannelAPI::ChannelAPI(const QString& uri, StreamType streamType) :
     m_streamType(streamType),

sdrbase/channel/channelapi.h

@@ -27,6 +27,7 @@
 #include <stdint.h>
 
 #include "export.h"
+#include "pipes/pipeendpoint.h"
 
 class DeviceAPI;
 
@@ -37,7 +38,7 @@ namespace SWGSDRangel
     class SWGChannelActions;
 }
 
-class SDRBASE_API ChannelAPI {
+class SDRBASE_API ChannelAPI : public PipeEndPoint {
 public:
     enum StreamType //!< This is the same enum as in PluginInterface
     {

sdrbase/feature/feature.cpp

@@ -23,6 +23,7 @@
 #include "SWGDeviceState.h"
 
 #include "feature.h"
+#include "maincore.h"
 
 Feature::Feature(const QString& uri, WebAPIAdapterInterface *webAPIAdapterInterface) :
 	m_webAPIAdapterInterface(webAPIAdapterInterface),
@@ -45,6 +46,18 @@ void Feature::handleInputMessages()
 	}
 }
 
+void Feature::handlePipeMessageQueue(MessageQueue* messageQueue)
+{
+    Message* message;
+
+    while ((message = messageQueue->pop()) != nullptr)
+    {
+        if (handleMessage(*message)) {
+            delete message;
+        }
+    }
+}
+
 int Feature::webapiRunGet(
 	SWGSDRangel::SWGDeviceState& response,
 	QString& errorMessage) const

sdrbase/feature/feature.h

@@ -25,6 +25,7 @@
 #include <QByteArray>
 
 #include "export.h"
+#include "pipes/pipeendpoint.h"
 #include "util/messagequeue.h"
 
 class WebAPIAdapterInterface;
@@ -40,7 +41,7 @@ namespace SWGSDRangel
     class SWGChannelSettings;
 }
 
-class SDRBASE_API Feature : public QObject {
+class SDRBASE_API Feature : public QObject, public PipeEndPoint {
     Q_OBJECT
 public:
     enum FeatureState {
@@ -155,6 +156,8 @@ protected:
 
 protected slots:
 	void handleInputMessages();
+    void handlePipeMessageQueue(MessageQueue* messageQueue);
+    friend PipeEndPoint;
 
 private:
     QString m_name; //!< Unique identifier in a device set used for sorting may change depending on relative position in device set

sdrbase/maincore.cpp

@@ -48,6 +48,9 @@ MESSAGE_CLASS_DEFINITION(MainCore::MsgDeleteFeature, Message)
 MESSAGE_CLASS_DEFINITION(MainCore::MsgChannelReport, Message)
 MESSAGE_CLASS_DEFINITION(MainCore::MsgChannelSettings, Message)
 MESSAGE_CLASS_DEFINITION(MainCore::MsgChannelDemodReport, Message)
+MESSAGE_CLASS_DEFINITION(MainCore::MsgMapItem, Message)
+MESSAGE_CLASS_DEFINITION(MainCore::MsgPacket, Message)
+MESSAGE_CLASS_DEFINITION(MainCore::MsgTargetAzimuthElevation, Message)
 
 MainCore::MainCore()
 {

sdrbase/maincore.h

@@ -28,10 +28,10 @@
 #include "util/message.h"
 #include "pipes/messagepipes.h"
 #include "pipes/datapipes.h"
+#include "channel/channelapi.h"
 
 class DeviceSet;
 class FeatureSet;
-class ChannelAPI;
 class Feature;
 class PluginManager;
 class MessageQueue;
@@ -44,6 +44,8 @@ namespace SWGSDRangel
 {
     class SWGChannelReport;
     class SWGChannelSettings;
+    class SWGMapItem;
+    class SWGTargetAzimuthElevation;
 }
 
 class SDRBASE_API MainCore
@@ -498,6 +500,89 @@ public:
         { }
     };
 
+    // Message to Map feature to display an item on the map
+    class SDRBASE_API MsgMapItem : public Message {
+        MESSAGE_CLASS_DECLARATION
+
+    public:
+        const PipeEndPoint *getPipeSource() const { return m_pipeSource; }
+        SWGSDRangel::SWGMapItem *getSWGMapItem() const { return m_swgMapItem; }
+
+        static MsgMapItem* create(const PipeEndPoint *pipeSource, SWGSDRangel::SWGMapItem *swgMapItem)
+        {
+            return new MsgMapItem(pipeSource, swgMapItem);
+        }
+
+    private:
+        const PipeEndPoint *m_pipeSource;
+        SWGSDRangel::SWGMapItem *m_swgMapItem;
+
+        MsgMapItem(const PipeEndPoint *pipeSource, SWGSDRangel::SWGMapItem *swgMapItem) :
+            Message(),
+            m_pipeSource(pipeSource),
+            m_swgMapItem(swgMapItem)
+        { }
+    };
+
+    // Message to pass received packets between channels and features
+    class SDRBASE_API MsgPacket : public Message {
+        MESSAGE_CLASS_DECLARATION
+
+    public:
+        const PipeEndPoint *getPipeSource() const { return m_pipeSource; }
+        QByteArray getPacket() const { return m_packet; }
+        QDateTime getDateTime() const { return m_dateTime; }
+
+        static MsgPacket* create(const PipeEndPoint *pipeSource, QByteArray packet)
+        {
+            return new MsgPacket(pipeSource, packet, QDateTime::currentDateTime());
+        }
+
+        static MsgPacket* create(const PipeEndPoint *pipeSource, QByteArray packet, QDateTime dateTime)
+        {
+            return new MsgPacket(pipeSource, packet, dateTime);
+        }
+
+    private:
+        const PipeEndPoint *m_pipeSource;
+        QByteArray m_packet;
+        QDateTime m_dateTime;
+
+        MsgPacket(const PipeEndPoint *pipeSource, QByteArray packet, QDateTime dateTime) :
+            Message(),
+            m_pipeSource(pipeSource),
+            m_packet(packet),
+            m_dateTime(dateTime)
+        {
+        }
+    };
+
+    // Message to pass target azimuth and elevation between channels & features
+    class SDRBASE_API MsgTargetAzimuthElevation : public Message {
+        MESSAGE_CLASS_DECLARATION
+
+    public:
+        const PipeEndPoint *getPipeSource() const { return m_pipeSource; }
+        SWGSDRangel::SWGTargetAzimuthElevation *getSWGTargetAzimuthElevation() const { return m_swgTargetAzimuthElevation; }
+
+        static MsgTargetAzimuthElevation* create(const PipeEndPoint *pipeSource, SWGSDRangel::SWGTargetAzimuthElevation *swgTargetAzimuthElevation)
+        {
+            return new MsgTargetAzimuthElevation(pipeSource, swgTargetAzimuthElevation);
+        }
+
+    private:
+        const PipeEndPoint *m_pipeSource;
+        SWGSDRangel::SWGTargetAzimuthElevation *m_swgTargetAzimuthElevation;
+
+        MsgTargetAzimuthElevation(const PipeEndPoint *pipeSource, SWGSDRangel::SWGTargetAzimuthElevation *swgTargetAzimuthElevation) :
+            Message(),
+            m_pipeSource(pipeSource),
+            m_swgTargetAzimuthElevation(swgTargetAzimuthElevation)
+        { }
+    };
+
+
+
 	MainCore();
 	~MainCore();
 	static MainCore *instance();

sdrbase/pipes/messagepipes.cpp

@@ -21,6 +21,7 @@
 
 #include "messagepipesgcworker.h"
 #include "messagepipes.h"
+#include "pipeendpoint.h"
 
 MessagePipes::MessagePipes()
 {
@@ -41,19 +42,19 @@ MessagePipes::~MessagePipes()
 	}
 }
 
-MessageQueue *MessagePipes::registerChannelToFeature(const ChannelAPI *source, Feature *feature, const QString& type)
+MessageQueue *MessagePipes::registerChannelToFeature(const PipeEndPoint *source, Feature *feature, const QString& type)
 {
 	return m_registrations.registerProducerToConsumer(source, feature, type);
 }
 
-MessageQueue *MessagePipes::unregisterChannelToFeature(const ChannelAPI *source, Feature *feature, const QString& type)
+MessageQueue *MessagePipes::unregisterChannelToFeature(const PipeEndPoint *source, Feature *feature, const QString& type)
 {
 	MessageQueue *messageQueue = m_registrations.unregisterProducerToConsumer(source, feature, type);
 	m_gcWorker->addMessageQueueToDelete(messageQueue);
 	return messageQueue;
 }
 
-QList<MessageQueue*>* MessagePipes::getMessageQueues(const ChannelAPI *source, const QString& type)
+QList<MessageQueue*>* MessagePipes::getMessageQueues(const PipeEndPoint *source, const QString& type)
 {
 	return m_registrations.getElements(source, type);
 }

sdrbase/pipes/messagepipes.h

@@ -29,7 +29,7 @@
 #include "messagepipescommon.h"
 #include "elementpipesregistrations.h"
 
-class ChannelAPI;
+class PipeEndPoint;
 class Feature;
 class MessagePipesGCWorker;
 class MessageQueue;
@@ -43,12 +43,12 @@ public:
     MessagePipes& operator=(const MessagePipes&) = delete;
     ~MessagePipes();
 
-    MessageQueue *registerChannelToFeature(const ChannelAPI *source, Feature *feature, const QString& type);
+    MessageQueue *registerChannelToFeature(const PipeEndPoint *source, Feature *feature, const QString& type);
-    MessageQueue *unregisterChannelToFeature(const ChannelAPI *source, Feature *feature, const QString& type);
+    MessageQueue *unregisterChannelToFeature(const PipeEndPoint *source, Feature *feature, const QString& type);
-    QList<MessageQueue*>* getMessageQueues(const ChannelAPI *source, const QString& type);
+    QList<MessageQueue*>* getMessageQueues(const PipeEndPoint *source, const QString& type);
 
 private:
-    ElementPipesRegistrations<ChannelAPI, Feature, MessageQueue> m_registrations;
+    ElementPipesRegistrations<PipeEndPoint, Feature, MessageQueue> m_registrations;
     QThread m_gcThread; //!< Garbage collector thread
     MessagePipesGCWorker *m_gcWorker; //!< Garbage collector
 

sdrbase/pipes/messagepipescommon.h

@@ -26,14 +26,14 @@
 #include "util/message.h"
 #include "elementpipescommon.h"
 
-class ChannelAPI;
+class PipeEndPoint;
 class Feature;
 class MessageQueue;
 
 class SDRBASE_API MessagePipesCommon
 {
 public:
-    typedef ElementPipesCommon::RegistrationKey<ChannelAPI> ChannelRegistrationKey;
+    typedef ElementPipesCommon::RegistrationKey<PipeEndPoint> ChannelRegistrationKey;
 
     /** Send this message to stakeholders when the garbage collector finds that a channel was deleted */
     class SDRBASE_API MsgReportChannelDeleted : public Message {

sdrbase/pipes/messagepipesgcworker.cpp

@@ -15,15 +15,20 @@
 // along with this program. If not, see <http://www.gnu.org/licenses/>.          //
 ///////////////////////////////////////////////////////////////////////////////////
 
+#include "channel/channelapi.h"
 #include "feature/feature.h"
 #include "util/messagequeue.h"
 #include "maincore.h"
 #include "messagepipescommon.h"
 #include "messagepipesgcworker.h"
 
-bool MessagePipesGCWorker::MessagePipesGC::existsProducer(const ChannelAPI *channel)
+bool MessagePipesGCWorker::MessagePipesGC::existsProducer(const PipeEndPoint *pipeEndPoint)
 {
-    return MainCore::instance()->existsChannel(channel);
+    // Not overly sure about casting to both types here, but currently safeish as the
+    // existing functions only use the pointer address - and I presume these
+    // may be pointers to deleted objects anyway?
+    return MainCore::instance()->existsChannel((const ChannelAPI *)pipeEndPoint)
+        || MainCore::instance()->existsFeature((const Feature *)pipeEndPoint);
 }
 
 bool MessagePipesGCWorker::MessagePipesGC::existsConsumer(const Feature *feature)

sdrbase/pipes/messagepipesgcworker.h

@@ -50,10 +50,10 @@ public:
     bool isRunning() const { return m_running; }
 
 private:
-    class MessagePipesGC : public ElementPipesGC<ChannelAPI, Feature, MessageQueue>
+    class MessagePipesGC : public ElementPipesGC<PipeEndPoint, Feature, MessageQueue>
     {
     private:
-        virtual bool existsProducer(const ChannelAPI *channelAPI);
+        virtual bool existsProducer(const PipeEndPoint *pipeEndPoint);
         virtual bool existsConsumer(const Feature *feature);
         virtual void sendMessageToConsumer(const MessageQueue *messageQueue,  MessagePipesCommon::ChannelRegistrationKey key, Feature *feature);
     };

sdrbase/pipes/pipeendpoint.cpp

@@ -0,0 +1,147 @@
+///////////////////////////////////////////////////////////////////////////////////
+// Copyright (C) 2021 Jon Beniston, M7RCE                                        //
+//                                                                               //
+// 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                  //
+// (at your option) any later version.                                           //
+//                                                                               //
+// 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 <vector>
+
+#include <QRegExp>
+#include <QDebug>
+
+#include "dsp/dspengine.h"
+#include "device/deviceset.h"
+#include "channel/channelapi.h"
+#include "feature/featureset.h"
+#include "feature/feature.h"
+#include "maincore.h"
+
+#include "pipeendpoint.h"
+
+MESSAGE_CLASS_DEFINITION(PipeEndPoint::MsgReportPipes, Message)
+
+QList<PipeEndPoint::AvailablePipeSource> PipeEndPoint::updateAvailablePipeSources(QString pipeName, QStringList pipeTypes, QStringList pipeURIs, Feature *destinationFeature)
+{
+    MainCore *mainCore = MainCore::instance();
+    MessagePipes& messagePipes = mainCore->getMessagePipes();
+    std::vector<DeviceSet*>& deviceSets = mainCore->getDeviceSets();
+    QHash<PipeEndPoint *, AvailablePipeSource> availablePipes;
+
+    int deviceIndex = 0;
+    for (std::vector<DeviceSet*>::const_iterator it = deviceSets.begin(); it != deviceSets.end(); ++it, deviceIndex++)
+    {
+        DSPDeviceSourceEngine *deviceSourceEngine =  (*it)->m_deviceSourceEngine;
+        DSPDeviceSinkEngine *deviceSinkEngine =  (*it)->m_deviceSinkEngine;
+
+        if (deviceSourceEngine || deviceSinkEngine)
+        {
+            for (int chi = 0; chi < (*it)->getNumberOfChannels(); chi++)
+            {
+                ChannelAPI *channel = (*it)->getChannelAt(chi);
+                int i = pipeURIs.indexOf(channel->getURI());
+
+                if (i >= 0)
+                {
+                    if (!availablePipes.contains(channel))
+                    {
+                        MessageQueue *messageQueue = messagePipes.registerChannelToFeature(channel, destinationFeature, pipeName);
+                        QObject::connect(
+                            messageQueue,
+                            &MessageQueue::messageEnqueued,
+                            destinationFeature,
+                            [=](){ destinationFeature->handlePipeMessageQueue(messageQueue); },
+                            Qt::QueuedConnection
+                        );
+                    }
+
+                    AvailablePipeSource availablePipe =
+                        AvailablePipeSource{
+                            deviceSinkEngine != nullptr ? AvailablePipeSource::TX : AvailablePipeSource::RX,
+                            deviceIndex,
+                            chi,
+                            channel,
+                            pipeTypes.at(i)
+                        };
+                    availablePipes[channel] = availablePipe;
+                }
+            }
+        }
+    }
+
+    std::vector<FeatureSet*>& featureSets = mainCore->getFeatureeSets();
+    int featureIndex = 0;
+    for (std::vector<FeatureSet*>::const_iterator it = featureSets.begin(); it != featureSets.end(); ++it, featureIndex++)
+    {
+        for (int fi = 0; fi < (*it)->getNumberOfFeatures(); fi++)
+        {
+            Feature *feature = (*it)->getFeatureAt(fi);
+            int i = pipeURIs.indexOf(feature->getURI());
+
+            if (i >= 0)
+            {
+                if (!availablePipes.contains(feature))
+                {
+                    MessageQueue *messageQueue = messagePipes.registerChannelToFeature(feature, destinationFeature, pipeName);
+                    QObject::connect(
+                        messageQueue,
+                        &MessageQueue::messageEnqueued,
+                        destinationFeature,
+                        [=](){ destinationFeature->handlePipeMessageQueue(messageQueue); },
+                        Qt::QueuedConnection
+                    );
+                }
+
+                AvailablePipeSource availablePipe =
+                    AvailablePipeSource{
+                        AvailablePipeSource::Feature,
+                        featureIndex,
+                        fi,
+                        feature,
+                        pipeTypes.at(i)
+                    };
+                availablePipes[feature] = availablePipe;
+            }
+        }
+    }
+
+    QList<AvailablePipeSource> availablePipeList;
+    QHash<PipeEndPoint*, AvailablePipeSource>::iterator it = availablePipes.begin();
+
+    for (; it != availablePipes.end(); ++it) {
+        availablePipeList.push_back(*it);
+    }
+    return availablePipeList;
+}
+
+PipeEndPoint *PipeEndPoint::getPipeEndPoint(const QString name, const QList<AvailablePipeSource> &availablePipeSources)
+{
+    QRegExp re("([TRF])([0-9]+):([0-9]+) ([a-zA-Z0-9]+)");
+    if (re.exactMatch(name))
+    {
+        QString type = re.capturedTexts()[1];
+        int setIndex = re.capturedTexts()[2].toInt();
+        int index = re.capturedTexts()[3].toInt();
+        QString id = re.capturedTexts()[4];
+
+        QListIterator<AvailablePipeSource> itr(availablePipeSources);
+        while (itr.hasNext()) {
+            AvailablePipeSource p = itr.next();
+            if ((p.m_setIndex == setIndex) && (p.m_index == index) && (id == p.m_id))
+                return p.m_source;
+        }
+    }
+    else
+        qDebug() << "PipeEndPoint::getPipeEndPoint: " << name << " is malformed";
+    return nullptr;
+}

sdrbase/pipes/pipeendpoint.h

@@ -0,0 +1,100 @@
+///////////////////////////////////////////////////////////////////////////////////
+// Copyright (C) 2021 Jon Beniston, M7RCE                                        //
+//                                                                               //
+// Parent for ChannelAPI and Features, where either can be used.                 //
+//                                                                               //
+// 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                  //
+// (at your option) any later version.                                           //
+//                                                                               //
+// 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 SDRBASE_PIPES_PIPEENDPOINT_H_
+#define SDRBASE_PIPES_PIPEENDPOINT_H_
+
+#include <QList>
+#include <QString>
+#include <QStringList>
+
+#include "util/message.h"
+#include "export.h"
+
+class Feature;
+
+class SDRBASE_API PipeEndPoint {
+public:
+
+    // Used by pipe sinks (features) to record details about available pipe sources (channels or features)
+    struct AvailablePipeSource
+    {
+        enum {RX, TX, Feature} m_type;
+        int m_setIndex;
+        int m_index;
+        PipeEndPoint *m_source;
+        QString m_id;
+
+        AvailablePipeSource() = default;
+        AvailablePipeSource(const AvailablePipeSource&) = default;
+        AvailablePipeSource& operator=(const AvailablePipeSource&) = default;
+        friend bool operator==(const AvailablePipeSource &lhs, const AvailablePipeSource &rhs)
+        {
+            return (lhs.m_type == rhs.m_type)
+                && (lhs.m_setIndex == rhs.m_setIndex)
+                && (lhs.m_source == rhs.m_source)
+                && (lhs.m_id == rhs.m_id);
+        }
+
+        QString getTypeName() const
+        {
+            QStringList typeNames = {"R", "T", "F"};
+            return typeNames[m_type];
+        }
+
+        // Name for use in GUI combo boxes and WebAPI
+        QString getName() const
+        {
+            QString type;
+
+            return QString("%1%2:%3 %4").arg(getTypeName())
+                                        .arg(m_setIndex)
+                                        .arg(m_index)
+                                        .arg(m_id);
+        }
+    };
+
+    class SDRBASE_API MsgReportPipes : public Message {
+        MESSAGE_CLASS_DECLARATION
+
+    public:
+        QList<AvailablePipeSource>& getAvailablePipes() { return m_availablePipes; }
+
+        static MsgReportPipes* create() {
+            return new MsgReportPipes();
+        }
+
+    private:
+        QList<AvailablePipeSource> m_availablePipes;
+
+        MsgReportPipes() :
+            Message()
+        {}
+    };
+
+
+protected:
+
+    // Utility functions for pipe sinks to manage list of sources
+    QList<AvailablePipeSource> updateAvailablePipeSources(QString pipeName, QStringList pipeTypes, QStringList pipeURIs, Feature *destinationFeature);
+    PipeEndPoint *getPipeEndPoint(const QString name, const QList<AvailablePipeSource> &availablePipeSources);
+
+};
+
+#endif // SDRBASE_PIPES_PIPEENDPOINT_H_