Merge remote-tracking branch 'upstream/master'

Conflicts:
	src/InputFileReader.cpp

1 files changed, 266 insertions, 210 deletions

diff --git a/src/DabMod.cpp b/src/DabMod.cpp
index 91c0b9d..ec1a4cd 100644
--- a/src/DabMod.cpp
+++ b/src/DabMod.cpp
@@ -3,7 +3,7 @@
    Her Majesty the Queen in Right of Canada (Communications Research
    Center Canada)
 
-   Copyright (C) 2014
+   Copyright (C) 2014, 2015
    Matthias P. Braendli, matthias.braendli@mpb.li
 
     http://opendigitalradio.org
@@ -30,11 +30,12 @@
 #endif
 
 #include "porting.h"
-
+#include "Utils.h"
 #include "Log.h"
 #include "DabModulator.h"
 #include "InputMemory.h"
 #include "OutputFile.h"
+#include "FormatConverter.h"
 #if defined(HAVE_OUTPUT_UHD)
 #   include "OutputUHD.h"
 #endif
@@ -45,6 +46,8 @@
 #include "FIRFilter.h"
 #include "RemoteControl.h"
 
+#include <boost/shared_ptr.hpp>
+#include <boost/make_shared.hpp>
 #include <boost/property_tree/ptree.hpp>
 #include <boost/property_tree/ini_parser.hpp>
 #include <complex>
@@ -67,125 +70,78 @@
 #   define memalign(a, b)   malloc(b)
 #endif
 
+#define ZMQ_INPUT_MAX_FRAME_QUEUE 50
+
 
 typedef std::complex<float> complexf;
 
+using namespace boost;
 
-bool running = true;
+volatile sig_atomic_t running = 1;
 
 void signalHandler(int signalNb)
 {
     PDEBUG("signalHandler(%i)\n", signalNb);
 
-    running = false;
+    running = 0;
 }
 
-
-void printUsage(char* progName, FILE* out = stderr)
+struct modulator_data
 {
-    fprintf(out, "Welcome to %s %s, compiled at %s, %s\n\n",
-            PACKAGE,
-#if defined(GITVERSION)
-            GITVERSION,
-#else
-            VERSION,
-#endif
-            __DATE__, __TIME__);
-    fprintf(out, "Usage with configuration file:\n");
-    fprintf(out, "\t%s [-C] config_file.ini\n\n", progName);
-
-    fprintf(out, "Usage with command line options:\n");
-    fprintf(out, "\t%s"
-            " input"
-            " (-f filename | -u uhddevice -F frequency) "
-            " [-G txgain]"
-            " [-o offset]"
-            " [-O offsetfile]"
-            " [-T filter_taps_file]"
-            " [-a gain]"
-            " [-c clockrate]"
-            " [-g gainMode]"
-            " [-h]"
-            " [-l]"
-            " [-m dabMode]"
-            " [-r samplingRate]"
-            "\n", progName);
-    fprintf(out, "Where:\n");
-    fprintf(out, "input:         ETI input filename (default: stdin).\n");
-    fprintf(out, "-f name:       Use file output with given filename. (use /dev/stdout for standard output)\n");
-    fprintf(out, "-u device:     Use UHD output with given device string. (use "" for default device)\n");
-    fprintf(out, "-F frequency:  Set the transmit frequency when using UHD output. (mandatory option when using UHD)\n");
-    fprintf(out, "-G txgain:     Set the transmit gain for the UHD driver (default: 0)\n");
-    fprintf(out, "-o:            (UHD only) Set the timestamp offset added to the timestamp in the ETI. The offset is a double.\n");
-    fprintf(out, "-O:            (UHD only) Set the file containing the timestamp offset added to the timestamp in the ETI.\n"
-                                 "The file is read every six seconds, and must contain a double value.\n");
-    fprintf(out, "                  Specifying either -o or -O has two implications: It enables synchronous transmission,\n"
-                 "                  requiring an external REFCLK and PPS signal and frames that do not contain a valid timestamp\n"
-                 "                  get muted.\n\n");
-    fprintf(out, "-T taps_file:  Enable filtering before the output, using the specified file containing the filter taps.\n");
-    fprintf(out, "-a gain:       Apply digital amplitude gain.\n");
-    fprintf(out, "-c rate:       Set the DAC clock rate and enable Cic Equalisation.\n");
-    fprintf(out, "-g:            Set computation gain mode: "
-            "%u FIX, %u MAX, %u VAR\n", GAIN_FIX, GAIN_MAX, GAIN_VAR);
-    fprintf(out, "-h:            Print this help.\n");
-    fprintf(out, "-l:            Loop file when reach end of file.\n");
-    fprintf(out, "-m mode:       Set DAB mode: (0: auto, 1-4: force).\n");
-    fprintf(out, "-r rate:       Set output sampling rate (default: 2048000).\n");
-}
+    modulator_data() :
+        inputReader(NULL),
+        framecount(0),
+        flowgraph(NULL),
+        rcs(NULL) {}
 
