qrtplib: copy Audio to UDP/RTP: use a single UDP socket for UDP and RTP

qrtplib/rtpudptransmitter.cpp

@@ -48,6 +48,7 @@ RTPUDPTransmitter::RTPUDPTransmitter() :
     m_init = false;
     m_rtcpsock = 0;
     m_rtpsock = 0;
+    m_deletesocketswhendone = false;
     m_waitingfordata = false;
     m_rtcpPort = 0;
     m_rtpPort = 0;
@@ -122,15 +123,24 @@ int RTPUDPTransmitter::Create(std::size_t maximumpacketsize, const RTPTransmissi
         }
     }
 
-    m_rtpsock = new QUdpSocket();
-
-    // If we're multiplexing, we're just going to set the RTCP socket to equal the RTP socket
-    if (params->GetRTCPMultiplexing())
+    if (params->GetUseExistingSockets(&m_rtpsock, &m_rtcpsock))
     {
-        m_rtcpsock = m_rtpsock;
-        m_rtcpPort = m_rtpPort;
-    } else {
-        m_rtcpsock = new QUdpSocket();
+        m_deletesocketswhendone = false;
+    }
+    else
+    {
+        m_deletesocketswhendone = true;
+
+        m_rtpsock = new QUdpSocket();
+
+        // If we're multiplexing, we're just going to set the RTCP socket to equal the RTP socket
+        if (params->GetRTCPMultiplexing())
+        {
+            m_rtcpsock = m_rtpsock;
+            m_rtcpPort = m_rtpPort;
+        } else {
+            m_rtcpsock = new QUdpSocket();
+        }
     }
 
     // set socket buffer sizes
@@ -174,17 +184,6 @@ int RTPUDPTransmitter::BindSockets()
     return 0;
 }
 
-void RTPUDPTransmitter::moveToThread(QThread *thread)
-{
-    if (m_rtpsock) {
-        m_rtpsock->moveToThread(thread);
-    }
-
-    if (m_rtpsock != m_rtcpsock) {
-        m_rtcpsock->moveToThread(thread);
-    }
-}
-
 void RTPUDPTransmitter::Destroy()
 {
     if (!m_init) {
@@ -196,11 +195,14 @@ void RTPUDPTransmitter::Destroy()
         return;
     }
 
-    if (m_rtpsock != m_rtcpsock) {
-        delete m_rtcpsock;
-    }
+    if (m_deletesocketswhendone)
+    {
+        if (m_rtpsock != m_rtcpsock) {
+            delete m_rtcpsock;
+        }
 
-    delete m_rtpsock;
+        delete m_rtpsock;
+    }
 
     m_created = false;
 }

qrtplib/rtpudptransmitter.h

@@ -142,6 +142,16 @@ public:
         m_forcedrtcpport = rtcpport;
     }
 
+    /** Use sockets that have already been created, no checks on port numbers
+     *  will be done, and no buffer sizes will be set; you'll need to close
+     *  the sockets yourself when done, it will **not** be done automatically. */
+    void SetUseExistingSockets(QUdpSocket *rtpsocket, QUdpSocket *rtcpsocket)
+    {
+        m_rtpsock = rtpsocket;
+        m_rtcpsock = rtcpsocket;
+        m_useexistingsockets = true;
+    }
+
     /** Returns the RTP socket's send buffer size. */
     int GetRTPSendBufferSize() const
     {
@@ -184,6 +194,20 @@ public:
         return m_forcedrtcpport;
     }
 
+    /** Returns true and fills in sockets if existing sockets were set
+     *  using RTPUDPv4TransmissionParams::SetUseExistingSockets. */
+    bool GetUseExistingSockets(QUdpSocket **rtpsocket, QUdpSocket **rtcpsocket) const
+    {
+        if (!m_useexistingsockets) {
+            return false;
+        }
+
+        *rtpsocket = m_rtpsock;
+        *rtcpsocket = m_rtcpsock;
+
+        return true;
+    }
+
 private:
     QHostAddress m_bindAddress;
     QNetworkInterface m_mcastInterface;
@@ -195,6 +219,7 @@ private:
     uint16_t m_forcedrtcpport;
 
     QUdpSocket *m_rtpsock, *m_rtcpsock;
