Added network sink and fixes to the networking lib

CMakeLists.txt

@@ -25,6 +25,7 @@ option(OPT_BUILD_PLUTOSDR_SOURCE "Build PlutoSDR Source Module (Depedencies: lib
 # Sinks
 option(OPT_BUILD_AUDIO_SINK "Build Audio Sink Module (Depedencies: rtaudio)" ON)
 option(OPT_BUILD_PORTAUDIO_SINK "Build PortAudio Sink Module (Depedencies: portaudio)" OFF)
+option(OPT_BUILD_NETWORK_SINK "Build Audio Sink Module (no dependencies required)" ON)
 option(OPT_BUILD_NEW_PORTAUDIO_SINK "Build the new PortAudio Sink Module (Depedencies: portaudio)" OFF)
 
 # Decoders
@@ -115,6 +116,10 @@ if (OPT_BUILD_PORTAUDIO_SINK)
 add_subdirectory("portaudio_sink")
 endif (OPT_BUILD_PORTAUDIO_SINK)
 
+if (OPT_BUILD_NETWORK_SINK)
+add_subdirectory("network_sink")
+endif (OPT_BUILD_NETWORK_SINK)
+
 if (OPT_BUILD_NEW_PORTAUDIO_SINK)
 add_subdirectory("new_portaudio_sink")
 endif (OPT_BUILD_NEW_PORTAUDIO_SINK)

core/src/utils/networking.cpp

@@ -7,8 +7,10 @@ namespace net {
     extern bool winsock_init = false;
 #endif
 
-    ConnClass::ConnClass(Socket sock) {
+    ConnClass::ConnClass(Socket sock, struct sockaddr_in raddr, bool udp) {
         _sock = sock;
+        _udp = udp;
+        remoteAddr = raddr;
         connectionOpen = true;
         readWorkerThread = std::thread(&ConnClass::readWorker, this);
         writeWorkerThread = std::thread(&ConnClass::writeWorker, this);
@@ -63,11 +65,16 @@ namespace net {
     int ConnClass::read(int count, uint8_t* buf) {
         if (!connectionOpen) { return -1; }
         std::lock_guard lck(readMtx);
-#ifdef _WIN32
+        int ret;
-        int ret = recv(_sock, (char*)buf, count, 0);
+
-#else
+        if (_udp) {
-        int ret = ::read(_sock, buf, count);
+            int fromLen = sizeof(remoteAddr);
-#endif
+            ret = recvfrom(_sock, (char*)buf, count, 0, (struct sockaddr*)&remoteAddr, &fromLen);
+        }
+        else {
+            ret = recv(_sock, (char*)buf, count, 0);
+        }
+
         if (ret <= 0) {
             {
                 std::lock_guard lck(connectionOpenMtx);
@@ -81,11 +88,16 @@ namespace net {
     bool ConnClass::write(int count, uint8_t* buf) {
         if (!connectionOpen) { return false; }
         std::lock_guard lck(writeMtx);
-#ifdef _WIN32
+        int ret;
-        int ret = send(_sock, (char*)buf, count, 0);
+
-#else
+        if (_udp) {
-        int ret = ::write(_sock, buf, count);
+            int fromLen = sizeof(remoteAddr);
-#endif
+            ret = sendto(_sock, (char*)buf, count, 0, (struct sockaddr*)&remoteAddr, sizeof(remoteAddr));
+        }
+        else {
+            ret = send(_sock, (char*)buf, count, 0);
+        }
+
         if (ret <= 0) {
             {
                 std::lock_guard lck(connectionOpenMtx);
@@ -200,7 +212,7 @@ namespace net {
 
         // Accept socket
         _sock = ::accept(sock, NULL, NULL);
-        if (_sock < 0) {
+        if (_sock < 0 || _sock == SOCKET_ERROR) {
             listening = false;
             throw std::runtime_error("Could not bind socket");
             return NULL;
@@ -282,8 +294,8 @@ namespace net {
     }
 
 
-    Conn connect(Protocol proto, std::string host, uint16_t port) {
+    Conn connect(std::string host, uint16_t port) {
-        Socket sock;
+    Socket sock;
 
 #ifdef _WIN32
         // Initilize WinSock2
@@ -299,7 +311,7 @@ namespace net {
 #endif
 
         // Create a socket
-        sock = socket(AF_INET, SOCK_STREAM, (proto == PROTO_TCP) ? IPPROTO_TCP : IPPROTO_UDP);
+        sock = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
         if (sock < 0) {
             throw std::runtime_error("Could not create socket");
             return NULL;
@@ -328,7 +340,7 @@ namespace net {
         return Conn(new ConnClass(sock));
     }
 
-    Listener listen(Protocol proto, std::string host, uint16_t port) {
+    Listener listen(std::string host, uint16_t port) {
         Socket listenSock;
 
 #ifdef _WIN32
@@ -345,7 +357,7 @@ namespace net {
 #endif
 
         // Create a socket
-        listenSock = socket(AF_INET, SOCK_STREAM, (proto == PROTO_TCP) ? IPPROTO_TCP : IPPROTO_UDP);
+        listenSock = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
         if (listenSock < 0) {
             throw std::runtime_error("Could not create socket");
             return NULL;
@@ -379,4 +391,67 @@ namespace net {
 
         return Listener(new ListenerClass(listenSock));
     }
+
+    Conn openUDP(std::string host, uint16_t port, std::string remoteHost, uint16_t remotePort, bool bindSocket) {
+        Socket sock;
+
+#ifdef _WIN32
+        // Initilize WinSock2
+        if (!winsock_init) {
+            WSADATA wsa;
+            if (WSAStartup(MAKEWORD(2,2),&wsa)) {
+                throw std::runtime_error("Could not initialize WinSock2");
+                return NULL;
+            }
+            winsock_init = true;
+        }
+        assert(winsock_init);
+#endif
+
+        // Create a socket
+        sock = socket(AF_INET, SOCK_DGRAM, 0);
+        if (sock < 0) {
+            throw std::runtime_error("Could not create socket");
+            return NULL;
+        }
+
+        // Get address from local hostname/ip
+        hostent* _host = gethostbyname(host.c_str());
+        if (_host == NULL || _host->h_addr_list[0] == NULL) {
+            throw std::runtime_error("Could get address from host");
+            return NULL;
+        }
+        uint32_t* naddr = (uint32_t*)_host->h_addr_list[0];
+
+        // Get address from remote hostname/ip
+        hostent* _remoteHost = gethostbyname(remoteHost.c_str());
+        if (_remoteHost == NULL || _remoteHost->h_addr_list[0] == NULL) {
+            throw std::runtime_error("Could get address from host");
+            return NULL;
+        }
+        uint32_t* rnaddr = (uint32_t*)_remoteHost->h_addr_list[0];
+
+        // Create host address
+        struct sockaddr_in addr;
+        addr.sin_addr.s_addr = *naddr;
+        addr.sin_family = AF_INET;
+	    addr.sin_port = htons(port);
+
+        // Create remote host address
+        struct sockaddr_in raddr;
+        raddr.sin_addr.s_addr = *rnaddr;
+        raddr.sin_family = AF_INET;
+	    raddr.sin_port = htons(remotePort);
+
+        // Bind socket
+        if (bindSocket) {
+            if (bind(sock, (struct sockaddr*)&addr, sizeof(addr)) < 0) {
+                throw std::runtime_error("Could not bind socket");
+                return NULL;
+            }
+        }
+        
+
+        return Conn(new ConnClass(sock, raddr, true));
+    }
 }

core/src/utils/networking.h

@@ -27,11 +27,6 @@ namespace net {
     typedef int Socket;
 #endif
 
-    enum Protocol {
-        PROTO_TCP,
-        PROTO_UDP
-    };
-
     struct ConnReadEntry {
         int count;
         uint8_t* buf;
@@ -46,7 +41,7 @@ namespace net {
 
     class ConnClass {
     public:
-        ConnClass(Socket sock);
+        ConnClass(Socket sock, struct sockaddr_in raddr = {}, bool udp = false);
         ~ConnClass();
 
         void close();
@@ -80,6 +75,8 @@ namespace net {
         std::thread writeWorkerThread;
 
         Socket _sock;
+        bool _udp;
+        struct sockaddr_in remoteAddr;
 
     };
 
@@ -119,8 +116,9 @@ namespace net {
 
     typedef std::unique_ptr<ListenerClass> Listener;
 
-    Conn connect(Protocol proto, std::string host, uint16_t port);
+    Conn connect(std::string host, uint16_t port);
-    Listener listen(Protocol proto, std::string host, uint16_t port);
+    Listener listen(std::string host, uint16_t port);
+    Conn openUDP(std::string host, uint16_t port, std::string remoteHost, uint16_t remotePort, bool bindSocket = true);
 
 #ifdef _WIN32
     extern bool winsock_init;

core/src/version.h

@@ -1,3 +1,3 @@
 #pragma once
 
-#define VERSION_STR "1.0.0_rc2"
+#define VERSION_STR "1.0.0_rc3"

network_sink/CMakeLists.txt

@@ -0,0 +1,24 @@
+cmake_minimum_required(VERSION 3.13)
+project(network_sink)
+
+if (MSVC)
+    set(CMAKE_CXX_FLAGS "-O2 /std:c++17 /EHsc")
+elseif (CMAKE_CXX_COMPILER_ID MATCHES "Clang")
+    set(CMAKE_CXX_FLAGS "-O3 -std=c++17 -Wno-unused-command-line-argument -undefined dynamic_lookup")
+else ()
+    set(CMAKE_CXX_FLAGS "-O3 -std=c++17")
+endif ()
+
+file(GLOB SRC "src/*.cpp")
+
+include_directories("src/")
+include_directories("../recorder/src")
+include_directories("../meteor_demodulator/src")
+include_directories("../radio/src")
+
+add_library(network_sink SHARED ${SRC})
+target_link_libraries(network_sink PRIVATE sdrpp_core)
+set_target_properties(network_sink PROPERTIES PREFIX "")
+
+# Install directives
+install(TARGETS network_sink DESTINATION lib/sdrpp/plugins)

network_sink/src/main.cpp

@@ -0,0 +1,358 @@
+#include <utils/networking.h>
+#include <imgui.h>
+#include <module.h>
+#include <gui/gui.h>
+#include <signal_path/signal_path.h>
+#include <signal_path/sink.h>
+#include <dsp/audio.h>
+#include <dsp/processing.h>
+#include <spdlog/spdlog.h>
+#include <config.h>
+#include <options.h>
+#include <gui/style.h>
+
+#define CONCAT(a, b) ((std::string(a) + b).c_str())
+
+SDRPP_MOD_INFO {
+    /* Name:            */ "network_sink",
+    /* Description:     */ "Network sink module for SDR++",
+    /* Author:          */ "Ryzerth",
+    /* Version:         */ 0, 1, 0,
+    /* Max instances    */ 1
+};
+
+ConfigManager config;
+
+enum {
+    SINK_MODE_TCP,
+    SINK_MODE_UDP
+};
+
+const char* sinkModesTxt = "TCP\0UDP\0";
+
+class NetworkSink : SinkManager::Sink {
+public:
+    NetworkSink(SinkManager::Stream* stream, std::string streamName) {
+        _stream = stream;
+        _streamName = streamName;
+
+        // Load config
+        config.acquire();
+        if (!config.conf.contains(_streamName)) {
+            config.conf[_streamName]["hostname"] = "localhost";
+            config.conf[_streamName]["port"] = 7355;
+            config.conf[_streamName]["protocol"] = SINK_MODE_UDP; // UDP
+            config.conf[_streamName]["sampleRate"] = 48000.0;
+            config.conf[_streamName]["stereo"] = false;
+            config.conf[_streamName]["listening"] = false;
+        }
+        std::string host = config.conf[_streamName]["hostname"];
+        strcpy(hostname, host.c_str());
+        port = config.conf[_streamName]["port"];
+        modeId = config.conf[_streamName]["protocol"];
+        sampleRate = config.conf[_streamName]["sampleRate"];
+        stereo = config.conf[_streamName]["stereo"];
+        bool startNow = config.conf[_streamName]["listening"];
+        config.release(true);
+
+        netBuf = new int16_t[STREAM_BUFFER_SIZE];
+
+        packer.init(_stream->sinkOut, 512);
+        s2m.init(&packer.out);
+        monoSink.init(&s2m.out, monoHandler, this);
+        stereoSink.init(&packer.out, stereoHandler, this);
+        
+
+        // Create a list of sample rates
+        for (int sr = 12000; sr < 200000; sr += 12000) {
+            sampleRates.push_back(sr);
+        }
+        for (int sr = 11025; sr < 192000; sr += 11025) {
+            sampleRates.push_back(sr);
+        }
+
+        // Sort sample rate list
+        std::sort(sampleRates.begin(), sampleRates.end(), [](double a, double b) { return (a < b); });
+
+        // Generate text list for UI
+        char buffer[128];
+        int id = 0;
+        int _48kId;
+        bool found = false;
+        for (auto sr : sampleRates) {
+            sprintf(buffer, "%d", (int)sr);
+            sampleRatesTxt += buffer;
+            sampleRatesTxt += '\0';
+            if (sr == sampleRate) { srId = id; found = true; }
+            if (sr == 48000.0) { _48kId = id; }
+            id++;
+        }
+        if (!found) { srId = _48kId; sampleRate = 48000.0; }
+        _stream->setSampleRate(sampleRate);
+
+        // Start if needed
+        if (startNow) { startServer(); }
+    }
+
+    ~NetworkSink() {
+        stopServer();
+        delete[] netBuf;
+    }
+
+    void start() {
+        if (running) {
+            return;
+        }
+        doStart();
+        running = true;
+    }
+
+    void stop() {
+        if (!running) {
+            return;
+        }
+        doStop();
+        running = false;
+    }
+
+    void menuHandler() {
+        float menuWidth = ImGui::GetContentRegionAvailWidth();
+
+        bool listening = (listener && listener->isListening()) || (conn && conn->isOpen());
+
+        if (listening) { style::beginDisabled(); }
+        if (ImGui::InputText(CONCAT("##_network_sink_host_", _streamName), hostname, 1023)) {
+            config.acquire();
+            config.conf[_streamName]["hostname"] = hostname;
+            config.release(true);
+        }
+        ImGui::SameLine();
+        ImGui::SetNextItemWidth(menuWidth - ImGui::GetCursorPosX());
+        if (ImGui::InputInt(CONCAT("##_network_sink_port_", _streamName), &port, 0, 0)) {
+            config.acquire();
+            config.conf[_streamName]["port"] = port;
+            config.release(true);
+        }
+
+        ImGui::Text("Protocol");
+        ImGui::SameLine();
+        ImGui::SetNextItemWidth(menuWidth - ImGui::GetCursorPosX());
+        if (ImGui::Combo(CONCAT("##_network_sink_mode_", _streamName), &modeId, sinkModesTxt)) {
+            config.acquire();
+            config.conf[_streamName]["protocol"] = modeId;
+            config.release(true);
+        }
+
+        if (listening) { style::endDisabled(); }
+
+        ImGui::Text("Samplerate");
+        ImGui::SameLine();
+        ImGui::SetNextItemWidth(menuWidth - ImGui::GetCursorPosX());
+        if (ImGui::Combo(CONCAT("##_network_sink_sr_", _streamName), &srId, sampleRatesTxt.c_str())) {
+            sampleRate = sampleRates[srId];
+            _stream->setSampleRate(sampleRate);
+            packer.setSampleCount(sampleRate / 60);
+            config.acquire();
+            config.conf[_streamName]["sampleRate"] = sampleRate;
+            config.release(true);
+        }
+
+        if (ImGui::Checkbox(CONCAT("Stereo##_network_sink_stereo_", _streamName), &stereo)) {
+            stop();
+            start();
+            config.acquire();
+            config.conf[_streamName]["stereo"] = stereo;
+            config.release(true);
+        }
+
+        if (listening && ImGui::Button(CONCAT("Stop##_network_sink_stop_", _streamName), ImVec2(menuWidth, 0))) {
+            stopServer();
+            config.acquire();
+            config.conf[_streamName]["listening"] = false;
+            config.release(true);
+        }
+        else if (!listening && ImGui::Button(CONCAT("Start##_network_sink_stop_", _streamName), ImVec2(menuWidth, 0))) {
+            startServer();
+            config.acquire();
+            config.conf[_streamName]["listening"] = true;
+            config.release(true);
+        }
+
+        ImGui::Text("Status:");
+        ImGui::SameLine();
+        if (conn && conn->isOpen()) {
+            ImGui::TextColored(ImVec4(0.0, 1.0, 0.0, 1.0), (modeId == SINK_MODE_TCP) ? "Connected" : "Sending");
+        }
+        else if (listening) {
+            ImGui::TextColored(ImVec4(1.0, 1.0, 0.0, 1.0), "Listening");
+        }
+        else {
+            ImGui::Text("Idle");
+        }
+    }
+
+private:
+    void doStart() {
+        packer.start();
+        if (stereo) {
+            stereoSink.start();
+        }
+        else {
+            spdlog::warn("Starting");
+            s2m.start();
+            monoSink.start();
+        }
+    }
+
+    void doStop() {
+        packer.stop();
+        s2m.stop();
+        monoSink.stop();
+        stereoSink.stop();
+    }
+
+    void startServer() {
+        if (modeId == SINK_MODE_TCP) {
+            listener = net::listen(hostname, port);
+            if (listener) {
+                listener->acceptAsync(clientHandler, this);
+            }
+        }
+        else {
+            conn = net::openUDP("0.0.0.0", port, hostname, port, false);
+        }
+    }
+    
+    void stopServer() {
+        if (conn) { conn->close(); }
+        if (listener) { listener->close(); }
+    }
+
+    static void monoHandler(float* samples, int count, void* ctx) {
+        NetworkSink* _this = (NetworkSink*)ctx;
+        std::lock_guard lck(_this->connMtx);
+        if (!_this->conn || !_this->conn->isOpen()) { return; }
+        
+        volk_32f_s32f_convert_16i(_this->netBuf, (float*)samples, 32768.0f, count);
+
+        _this->conn->write(count*sizeof(int16_t), (uint8_t*)_this->netBuf);
+    }
+
+    static void stereoHandler(dsp::stereo_t* samples, int count, void* ctx) {
+        NetworkSink* _this = (NetworkSink*)ctx;
+        std::lock_guard lck(_this->connMtx);
+        if (!_this->conn || !_this->conn->isOpen()) { return; }
+
+        volk_32f_s32f_convert_16i(_this->netBuf, (float*)samples, 32768.0f, count*2);
+
+        _this->conn->write(count*2*sizeof(int16_t), (uint8_t*)_this->netBuf);
+    }
+
+    static void clientHandler(net::Conn client, void* ctx) {
+        NetworkSink* _this = (NetworkSink*)ctx;
+
+        {
+            std::lock_guard lck(_this->connMtx);
+            _this->conn = std::move(client);
+        }
+
+        if (_this->conn) {
+            _this->conn->waitForEnd();
+            _this->conn->close();
+        }
+        else {
+            
+        }
+
+        _this->listener->acceptAsync(clientHandler, _this);
+    }
+    
+    SinkManager::Stream* _stream;
+    dsp::Packer<dsp::stereo_t> packer;
+    dsp::StereoToMono s2m;
+    dsp::HandlerSink<float> monoSink;
+    dsp::HandlerSink<dsp::stereo_t> stereoSink;
+
+    std::string _streamName;
+
+    int srId = 0;
+    bool running = false;
+
+    char hostname[1024];
+    int port = 4242;
+
+    int modeId = 1;
+
+    std::vector<unsigned int> sampleRates;
+    std::string sampleRatesTxt;
+    unsigned int sampleRate = 48000;
+    bool stereo = false;
+
+    int16_t* netBuf;
+
+    net::Listener listener;
+    net::Conn conn;
+    std::mutex connMtx;
+
+};
+
+class NetworkSinkModule : public ModuleManager::Instance {
+public:
+    NetworkSinkModule(std::string name) {
+        this->name = name;
+        provider.create = create_sink;
+        provider.ctx = this;
+
+        sigpath::sinkManager.registerSinkProvider("Network", provider);
+    }
+
+    ~NetworkSinkModule() {
+        // Unregister sink, this will automatically stop and delete all instances of the audio sink
+        sigpath::sinkManager.unregisterSinkProvider("Network");
+    }
+
+    void postInit() {}
+
+    void enable() {
+        enabled = true;
+    }
+
+    void disable() {
+        enabled = false;
+    }
+
+    bool isEnabled() {
+        return enabled;
+    }
+
+private:
+    static SinkManager::Sink* create_sink(SinkManager::Stream* stream, std::string streamName, void* ctx) {
+        return (SinkManager::Sink*)(new NetworkSink(stream, streamName));
+    }
+
+    std::string name;
+    bool enabled = true;
+    SinkManager::SinkProvider provider;
+
+};
+
+MOD_EXPORT void _INIT_() {
+    json def = json({});
+    config.setPath(options::opts.root + "/network_sink_config.json");
+    config.load(def);
+    config.enableAutoSave();
+}
+
+MOD_EXPORT void* _CREATE_INSTANCE_(std::string name) {
+    NetworkSinkModule* instance = new NetworkSinkModule(name);
+    return instance;
+}
+
+MOD_EXPORT void _DELETE_INSTANCE_(void* instance) {
+    delete (NetworkSinkModule*)instance;
+}
+
+MOD_EXPORT void _END_() {
+    config.disableAutoSave();
+    config.save();
+}

readme.md

@@ -287,6 +287,7 @@ Modules in beta are still included in releases for the most part but not enabled
 | Name               | Stage      | Dependencies | Option                       | Built by default| Built in Release | Enabled in SDR++ by default |
 |--------------------|------------|--------------|------------------------------|:---------------:|:----------------:|:---------------------------:|
 | audio_sink         | Working    | rtaudio      | OPT_BUILD_AUDIO_SINK         | ✅              | ✅              | ✅                         |
+| network_sink       | Beta       | -            | OPT_BUILD_NETWORK_SINK       | ✅              | ✅              | ✅                         |
 | new_portaudio_sink | Beta       | portaudio    | OPT_BUILD_NEW_PORTAUDIO_SINK | ⛔              | ✅              | ⛔                         |
 
 ## Decoders

rigctl_server/src/main.cpp

@@ -200,7 +200,7 @@ private:
 
     void startServer() {
         try {
-            listener = net::listen(net::PROTO_TCP, hostname, port);
+            listener = net::listen(hostname, port);
             listener->acceptAsync(clientHandler, this);
         }
         catch (std::exception e) {

spyserver_source/src/spyserver_client.cpp

@@ -163,7 +163,7 @@ namespace spyserver {
     }
 
     SpyServerClient connect(std::string host, uint16_t port, dsp::stream<dsp::complex_t>* out) {
-        net::Conn conn = net::connect(net::PROTO_TCP, host, port);
+        net::Conn conn = net::connect(host, port);
         if (!conn) {
             return NULL;
         }