+    InputReader* inputReader;
+    Buffer data;
+    uint64_t framecount;
 
-void printVersion(FILE *out = stderr)
-{
-    fprintf(out, "Welcome to %s %s, compiled at %s, %s\n\n",
-            PACKAGE, VERSION, __DATE__, __TIME__);
-    fprintf(out,
-            "    ODR-DabMod is copyright (C) Her Majesty the Queen in Right of Canada,\n"
-            "    2009, 2010, 2011, 2012 Communications Research Centre (CRC),\n"
-            "     and\n"
-            "    Copyright (C) 2014 Matthias P. Braendli, matthias.braendli@mpb.li\n"
-            "\n"
-            "    http://opendigitalradio.org\n"
-            "\n"
-            "    This program is available free of charge and is licensed to you on a\n"
-            "    non-exclusive basis; you may not redistribute it.\n"
-            "\n"
-            "    This program is provided \"AS IS\" in the hope that it will be useful, but\n"
-            "    WITHOUT ANY WARRANTY with respect to its accurancy or usefulness; witout\n"
-            "    even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR\n"
-            "    PURPOSE and NONINFRINGEMENT.\n"
-            "\n"
-            "    In no event shall CRC be LIABLE for any LOSS, DAMAGE or COST that may be\n"
-            "    incurred in connection with the use of this software.\n"
-            "\n"
-#if USE_KISS_FFT
-            "ODR-DabMod makes use of the following open source packages:\n"
-            "    Kiss FFT v1.2.9 (Revised BSD) - http://kissfft.sourceforge.net/\n"
-#endif
-           );
+    Flowgraph* flowgraph;
+    RemoteControllers* rcs;
+};
 
-}
+enum run_modulator_state {
+    MOD_FAILURE,
+    MOD_NORMAL_END,
+    MOD_AGAIN
+};
 
+run_modulator_state run_modulator(Logger& logger, modulator_data& m);
 
