DSD demod: implement RTP over UDP for audio copy (part 2)

plugins/channelrx/demoddsd/dsddemod.cpp

@@ -22,6 +22,7 @@
 #include <complex.h>
 
 #include "audio/audiooutput.h"
+#include "audio/audionetsink.h"
 #include "dsp/dspengine.h"
 #include "dsp/threadedbasebandsamplesink.h"
 #include "dsp/downchannelizer.h"
@@ -74,9 +75,14 @@ DSDDemod::DSDDemod(DeviceSourceAPI *deviceAPI) :
 	DSPEngine::instance()->addAudioSink(&m_audioFifo1);
     DSPEngine::instance()->addAudioSink(&m_audioFifo2);
 
-    m_udpBufferAudio = new UDPSink<AudioSample>(this, m_udpBlockSize, m_settings.m_udpPort);
+//    m_udpBufferAudio = new UDPSink<AudioSample>(this, m_udpBlockSize, m_settings.m_udpPort);
-    m_audioFifo1.setUDPSink(m_udpBufferAudio);
+//    m_audioFifo1.setUDPSink(m_udpBufferAudio);
-    m_audioFifo2.setUDPSink(m_udpBufferAudio);
+//    m_audioFifo2.setUDPSink(m_udpBufferAudio);
+    m_audioNetSink = new AudioNetSink(0); // parent thread allocated dynamically
+    m_audioNetSink->setDestination(m_settings.m_udpAddress, m_settings.m_udpPort);
+    m_audioNetSink->setStereo(true);
+    m_audioFifo1.setAudioNetSink(m_audioNetSink);
+    m_audioFifo2.setAudioNetSink(m_audioNetSink);
 
     m_channelizer = new DownChannelizer(this);
     m_threadedChannelizer = new ThreadedBasebandSampleSink(m_channelizer, this);
@@ -92,7 +98,8 @@ DSDDemod::~DSDDemod()
     delete[] m_sampleBuffer;
 	DSPEngine::instance()->removeAudioSink(&m_audioFifo1);
     DSPEngine::instance()->removeAudioSink(&m_audioFifo2);
-    delete m_udpBufferAudio;
+//    delete m_udpBufferAudio;
+    delete m_audioNetSink;
 
     m_deviceAPI->removeChannelAPI(this);
     m_deviceAPI->removeThreadedSink(m_threadedChannelizer);
@@ -363,6 +370,14 @@ bool DSDDemod::handleMessage(const Message& cmd)
 		m_dsdDecoder.setMyPoint(cfg.getMyLatitude(), cfg.getMyLongitude());
 		return true;
 	}