+    bool m_useexistingsockets;
 };
 
 inline RTPUDPTransmissionParams::RTPUDPTransmissionParams() :
@@ -210,6 +235,7 @@ inline RTPUDPTransmissionParams::RTPUDPTransmissionParams() :
     m_forcedrtcpport = 0;
     m_rtpsock = 0;
     m_rtcpsock = 0;
+    m_useexistingsockets = false;
 }
 
 /** Additional information about the UDP over IPv4 transmitter. */
@@ -321,6 +347,7 @@ private:
     bool m_created;
     bool m_waitingfordata;
     QUdpSocket *m_rtpsock, *m_rtcpsock;
+    bool m_deletesocketswhendone;
     QHostAddress m_localIP; //!< from parameters bind IP
     QNetworkInterface m_multicastInterface; //!< from parameters multicast interface
     uint16_t m_rtpPort, m_rtcpPort;

sdrbase/audio/audionetsink.cpp

@@ -16,34 +16,36 @@
 ///////////////////////////////////////////////////////////////////////////////////
 
 #include "audionetsink.h"
-#include "util/udpsink.h"
 #include "util/rtpsink.h"
 
+#include <unistd.h>
+#include <QUdpSocket>
+
 const int AudioNetSink::m_udpBlockSize = 512;
 
 AudioNetSink::AudioNetSink(QObject *parent, bool stereo) :
     m_type(SinkUDP),
-    m_udpBufferAudioMono(0),
-    m_rtpBufferAudio(0)
+    m_rtpBufferAudio(0),
+    m_bufferIndex(0),
+    m_port(9998)
 {
-    m_udpBufferAudioMono = new UDPSink<int16_t>(parent, m_udpBlockSize);
-    //m_rtpBufferAudio = new RTPSink("127.0.0.1", 9999, stereo ? RTPSink::PayloadL16Stereo : RTPSink::PayloadL16Mono);
+    m_udpSocket = new QUdpSocket(parent);
+    m_rtpBufferAudio = new RTPSink(m_udpSocket, stereo);
 }
 
 AudioNetSink::~AudioNetSink()
 {
-    if (m_udpBufferAudioMono) {
-        delete m_udpBufferAudioMono;
-    }
     if (m_rtpBufferAudio) {
         delete m_rtpBufferAudio;
     }
+
+    m_udpSocket->deleteLater(); // this thread is not the owner thread (was moved)
 }
 