-int main(int argc, char* argv[])
+int launch_modulator(int argc, char* argv[])
 {
     int ret = 0;
     bool loop = false;
     std::string inputName = "";
     std::string inputTransport = "file";
+    unsigned inputMaxFramesQueued = ZMQ_INPUT_MAX_FRAME_QUEUE;
 
     std::string outputName;
     int useZeroMQOutput = 0;
+    std::string zmqOutputSocketType = "";
     int useFileOutput = 0;
+    std::string fileOutputFormat = "complexf";
     int useUHDOutput = 0;
 
-    uint64_t frame = 0;
     size_t outputRate = 2048000;
     size_t clockRate = 0;
     unsigned dabMode = 0;
     float digitalgain = 1.0f;
     float normalise = 1.0f;
     GainMode gainMode = GAIN_VAR;
-    Buffer data;
+
+
+    /* UHD requires the input I and Q samples to be in the interval
+     * [-1.0,1.0], otherwise they get truncated, which creates very
+     * wide-spectrum spikes. Depending on the Transmission Mode, the
+     * Gain Mode and the sample rate (and maybe other parameters), the
+     * samples can have peaks up to about 48000. The value of 50000
+     * should guarantee that with a digital gain of 1.0, UHD never clips
+     * our samples.
+     */
+    const float normalise_factor = 50000.0f;
 
     std::string filterTapsFilename = "";
 
@@ -198,27 +154,37 @@ int main(int argc, char* argv[])
     OutputUHDConfig outputuhd_conf;
 #endif
 
+    modulator_data m;
+
     // To handle the timestamp offset of the modulator
     struct modulator_offset_config modconf;
     modconf.use_offset_file = false;
     modconf.use_offset_fixed = false;
     modconf.delay_calculation_pipeline_stages = 0;
 
-    Flowgraph* flowgraph = NULL;
-    DabModulator* modulator = NULL;
-    InputMemory* input = NULL;
-    ModOutput* output = NULL;
+    shared_ptr<Flowgraph> flowgraph(new Flowgraph());
+    shared_ptr<FormatConverter> format_converter;
+    shared_ptr<ModOutput> output;
+
+    RemoteControllers rcs;
+    m.rcs = &rcs;
 
-    BaseRemoteController* rc = NULL;
+    bool run_again = true;
 
     Logger logger;
     InputFileReader inputFileReader(logger);
-#if defined(HAVE_INPUT_ZEROMQ)
-    InputZeroMQReader inputZeroMQReader(logger);
+#if defined(HAVE_ZEROMQ)
+    shared_ptr<InputZeroMQReader> inputZeroMQReader(new InputZeroMQReader(logger));
 #endif
-    InputReader* inputReader;
 
-    signal(SIGINT, signalHandler);
+    struct sigaction sa;
+    memset(&sa, 0, sizeof(struct sigaction));
+    sa.sa_handler = &signalHandler;
+
+    if (sigaction(SIGINT, &sa, NULL) == -1) {
+        perror("sigaction");
+        return EXIT_FAILURE;
+    }
 
     // Set timezone to UTC
     setenv("TZ", "", 1);
@@ -250,7 +216,7 @@ int main(int argc, char* argv[])
 #if defined(HAVE_OUTPUT_UHD)
             if (useUHDOutput) {
                 fprintf(stderr, "Options -u and -f are mutually exclusive\n");
-                goto END_MAIN;
+                throw std::invalid_argument("Invalid command line options");
             }
 #endif
             outputName = optarg;
@@ -276,7 +242,7 @@ int main(int argc, char* argv[])
             if (modconf.use_offset_file)
             {
                 fprintf(stderr, "Options -o and -O are mutually exclusive\n");
-                goto END_MAIN;
+                throw std::invalid_argument("Invalid command line options");
             }
             modconf.use_offset_fixed = true;
             modconf.offset_fixed = strtod(optarg, NULL);
@@ -288,7 +254,7 @@ int main(int argc, char* argv[])
             if (modconf.use_offset_fixed)
             {
                 fprintf(stderr, "Options -o and -O are mutually exclusive\n");
-                goto END_MAIN;
+                throw std::invalid_argument("Invalid command line options");
             }
             modconf.use_offset_file = true;
             modconf.offset_filename = std::string(optarg);
@@ -309,7 +275,7 @@ int main(int argc, char* argv[])
 #if defined(HAVE_OUTPUT_UHD)
             if (useFileOutput) {
                 fprintf(stderr, "Options -u and -f are mutually exclusive\n");
-                goto END_MAIN;
+                throw std::invalid_argument("Invalid command line options");
             }
             outputuhd_conf.device = optarg;
             useUHDOutput = 1;
@@ -317,17 +283,17 @@ int main(int argc, char* argv[])
             break;
         case 'V':
             printVersion();
-            goto END_MAIN;
+            throw std::invalid_argument("");
             break;
         case '?':
         case 'h':
             printUsage(argv[0]);
-            goto END_MAIN;
+            throw std::invalid_argument("");
             break;
         default:
             fprintf(stderr, "Option '%c' not coded yet!\n", c);
             ret = -1;
-            goto END_MAIN;
+            throw std::invalid_argument("Invalid command line options");
         }
     }
 