+    else if (BasebandSampleSink::MsgThreadedSink::match(cmd))
+    {
+        BasebandSampleSink::MsgThreadedSink& cfg = (BasebandSampleSink::MsgThreadedSink&) cmd;
+        const QThread *thread = cfg.getThread();
+        qDebug("DSDDemod::handleMessage: BasebandSampleSink::MsgThreadedSink: %p", thread);
+        m_audioNetSink->moveToThread(const_cast<QThread*>(thread)); // use the thread for udp sinks
+        return true;
+    }
     else if (DSPSignalNotification::match(cmd))
     {
         return true;
@@ -477,8 +492,9 @@ void DSDDemod::applySettings(const DSDDemodSettings& settings, bool force)
     if ((settings.m_udpAddress != m_settings.m_udpAddress)
         || (settings.m_udpPort != m_settings.m_udpPort) || force)
     {
-        m_udpBufferAudio->setAddress(const_cast<QString&>(settings.m_udpAddress));
+//        m_udpBufferAudio->setAddress(const_cast<QString&>(settings.m_udpAddress));
-        m_udpBufferAudio->setPort(settings.m_udpPort);
+//        m_udpBufferAudio->setPort(settings.m_udpPort);
+        m_audioNetSink->setDestination(settings.m_udpAddress, settings.m_udpPort);
     }
 
     if ((settings.m_copyAudioToUDP != m_settings.m_copyAudioToUDP)
@@ -489,6 +505,26 @@ void DSDDemod::applySettings(const DSDDemodSettings& settings, bool force)
         m_audioFifo2.setCopyToUDP(settings.m_slot2On && !settings.m_slot1On && settings.m_copyAudioToUDP);
     }
 
+    if ((settings.m_copyAudioUseRTP != m_settings.m_copyAudioUseRTP) || force)
+    {
+        if (settings.m_copyAudioUseRTP)
+        {
+            if (m_audioNetSink->selectType(AudioNetSink::SinkRTP)) {
+                qDebug("DSDDemod::applySettings: set audio sink to RTP mode");
+            } else {
+                qWarning("DSDDemod::applySettings: RTP support for audio sink not available. Fall back too UDP");
+            }
+        }
+        else
+        {
+            if (m_audioNetSink->selectType(AudioNetSink::SinkUDP)) {
+                qDebug("DSDDemod::applySettings: set audio sink to UDP mode");
+            } else {
+                qWarning("DSDDemod::applySettings: failed to set audio sink to UDP mode");
+            }
+        }
+    }
+
     if ((settings.m_highPassFilter != m_settings.m_highPassFilter) || force)
     {
         m_dsdDecoder.useHPMbelib(settings.m_highPassFilter);

plugins/channelrx/demoddsd/dsddemod.h

@@ -199,7 +199,8 @@ private:
 	QMutex m_settingsMutex;
 
     PhaseDiscriminators m_phaseDiscri;
-    UDPSink<AudioSample> *m_udpBufferAudio;
+    //UDPSink<AudioSample> *m_udpBufferAudio;
+    AudioNetSink *m_audioNetSink;
 
     static const int m_udpBlockSize;
 

plugins/channelrx/demodwfm/wfmdemod.h

@@ -31,6 +31,7 @@
 #include "dsp/phasediscri.h"
 #include "audio/audiofifo.h"
 #include "util/message.h"
+#include "util/udpsink.h"
 
 #include "wfmdemodsettings.h"
 

sdrbase/audio/audiofifo.cpp

@@ -19,13 +19,15 @@
 #include <QTime>
 #include "dsp/dsptypes.h"
 #include "audio/audiofifo.h"
+#include "audio/audionetsink.h"
 
 #define MIN(x, y) ((x) < (y) ? (x) : (y))
 
 AudioFifo::AudioFifo() :
 	m_fifo(0),
 	m_sampleSize(sizeof(AudioSample)),
-	m_udpSink(0),
+	//m_udpSink(0),
+	m_audioNetSink(0),
 	m_copyToUDP(false)
 {
 	m_size = 0;
@@ -37,7 +39,8 @@ AudioFifo::AudioFifo() :
 AudioFifo::AudioFifo(uint32_t numSamples) :
 	m_fifo(0),
     m_sampleSize(sizeof(AudioSample)),
-    m_udpSink(0),
+    //m_udpSink(0),
+    m_audioNetSink(0),
     m_copyToUDP(false)
 {
 	QMutexLocker mutexLocker(&m_mutex);
@@ -75,9 +78,10 @@ uint AudioFifo::write(const quint8* data, uint32_t numSamples, int timeout_ms)
 	uint32_t remaining;
 	uint32_t copyLen;
 
-	if (m_copyToUDP && m_udpSink)
+	if (m_copyToUDP && m_audioNetSink)
 	{
-	    m_udpSink->write((AudioSample *) data, numSamples);
+	    //m_udpSink->write((AudioSample *) data, numSamples);
+	    m_audioNetSink->write((AudioSample *) data, numSamples);
 	}
 
 	if(m_fifo == 0)

sdrbase/audio/audiofifo.h

@@ -24,7 +24,9 @@
 
 #include "dsp/dsptypes.h"
 #include "util/export.h"
-#include "util/udpsink.h"
+//#include "util/udpsink.h"
+
+class AudioNetSink;
 
 class SDRBASE_API AudioFifo : public QObject {
 	Q_OBJECT
@@ -47,7 +49,8 @@ public:
 	inline bool isFull() const { return m_fill == m_size; }
 	inline uint32_t size() const { return m_size; }
 
-	void setUDPSink(UDPSink<AudioSample> *udpSink) { m_udpSink = udpSink; }
+	//void setUDPSink(UDPSink<AudioSample> *udpSink) { m_udpSink = udpSink; }
+	void setAudioNetSink(AudioNetSink *audioNetSink) { m_audioNetSink = audioNetSink; }
 	void setCopyToUDP(bool copyToUDP) { m_copyToUDP = copyToUDP; }
 
 private:
@@ -67,7 +70,8 @@ private:
 	QWaitCondition m_writeWaitCondition;
 	QWaitCondition m_readWaitCondition;
 
-	UDPSink<AudioSample> *m_udpSink;
+	//UDPSink<AudioSample> *m_udpSink;
+	AudioNetSink *m_audioNetSink;
 	bool m_copyToUDP;
 
 	bool create(uint32_t numSamples);

sdrbase/audio/audionetsink.cpp

@@ -89,6 +89,11 @@ void AudioNetSink::deleteDestination(const QString& address, uint16_t port)
     }
 }
 
+void AudioNetSink::setStereo(bool stereo)
+{
+    m_rtpBufferAudio->setPayloadType(stereo ? RTPSink::PayloadL16Stereo : RTPSink::PayloadL16Stereo);
+}
+
 void AudioNetSink::write(qint16 sample)
 {
     if (m_type == SinkUDP)
@@ -111,6 +116,36 @@ void AudioNetSink::write(qint16 sample)
     }
 }
 
+void AudioNetSink::write(AudioSample* samples, uint32_t numSamples)
+{
+    if (m_type == SinkUDP)
+    {
+        int samplesIndex = 0;
+
+        if (m_bufferIndex + numSamples*sizeof(AudioSample) >= m_udpBlockSize) // fill remainder of buffer and send it
+        {
+            memcpy(&m_data[m_bufferIndex], &samples[samplesIndex], m_udpBlockSize - m_bufferIndex); // fill remainder of buffer
+            m_udpSocket->writeDatagram((const char*)m_data, (qint64 ) m_udpBlockSize, m_address, m_port);
+            m_bufferIndex = 0;
+            samplesIndex += (m_udpBlockSize - m_bufferIndex) / sizeof(AudioSample);
+            numSamples -= (m_udpBlockSize - m_bufferIndex) / sizeof(AudioSample);
+        }
+
+        while (numSamples > m_udpBlockSize/sizeof(AudioSample)) // send directly from input without buffering
+        {
+            m_udpSocket->writeDatagram((const char*)&samples[samplesIndex], (qint64 ) m_udpBlockSize, m_address, m_port);
+            samplesIndex += m_udpBlockSize/sizeof(AudioSample);
+            numSamples -= m_udpBlockSize/sizeof(AudioSample);
+        }
+
+        memcpy(&m_data[m_bufferIndex], &samples[samplesIndex], numSamples*sizeof(AudioSample));
+    }
+    else if (m_type == SinkRTP)
+    {
+        m_rtpBufferAudio->write((uint8_t *) samples, numSamples*sizeof(AudioSample));
+    }
+}
+
 void AudioNetSink::moveToThread(QThread *thread)
 {
     m_udpSocket->moveToThread(thread);

sdrbase/audio/audionetsink.h

@@ -43,8 +43,10 @@ public:
     void setDestination(const QString& address, uint16_t port);
     void addDestination(const QString& address, uint16_t port);
     void deleteDestination(const QString& address, uint16_t port);
+    void setStereo(bool stereo);
 
     void write(qint16 sample);
+    void write(AudioSample* samples, uint32_t numSamples);
 
     bool isRTPCapable() const;
     bool selectType(SinkType type);