1 files changed, 52 insertions, 14 deletions

diff --git a/main.cpp b/main.cpp
index 7c27b80..5803262 100644
--- a/main.cpp
+++ b/main.cpp
@@ -23,6 +23,7 @@
 
 #include "OutputUHD.hpp"
 #include "utils.hpp"
+#include <zmq.hpp>
 
 size_t read_samples(FILE* fd, std::vector<complexf>& samples, size_t count)
 {
@@ -31,10 +32,10 @@ size_t read_samples(FILE* fd, std::vector<complexf>& samples, size_t count)
         samples.resize(count);
     }
 
-    size_t num_read = fread((char*)&samples.front(), sizeof(complexf), count, fd);
+    size_t num_read = fread(&samples.front(), sizeof(complexf), count, fd);
     if (num_read == 0) {
         rewind(fd);
-        num_read = fread((char*)&samples.front(), sizeof(complexf), count, fd);
+        num_read = fread(&samples.front(), sizeof(complexf), count, fd);
     }
 
     return num_read;
@@ -44,23 +45,41 @@ int main(int argc, char **argv)
 {
     double txgain = 0;
 
-    if (argc == 1) {
-        txgain = strtod(argv[1], nullptr);
+    if (argc == 3) {
+        txgain = strtod(argv[2], nullptr);
         if (!(0 <= txgain and txgain < 80)) {
             MDEBUG("txgain wrong: %f\n", txgain);
             return -1;
         }
     }
 
+    MDEBUG("TX Gain is %f\n", txgain);
+
+    if (argc < 2) {
+        MDEBUG("Require input file or url\n");
+        return -1;
+    }
+
+    std::string uri = argv[1];
+
     const size_t samps_per_buffer = 20480;
 
-    OutputUHD output_uhd;
-    output_uhd.Init(txgain);
+    OutputUHD output_uhd(txgain);
 
-    FILE* fd = fopen("input.iq", "rb");
-    if (!fd) {
-        MDEBUG("Could not open file\n");
-        return -1;
+    zmq::context_t ctx;
+    zmq::socket_t zmq_sock(ctx, ZMQ_SUB);
+
+    FILE* fd = nullptr;
+    if (uri.find("tcp://") != 0) {
+        fd = fopen(uri.c_str(), "rb");
+        if (!fd) {
+            MDEBUG("Could not open file\n");
+            return -1;
+        }
+    }
+    else {
+        zmq_sock.connect(uri);
+        zmq_sock.setsockopt(ZMQ_SUBSCRIBE, NULL, 0);
     }
 
     std::vector<complexf> input_samples(samps_per_buffer);
@@ -68,12 +87,31 @@ int main(int argc, char **argv)
     size_t total_samps_read = samps_read;
 
     double last_print_time = 0;
+    size_t sent = 0;
 
     do {
-        samps_read = read_samples(fd, input_samples, samps_per_buffer);
-
         double first_sample_time = 0;
-        output_uhd.Transmit(input_samples, samps_read, &first_sample_time);
+
+        if (fd) {
+            samps_read = read_samples(fd, input_samples, samps_per_buffer);
+            sent = output_uhd.Transmit(&input_samples.front(), samps_read, &first_sample_time);
+        }
+        else {
+            zmq::message_t msg;
+            if (not zmq_sock.recv(&msg)) {
+                MDEBUG("zmq recv error\n");
+                return -1;
+            }
+
+            if (msg.size() % sizeof(complexf) != 0) {
+                MDEBUG("Received incomplete size %zu\n", msg.size());
+                return -1;
+            }
+            samps_read = msg.size() / sizeof(complexf);
+
+            sent = output_uhd.Transmit((complexf*)msg.data(), samps_read, &first_sample_time);
+        }
+
         if (first_sample_time - last_print_time > 1) {
             MDEBUG("Tx %zu samples at t=%f\n", samps_read, first_sample_time);
             last_print_time = first_sample_time;
@@ -81,7 +119,7 @@ int main(int argc, char **argv)
 
         total_samps_read += samps_read;
     }
-    while (samps_read);
+    while (samps_read and sent);
 
     MDEBUG("Read %zu samples in total\n", total_samps_read);
 }