@@ -352,15 +318,12 @@ int main(int argc, char* argv[])
         "\n";
 
     std::cerr << "Compiled with features: " <<
-#if defined(HAVE_INPUT_ZEROMQ)
-        "input_zeromq " <<
+#if defined(HAVE_ZEROMQ)
+        "zeromq " <<
 #endif
 #if defined(HAVE_OUTPUT_UHD)
         "output_uhd " <<
 #endif
-#if defined(HAVE_OUTPUT_ZEROMQ)
-        "output_zeromq " <<
-#endif
         "\n";
 
     if (use_configuration_file && use_configuration_cmdline) {
@@ -371,7 +334,7 @@ int main(int argc, char* argv[])
     // No argument given ? You can't be serious ! Show usage.
     if (argc == 1) {
         printUsage(argv[0]);
-        goto END_MAIN;
+        throw std::invalid_argument("Invalid command line options");
     }
 
     // If only one argument is given, interpret as configuration file name
@@ -390,8 +353,9 @@ int main(int argc, char* argv[])
         }
         catch (boost::property_tree::ini_parser::ini_parser_error &e)
         {
-            fprintf(stderr, "Error, cannot read configuration file '%s'\n", configuration_file.c_str());
-            goto END_MAIN;
+            std::cerr << "Error, cannot read configuration file '" << configuration_file.c_str() << "'" << std::endl;
+            std::cerr << "       " << e.what() << std::endl;
+            throw std::runtime_error("Cannot read configuration file");
         }
 
         // remote controller:
@@ -399,14 +363,30 @@ int main(int argc, char* argv[])
             try {
                 int telnetport = pt.get<int>("remotecontrol.telnetport");
                 RemoteControllerTelnet* telnetrc = new RemoteControllerTelnet(telnetport);
-                rc = telnetrc;
+                rcs.add_controller(telnetrc);
             }
             catch (std::exception &e) {
                 std::cerr << "Error: " << e.what() << "\n";
                 std::cerr << "       telnet remote control enabled, but no telnetport defined.\n";
-                goto END_MAIN;
+                throw std::runtime_error("Configuration error");
+            }
+        }
+
+#if defined(HAVE_ZEROMQ)
+        if (pt.get("remotecontrol.zmqctrl", 0) == 1) {
+            try {
+                std::string zmqCtrlEndpoint = pt.get("remotecontrol.zmqctrlendpoint", "");
+                std::cerr << "ZmqCtrlEndpoint: " << zmqCtrlEndpoint << std::endl;
+                RemoteControllerZmq* zmqrc = new RemoteControllerZmq(zmqCtrlEndpoint);
+                rcs.add_controller(zmqrc);
+            }
+            catch (std::exception &e) {
+                std::cerr << "Error: " << e.what() << "\n";
+                std::cerr << "       zmq remote control enabled, but no endpoint defined.\n";
+                throw std::runtime_error("Configuration error");
             }
         }
+#endif
 
         // input params:
         if (pt.get("input.loop", 0) == 1) {
@@ -414,6 +394,9 @@ int main(int argc, char* argv[])
         }
 
         inputTransport = pt.get("input.transport", "file");
+        inputMaxFramesQueued = pt.get("input.max_frames_queued",
+                ZMQ_INPUT_MAX_FRAME_QUEUE);
+
         inputName = pt.get("input.source", "/dev/stdin");
 
         // log parameters:
@@ -430,7 +413,7 @@ int main(int argc, char* argv[])
             catch (std::exception &e) {
                 std::cerr << "Error: " << e.what() << "\n";
                 std::cerr << "       Configuration enables file log, but does not specify log filename\n";
-                goto END_MAIN;
+                throw std::runtime_error("Configuration error");
             }
 
             LogToFile* log_file = new LogToFile(logfilename);
