diff options
| -rw-r--r-- | src/DabMod.cpp | 42 | ||||
| -rw-r--r-- | src/InputZeroMQReader.cpp | 26 |
2 files changed, 47 insertions, 21 deletions
diff --git a/src/DabMod.cpp b/src/DabMod.cpp index ce2f249..8a0ee03 100644 --- a/src/DabMod.cpp +++ b/src/DabMod.cpp @@ -92,18 +92,12 @@ void signalHandler(int signalNb) struct modulator_data { - modulator_data() : - inputReader(nullptr), - framecount(0), - flowgraph(nullptr), - etiReader(nullptr) {} - - InputReader* inputReader; + std::shared_ptr<InputReader> inputReader; Buffer data; - uint64_t framecount; + uint64_t framecount = 0; - Flowgraph* flowgraph; - EtiReader* etiReader; + Flowgraph* flowgraph = nullptr; + EtiReader* etiReader = nullptr; }; enum class run_modulator_state_t { @@ -113,7 +107,7 @@ enum class run_modulator_state_t { reconfigure // Some sort of change of configuration we cannot handle happened }; -run_modulator_state_t run_modulator(modulator_data& m); +static run_modulator_state_t run_modulator(modulator_data& m); static void printModSettings(const mod_settings_t& mod_settings) { @@ -271,8 +265,6 @@ int launch_modulator(int argc, char* argv[]) printModSettings(mod_settings); - modulator_data m; - shared_ptr<FormatConverter> format_converter; if (mod_settings.useFileOutput and (mod_settings.fileOutputFormat == "s8" or @@ -379,7 +371,8 @@ int launch_modulator(int argc, char* argv[]) while (run_again) { Flowgraph flowgraph; - m.inputReader = inputReader.get(); + modulator_data m; + m.inputReader = inputReader; m.flowgraph = &flowgraph; m.data.setLength(6144); @@ -461,7 +454,7 @@ int launch_modulator(int argc, char* argv[]) return ret; } -run_modulator_state_t run_modulator(modulator_data& m) +static run_modulator_state_t run_modulator(modulator_data& m) { auto ret = run_modulator_state_t::failure; try { @@ -498,13 +491,26 @@ run_modulator_state_t run_modulator(modulator_data& m) } } if (framesize == 0) { - etiLog.level(info) << "End of file reached."; + if (dynamic_pointer_cast<InputFileReader>(m.inputReader)) { + etiLog.level(info) << "End of file reached."; + running = 0; + ret = run_modulator_state_t::normal_end; + } +#if defined(HAVE_ZEROMQ) + else if (dynamic_pointer_cast<InputZeroMQReader>(m.inputReader)) { + /* An empty frame marks a timeout. We ignore it, but we are + * now able to handle SIGINT properly. + */ + } +#endif // defined(HAVE_ZEROMQ) + // No need to handle the TCP input in a special way to get SIGINT working, + // because recv() will return with EINTR. } else { etiLog.level(error) << "Input read error."; + running = 0; + ret = run_modulator_state_t::normal_end; } - running = 0; - ret = run_modulator_state_t::normal_end; } } catch (const zmq_input_overflow& e) { diff --git a/src/InputZeroMQReader.cpp b/src/InputZeroMQReader.cpp index aa342d5..f6a816a 100644 --- a/src/InputZeroMQReader.cpp +++ b/src/InputZeroMQReader.cpp @@ -96,8 +96,6 @@ int InputZeroMQReader::Open(const string& uri, size_t max_queued_frames) int InputZeroMQReader::GetNextFrame(void* buffer) { - const size_t framesize = 6144; - if (not m_running) { return 0; } @@ -123,7 +121,17 @@ int InputZeroMQReader::GetNextFrame(void* buffer) throw zmq_input_overflow(); } - memcpy(buffer, &incoming->front(), framesize); + + const size_t framesize = 6144; + if (incoming->empty()) { + return 0; + } + else if (incoming->size() == framesize) { + memcpy(buffer, &incoming->front(), framesize); + } + else { + throw logic_error("ZMQ ETI not 6144"); + } return framesize; } @@ -170,6 +178,18 @@ void InputZeroMQReader::RecvProcess() if (success) try { while (m_running) { zmq::message_t incoming; + zmq::pollitem_t items[1]; + items[0].socket = subscriber; + items[0].events = ZMQ_POLLIN; + const int zmq_timeout_ms = 100; + const int num_events = zmq::poll(items, 1, zmq_timeout_ms); + if (num_events == 0) { + // timeout is signalled by an empty buffer + auto buf = make_shared<vector<uint8_t> >(); + m_in_messages.push(buf); + continue; + } + subscriber.recv(&incoming); if (m_to_drop) { |