-//bool AudioNetSink::isRTPCapable() const
-//{
-//        return m_rtpBufferAudio->isValid();
-//}
+bool AudioNetSink::isRTPCapable() const
+{
+        return m_rtpBufferAudio->isValid();
+}
 
 bool AudioNetSink::selectType(SinkType type)
 {
@@ -65,9 +67,9 @@ bool AudioNetSink::selectType(SinkType type)
 
 void AudioNetSink::setDestination(const QString& address, uint16_t port)
 {
-    if (m_udpBufferAudioMono) {
-        m_udpBufferAudioMono->setDestination(address, port);
-    }
+    m_address.setAddress(const_cast<QString&>(address));
+    m_port = port;
+
     if (m_rtpBufferAudio) {
         m_rtpBufferAudio->setDestination(address, port);
     }
@@ -89,21 +91,28 @@ void AudioNetSink::deleteDestination(const QString& address, uint16_t port)
 
 void AudioNetSink::write(qint16 sample)
 {
-    if (m_type == SinkUDP) {
-        m_udpBufferAudioMono->write(sample);
-    } else if (m_type == SinkRTP) {
+    if (m_type == SinkUDP)
+    {
+        if (m_bufferIndex >= m_udpBlockSize)
+        {
+            m_udpSocket->writeDatagram((const char*)m_data, (qint64 ) m_udpBlockSize, m_address, m_port);
+            m_bufferIndex = 0;
+        }
+        else
+        {
+            qint16 *p = (qint16*) &m_data[m_bufferIndex];
+            *p = sample;
+            m_bufferIndex += sizeof(qint16);
+        }
+    }
+    else if (m_type == SinkRTP)
+    {
         m_rtpBufferAudio->write((uint8_t *) &sample);
     }
 }
 
 void AudioNetSink::moveToThread(QThread *thread)
 {
-    if (m_udpBufferAudioMono) {
-        m_udpBufferAudioMono->moveToThread(thread);
-    }
-
-    if (m_rtpBufferAudio) {
-        m_rtpBufferAudio->moveToThread(thread);
-    }
+    m_udpSocket->moveToThread(thread);
 }
 

sdrbase/audio/audionetsink.h

@@ -21,10 +21,11 @@
 #include "dsp/dsptypes.h"
 #include "util/export.h"
 
-#include <stdint.h>
 #include <QObject>
+#include <QHostAddress>
+#include <stdint.h>
 
-template<typename T> class UDPSink;
+class QUdpSocket;
 class RTPSink;
 class QThread;
 
@@ -45,7 +46,7 @@ public:
 
     void write(qint16 sample);
 
-    bool isRTPCapable() const { return false; }
+    bool isRTPCapable() const;
     bool selectType(SinkType type);
 
     void moveToThread(QThread *thread);
@@ -54,8 +55,12 @@ public:
 
 protected:
     SinkType m_type;
-    UDPSink<qint16> *m_udpBufferAudioMono;
+    QUdpSocket *m_udpSocket;
     RTPSink *m_rtpBufferAudio;
+    char m_data[65536];
+    unsigned int m_bufferIndex;
+    QHostAddress m_address;
+    unsigned int m_port;
 };
 
 

sdrbase/util/rtpsink.cpp

@@ -19,19 +19,20 @@
 #include "dsp/dsptypes.h"
 #include <algorithm>
 
-RTPSink::RTPSink(const QString& address, uint16_t port, PayloadType payloadType) :
-    m_payloadType(payloadType),
+RTPSink::RTPSink(QUdpSocket *udpSocket, bool stereo) :
+    m_payloadType(stereo ? RTPSink::PayloadL16Stereo : RTPSink::PayloadL16Mono),
     m_sampleRate(48000),
     m_sampleBytes(0),
     m_packetSamples(0),
     m_bufferSize(0),
     m_sampleBufferIndex(0),
     m_byteBuffer(0),
-    m_destport(port),
+    m_destport(9998),
     m_mutex(QMutex::Recursive)
 {
 	m_rtpSessionParams.SetOwnTimestampUnit(1.0 / (double) m_sampleRate);
     m_rtpTransmissionParams.SetRTCPMultiplexing(true); // do not allocate another socket for RTCP
+    m_rtpTransmissionParams.SetUseExistingSockets(udpSocket, udpSocket);
 
     int status = m_rtpTransmitter.Init();
     if (status < 0) {
@@ -52,14 +53,12 @@ RTPSink::RTPSink(const QString& address, uint16_t port, PayloadType payloadType)
         qDebug("RTPSink::RTPSink: created session: %s", qrtplib::RTPGetErrorString(status).c_str());
     }
 
-    setPayloadType(payloadType);
+    setPayloadType(m_payloadType);
     m_valid = true;
 
     uint32_t endianTest32 = 1;
     uint8_t *ptr = (uint8_t*) &endianTest32;
     m_endianReverse = (*ptr == 1);
-
-    m_destip.setAddress(address);
 }
 
 RTPSink::~RTPSink()

sdrbase/util/rtpsink.h

@@ -33,6 +33,8 @@
 
 #include "util/export.h"
 
+class QUdpSocket;
+
 class RTPSink
 {
 public:
@@ -42,7 +44,7 @@ public:
         PayloadL16Stereo,
     } PayloadType;
 
-    RTPSink(const QString& address, uint16_t port, PayloadType payloadType = PayloadL16Mono);
+    RTPSink(QUdpSocket *udpSocket, bool stereo);
     ~RTPSink();
 
     bool isValid() const { return m_valid; }
@@ -55,10 +57,6 @@ public:
     void write(const uint8_t *sampleByte);
     void write(const uint8_t *sampleByte, int nbSamples);
 
-    void moveToThread(QThread *thread) {
-        m_rtpTransmitter.moveToThread(thread);
-    }
-
 protected:
     /** Reverse endianess in destination buffer */
     static void writeNetBuf(uint8_t *dest, const uint8_t *src, unsigned int elemLen, unsigned int bytesLen, bool endianReverse);