@@ -453,7 +436,7 @@ int main(int argc, char* argv[])
             catch (std::exception &e) {
                 std::cerr << "Error: " << e.what() << "\n";
                 std::cerr << "       Configuration enables firfilter, but does not specify filter taps file\n";
-                goto END_MAIN;
+                throw std::runtime_error("Configuration error");
             }
         }
 
@@ -465,7 +448,7 @@ int main(int argc, char* argv[])
         catch (std::exception &e) {
             std::cerr << "Error: " << e.what() << "\n";
             std::cerr << "       Configuration does not specify output\n";
-            goto END_MAIN;
+            throw std::runtime_error("Configuration error");
         }
 
         if (output_selected == "file") {
@@ -475,9 +458,11 @@ int main(int argc, char* argv[])
             catch (std::exception &e) {
                 std::cerr << "Error: " << e.what() << "\n";
                 std::cerr << "       Configuration does not specify file name for file output\n";
-                goto END_MAIN;
+                throw std::runtime_error("Configuration error");
             }
             useFileOutput = 1;
+
+            fileOutputFormat = pt.get("fileoutput.format", fileOutputFormat);
         }
 #if defined(HAVE_OUTPUT_UHD)
         else if (output_selected == "uhd") {
@@ -499,10 +484,11 @@ int main(int argc, char* argv[])
             outputuhd_conf.txgain = pt.get("uhdoutput.txgain", 0.0);
             outputuhd_conf.frequency = pt.get<double>("uhdoutput.frequency", 0);
             std::string chan = pt.get<std::string>("uhdoutput.channel", "");
+            outputuhd_conf.dabMode = dabMode;
 
             if (outputuhd_conf.frequency == 0 && chan == "") {
                 std::cerr << "       UHD output enabled, but neither frequency nor channel defined.\n";
-                goto END_MAIN;
+                throw std::runtime_error("Configuration error");
             }
             else if (outputuhd_conf.frequency == 0) {
                 double freq;
@@ -546,13 +532,13 @@ int main(int argc, char* argv[])
                 else if (chan == "13F") freq = 239200000;
                 else {
                     std::cerr << "       UHD output: channel " << chan << " does not exist in table\n";
-                    goto END_MAIN;
+                    throw std::out_of_range("UHD channel selection error");
                 }
                 outputuhd_conf.frequency = freq;
             }
             else if (outputuhd_conf.frequency != 0 && chan != "") {
                 std::cerr << "       UHD output: cannot define both frequency and channel.\n";
-                goto END_MAIN;
+                throw std::runtime_error("Configuration error");
             }
 
 
@@ -570,21 +556,22 @@ int main(int argc, char* argv[])
             }
             else {
                 std::cerr << "Error: UHD output: behaviour_refclk_lock_lost invalid." << std::endl;
-                goto END_MAIN;
+                throw std::runtime_error("Configuration error");
             }
 
             useUHDOutput = 1;
         }
 #endif
-#if defined(HAVE_OUTPUT_ZEROMQ)
+#if defined(HAVE_ZEROMQ)
         else if (output_selected == "zmq") {
             outputName = pt.get<std::string>("zmqoutput.listen");
+            zmqOutputSocketType = pt.get<std::string>("zmqoutput.socket_type");
             useZeroMQOutput = 1;
         }
 #endif
         else {
             std::cerr << "Error: Invalid output defined.\n";
-            goto END_MAIN;
+            throw std::runtime_error("Configuration error");
         }
 
 #if defined(HAVE_OUTPUT_UHD)
@@ -607,7 +594,7 @@ int main(int argc, char* argv[])
             catch (std::exception &e) {
                 std::cerr << "Error: " << e.what() << "\n";
                 std::cerr << "       Synchronised transmission enabled, but delay management specification is incomplete.\n";
-                goto END_MAIN;
+                throw std::runtime_error("Configuration error");
             }
         }
 
@@ -615,9 +602,9 @@ int main(int argc, char* argv[])
 #endif
     }
 
-    if (!rc) {
+    if (rcs.get_no_controllers() == 0) {
         logger.level(warn) << "No Remote-Control started";
-        rc = new RemoteControllerDummy();
+        rcs.add_controller(new RemoteControllerDummy());
     }
 
 
@@ -661,13 +648,13 @@ int main(int argc, char* argv[])
         printUsage(argv[0]);
         ret = -1;
         logger.level(error) << "Received invalid command line arguments";
-        goto END_MAIN;
+        throw std::invalid_argument("Invalid command line options");
     }
 
     if (!useFileOutput && !useUHDOutput && !useZeroMQOutput) {
         logger.level(error) << "Output not specified";
         fprintf(stderr, "Must specify output !");
-        goto END_MAIN;
+        throw std::runtime_error("Configuration error");
     }
 
     // Print settings
@@ -692,8 +679,10 @@ int main(int argc, char* argv[])
 #endif
     else if (useZeroMQOutput) {
         fprintf(stderr, " ZeroMQ\n"
-                        "  Listening on: %s\n",
-                        outputName.c_str());
+                        "  Listening on: %s\n"
+                        "  Socket type : %s\n",
+                        outputName.c_str(),
+                        zmqOutputSocketType.c_str());
     }
 
     fprintf(stderr, "  Sampling rate: ");
@@ -713,88 +702,146 @@ int main(int argc, char* argv[])
             fprintf(stderr, "Unable to open input file!\n");
             logger.level(error) << "Unable to open input file!";
             ret = -1;
-            goto END_MAIN;
+            throw std::runtime_error("Unable to open input");
         }
 
-        inputReader = &inputFileReader;
+        m.inputReader = &inputFileReader;
     }
     else if (inputTransport == "zeromq") {
-#if !defined(HAVE_INPUT_ZEROMQ)
+#if !defined(HAVE_ZEROMQ)
         fprintf(stderr, "Error, ZeroMQ input transport selected, but not compiled in!\n");
         ret = -1;
-        goto END_MAIN;
+        throw std::runtime_error("Unable to open input");
 #else
-        // The URL might start with zmq+tcp://
-        if (inputName.substr(0, 4) == "zmq+") {
-            inputZeroMQReader.Open(inputName.substr(4));
-        }
-        else {
-            inputZeroMQReader.Open(inputName);
-        }
-        inputReader = &inputZeroMQReader;
+        inputZeroMQReader->Open(inputName, inputMaxFramesQueued);
+        m.inputReader = inputZeroMQReader.get();
 #endif
     }
     else
     {
         fprintf(stderr, "Error, invalid input transport %s selected!\n", inputTransport.c_str());
         ret = -1;
-        goto END_MAIN;
+        throw std::runtime_error("Unable to open input");
     }
 
-
     if (useFileOutput) {
-        // Opening COFDM output file
-        output = new OutputFile(outputName);
+        if (fileOutputFormat == "complexf") {
+            output = make_shared<OutputFile>(outputName);
+        }
+        else if (fileOutputFormat == "s8") {
+            // We must normalise the samples to the interval [-127.0; 127.0]
+            normalise = 127.0f / normalise_factor;
+
+            format_converter = make_shared<FormatConverter>();
+
+            output = make_shared<OutputFile>(outputName);
+        }
     }
 #if defined(HAVE_OUTPUT_UHD)
     else if (useUHDOutput) {
-
-        /* UHD requires the input I and Q samples to be in the interval
-         * [-1.0,1.0], otherwise they get truncated, which creates very
-         * wide-spectrum spikes. Depending on the Transmission Mode, the
-         * Gain Mode and the sample rate (and maybe other parameters), the
-         * samples can have peaks up to about 48000. The value of 50000
-         * should guarantee that with a digital gain of 1.0, UHD never clips
-         * our samples.
-         */
-        normalise = 1.0f/50000.0f;
-
+        normalise = 1.0f / normalise_factor;
         outputuhd_conf.sampleRate = outputRate;
-        try {
-            output = new OutputUHD(outputuhd_conf, logger);
-            ((OutputUHD*)output)->enrol_at(*rc);
-        }
-        catch (std::exception& e) {
-            logger.level(error) << "UHD initialisation failed:" << e.what();
-            goto END_MAIN;
-        }
+        output = make_shared<OutputUHD>(outputuhd_conf, &logger);
+        ((OutputUHD*)output.get())->enrol_at(rcs);
     }
 #endif
-#if defined(HAVE_OUTPUT_ZEROMQ)
+#if defined(HAVE_ZEROMQ)
     else if (useZeroMQOutput) {
         /* We normalise the same way as for the UHD output */
-        normalise = 1.0f/50000.0f;
-
-        output = new OutputZeroMQ(outputName);
+        normalise = 1.0f / normalise_factor;
+        if (zmqOutputSocketType == "pub") {
+            output = make_shared<OutputZeroMQ>(outputName, ZMQ_PUB);
+        }
+        else if (zmqOutputSocketType == "rep") {
+            output = make_shared<OutputZeroMQ>(outputName, ZMQ_REP);
+        }
+        else {
+            std::stringstream ss;
+            ss << "ZeroMQ output socket type " << zmqOutputSocketType << " invalid";
+            throw std::invalid_argument(ss.str());
+        }
     }
 #endif
 
-    flowgraph = new Flowgraph();
-    data.setLength(6144);
-    input = new InputMemory(&data);
-    modulator = new DabModulator(modconf, rc, logger, outputRate, clockRate,
-            dabMode, gainMode, digitalgain, normalise, filterTapsFilename);
-    flowgraph->connect(input, modulator);
-    flowgraph->connect(modulator, output);
+
+    while (run_again) {
+        Flowgraph flowgraph;
+
+        m.flowgraph = &flowgraph;
+        m.data.setLength(6144);
+
+        shared_ptr<InputMemory> input(new InputMemory(&m.data));
+        shared_ptr<DabModulator> modulator(
+                new DabModulator(modconf, &rcs, logger, outputRate, clockRate,
+                    dabMode, gainMode, digitalgain, normalise, filterTapsFilename));
+
+        flowgraph.connect(input, modulator);
+        if (format_converter) {
+            flowgraph.connect(modulator, format_converter);
+            flowgraph.connect(format_converter, output);
+        }
+        else {
+            flowgraph.connect(modulator, output);
+        }
 
 #if defined(HAVE_OUTPUT_UHD)
-    if (useUHDOutput) {
-        ((OutputUHD*)output)->setETIReader(modulator->getEtiReader());
-    }
+        if (useUHDOutput) {
+            ((OutputUHD*)output.get())->setETIReader(modulator->getEtiReader());
+        }
+#endif
+
+        m.inputReader->PrintInfo();
+
+        run_modulator_state st = run_modulator(logger, m);
+
+        switch (st) {
+            case MOD_FAILURE:
+                fprintf(stderr, "\nModulator failure.\n");
+                run_again = false;
+                ret = 1;
+                break;
+#if defined(HAVE_ZEROMQ)
+            case MOD_AGAIN:
+                fprintf(stderr, "\nRestart modulator\n");
+                running = true;
+                if (inputTransport == "zeromq") {
+                    run_again = true;
+
+                    // Create a new input reader
+                    inputZeroMQReader = make_shared<InputZeroMQReader>(logger);
+                    inputZeroMQReader->Open(inputName, inputMaxFramesQueued);
+                    m.inputReader = inputZeroMQReader.get();
+                }
+                break;
 #endif
+            case MOD_NORMAL_END:
+            default:
+                fprintf(stderr, "\nModulator stopped.\n");
+                ret = 0;
+                run_again = false;
+                break;
+        }
+
+        fprintf(stderr, "\n\n");
+        fprintf(stderr, "%lu DAB frames encoded\n", m.framecount);
+        fprintf(stderr, "%f seconds encoded\n", (float)m.framecount * 0.024f);
+
+        fprintf(stderr, "\nCleaning flowgraph...\n");
+
+        m.data.setLength(0);
+    }
 
-    inputReader->PrintInfo();
+    ////////////////////////////////////////////////////////////////////////
+    // Cleaning things
+    ////////////////////////////////////////////////////////////////////////
 
+    logger.level(info) << "Terminating";
+    return ret;
+}
+
+run_modulator_state run_modulator(Logger& logger, modulator_data& m)
+{
+    run_modulator_state ret = MOD_FAILURE;
     try {
         while (running) {
 
@@ -803,59 +850,68 @@ int main(int argc, char* argv[])
             PDEBUG("*****************************************\n");
             PDEBUG("* Starting main loop\n");
             PDEBUG("*****************************************\n");
-            while ((framesize = inputReader->GetNextFrame(data.getData())) > 0) {
+            while ((framesize = m.inputReader->GetNextFrame(m.data.getData())) > 0) {
                 if (!running) {
                     break;
                 }
 
-                frame++;
+                m.framecount++;
 
                 PDEBUG("*****************************************\n");
-                PDEBUG("* Read frame %lu\n", frame);
+                PDEBUG("* Read frame %lu\n", m.framecount);
                 PDEBUG("*****************************************\n");
 
                 ////////////////////////////////////////////////////////////////
-                // Proccessing data
+                // Processing data
                 ////////////////////////////////////////////////////////////////
-                flowgraph->run();
+                m.flowgraph->run();
 
                 /* Check every once in a while if the remote control
                  * is still working */
-                if (rc && (frame % 250) == 0 && rc->fault_detected()) {
-                    fprintf(stderr,
-                            "Detected Remote Control fault, restarting it\n");
-                    rc->restart();
+                if (m.rcs->get_no_controllers() > 0 && (m.framecount % 250) == 0) {
+                    m.rcs->check_faults();
                 }
             }
             if (framesize == 0) {
-                fprintf(stderr, "End of file reached.\n");
+                logger.level(info) << "End of file reached.";
             }
             else {
-                fprintf(stderr, "Input read error.\n");
+                logger.level(error) << "Input read error.";
             }
-            running = false;
+            running = 0;
+            ret = MOD_NORMAL_END;
         }
+#if defined(HAVE_OUTPUT_UHD)
+    } catch (fct_discontinuity_error& e) {
+        // The OutputUHD saw a FCT discontinuity
+        logger.level(warn) << e.what();
+        ret = MOD_AGAIN;
+#endif
+    } catch (zmq_input_overflow& e) {
+        // The ZeroMQ input has overflowed its buffer
+        logger.level(warn) << e.what();
+        ret = MOD_AGAIN;
     } catch (std::exception& e) {
-        fprintf(stderr, "EXCEPTION: %s\n", e.what());
-        ret = -1;
+        logger.level(error) << "Exception caught: " << e.what();
+        ret = MOD_FAILURE;
     }
 
-END_MAIN:
-    ////////////////////////////////////////////////////////////////////////
-    // Cleaning things
-    ////////////////////////////////////////////////////////////////////////
-    fprintf(stderr, "\n\n");
-    fprintf(stderr, "%lu DAB frames encoded\n", frame);
-    fprintf(stderr, "%f seconds encoded\n", (float)frame * 0.024f);
-
-    fprintf(stderr, "\nCleaning flowgraph...\n");
-    delete flowgraph;
-
-    // Cif
-    fprintf(stderr, "\nCleaning buffers...\n");
-
-    logger.level(info) << "Terminating";
-
     return ret;
 }
 
+int main(int argc, char* argv[])
+{
+    try {
+        return launch_modulator(argc, argv);
+    }
+    catch (std::invalid_argument& e) {
+        std::string what(e.what());
+        if (not what.empty()) {
+            std::cerr << "Modulator error: " << what << std::endl;
+        }
+    }
+    catch (std::runtime_error& e) {
+        std::cerr << "Modulator runtime error: " << e.what() << std::endl;
+    }
+}
+