From 128768f7fd719eb455a946a0f716d7128b4ded63 Mon Sep 17 00:00:00 2001 From: "Matthias P. Braendli" Date: Sun, 30 Oct 2016 15:29:31 +0100 Subject: Start reworking inputs, break all but Prbs and ZMQ --- src/ConfigParser.cpp | 178 +++++++++++++-- src/DabMux.cpp | 18 +- src/DabMux.h | 7 +- src/Makefile.am | 20 +- src/MuxElements.h | 4 +- src/dabInput.h | 15 -- src/dabInputPrbs.cpp | 98 -------- src/dabInputPrbs.h | 52 ----- src/dabInputZmq.cpp | 619 -------------------------------------------------- src/dabInputZmq.h | 276 ----------------------- src/input/Prbs.cpp | 101 +++++++++ src/input/Prbs.h | 56 +++++ src/input/Zmq.cpp | 625 +++++++++++++++++++++++++++++++++++++++++++++++++++ src/input/Zmq.h | 271 ++++++++++++++++++++++ 14 files changed, 1220 insertions(+), 1120 deletions(-) delete mode 100644 src/dabInputPrbs.cpp delete mode 100644 src/dabInputPrbs.h delete mode 100644 src/dabInputZmq.cpp delete mode 100644 src/dabInputZmq.h create mode 100644 src/input/Prbs.cpp create mode 100644 src/input/Prbs.h create mode 100644 src/input/Zmq.cpp create mode 100644 src/input/Zmq.h (limited to 'src') diff --git a/src/ConfigParser.cpp b/src/ConfigParser.cpp index 7e3f855..bdc2099 100644 --- a/src/ConfigParser.cpp +++ b/src/ConfigParser.cpp @@ -48,27 +48,14 @@ #include #include #include "dabOutput/dabOutput.h" -#include "dabInput.h" +#include "input/inputs.h" #include "utils.h" -#include "dabInputFile.h" -#include "dabInputFifo.h" -#include "dabInputMpegFile.h" -#include "dabInputMpegFifo.h" -#include "dabInputDabplusFile.h" -#include "dabInputDabplusFifo.h" -#include "dabInputPacketFile.h" -#include "dabInputEnhancedPacketFile.h" -#include "dabInputEnhancedFifo.h" -#include "dabInputUdp.h" -#include "dabInputPrbs.h" -#include "dabInputRawFile.h" -#include "dabInputRawFifo.h" -#include "dabInputDmbFile.h" -#include "dabInputDmbUdp.h" -#include "dabInputZmq.h" #include "DabMux.h" #include "ManagementServer.h" +#include "input/Prbs.h" +#include "input/Zmq.h" + #ifdef _WIN32 # pragma warning ( disable : 4103 ) @@ -541,13 +528,13 @@ void parse_ptree( } } -static dab_input_zmq_config_t setup_zmq_input( +static Inputs::dab_input_zmq_config_t setup_zmq_input( const boost::property_tree::ptree &pt, const std::string& subchanuid) { using boost::property_tree::ptree_error; - dab_input_zmq_config_t zmqconfig; + Inputs::dab_input_zmq_config_t zmqconfig; try { zmqconfig.buffer_size = pt.get("zmq-buffer"); @@ -621,6 +608,7 @@ static void setup_subchannel_from_ptree(DabSubchannel* subchan, subchan->inputUri = inputUri; +#if OLD_INPUTS // {{{ /* The input is of the old_style type, * with the struct of function pointers, * and needs to be a DabInputCompatible @@ -714,7 +702,7 @@ static void setup_subchannel_from_ptree(DabSubchannel* subchan, } else if (type == "data" and proto == "prbs") { input_is_old_style = false; - subchan->input = new DabInputPrbs(); + subchan->input = make_shared(); subchan->type = subchannel_type_t::DataDmb; subchan->bitrate = DEFAULT_DATA_BITRATE; } else if (type == "data") { @@ -928,5 +916,155 @@ static void setup_subchannel_from_ptree(DabSubchannel* subchan, subchan->input = new DabInputCompatible(operations); } // else { it's already been created! } +#endif // 0 }}} + + dabProtection* protection = &subchan->protection; + + const bool nonblock = pt.get("nonblock", false); + + if (type == "dabplus" or type == "audio") { + subchan->type = subchannel_type_t::Audio; + subchan->bitrate = 0; + + if (proto == "file") { + if (nonblock) { + // TODO + } + } + else if (proto == "tcp" || + proto == "epgm" || + proto == "ipc") { + + if (nonblock) { + etiLog.level(warn) << "The nonblock option is meaningless for the zmq input"; + } + + auto zmqconfig = setup_zmq_input(pt, subchanuid); + + if (type == "audio") { + auto inzmq = make_shared(subchanuid, zmqconfig); + rcs.enrol(inzmq.get()); + subchan->input = inzmq; + } + else if (type == "dabplus") { + auto inzmq = make_shared(subchanuid, zmqconfig); + rcs.enrol(inzmq.get()); + subchan->input = inzmq; + } + + if (proto == "epgm") { + etiLog.level(warn) << "Using untested epgm:// zeromq input"; + } + else if (proto == "ipc") { + etiLog.level(warn) << "Using untested ipc:// zeromq input"; + } + } + else { + stringstream ss; + ss << "Subchannel with uid " << subchanuid << + ": Invalid protocol for " << type << " input (" << + proto << ")" << endl; + throw runtime_error(ss.str()); + } + } + else if (type == "data" and proto == "prbs") { + subchan->input = make_shared(); + subchan->type = subchannel_type_t::DataDmb; + subchan->bitrate = DEFAULT_DATA_BITRATE; + } + else { + stringstream ss; + ss << "Subchannel with uid " << subchanuid << " has unknown type!"; + throw runtime_error(ss.str()); + } + subchan->startAddress = 0; + + if (type == "audio") { + protection->form = UEP; + protection->level = 2; + protection->uep.tableIndex = 0; + } + else { + protection->level = 2; + protection->form = EEP; + protection->eep.profile = EEP_A; + } + + /* Get bitrate */ + try { + subchan->bitrate = pt.get("bitrate"); + if ((subchan->bitrate & 0x7) != 0) { + stringstream ss; + ss << "Subchannel with uid " << subchanuid << + ": Bitrate (" << subchan->bitrate << " not a multiple of 8!"; + throw runtime_error(ss.str()); + } + } + catch (ptree_error &e) { + stringstream ss; + ss << "Error, no bitrate defined for subchannel " << subchanuid; + throw runtime_error(ss.str()); + } + + /* Get id */ + try { + subchan->id = hexparse(pt.get("id")); + } + catch (ptree_error &e) { + for (int i = 0; i < 64; ++i) { // Find first free subchannel + vector::iterator subchannel = getSubchannel(ensemble->subchannels, i); + if (subchannel == ensemble->subchannels.end()) { + subchannel = ensemble->subchannels.end() - 1; + subchan->id = i; + break; + } + } + } + + /* Get optional protection profile */ + string profile = pt.get("protection-profile", ""); + + if (profile == "EEP_A") { + protection->form = EEP; + protection->eep.profile = EEP_A; + } + else if (profile == "EEP_B") { + protection->form = EEP; + protection->eep.profile = EEP_B; + } + else if (profile == "UEP") { + protection->form = UEP; + } + + /* Get protection level */ + try { + int level = pt.get("protection"); + + if (protection->form == UEP) { + if ((level < 1) || (level > 5)) { + stringstream ss; + ss << "Subchannel with uid " << subchanuid << + ": protection level must be between " + "1 to 5 inclusively (current = " << level << " )"; + throw runtime_error(ss.str()); + } + } + else if (protection->form == EEP) { + if ((level < 1) || (level > 4)) { + stringstream ss; + ss << "Subchannel with uid " << subchanuid << + ": protection level must be between " + "1 to 4 inclusively (current = " << level << " )"; + throw runtime_error(ss.str()); + } + } + protection->level = level - 1; + } + catch (ptree_error &e) { + stringstream ss; + ss << "Subchannel with uid " << subchanuid << + ": protection level undefined!"; + throw runtime_error(ss.str()); + } } diff --git a/src/DabMux.cpp b/src/DabMux.cpp index 32ddb39..3927420 100644 --- a/src/DabMux.cpp +++ b/src/DabMux.cpp @@ -94,22 +94,8 @@ typedef DWORD32 uint32_t; # include "Eti.h" #endif -#include "dabInputFile.h" -#include "dabInputFifo.h" -#include "dabInputMpegFile.h" -#include "dabInputMpegFifo.h" -#include "dabInputDabplusFile.h" -#include "dabInputDabplusFifo.h" -#include "dabInputPacketFile.h" -#include "dabInputEnhancedPacketFile.h" -#include "dabInputEnhancedFifo.h" -#include "dabInputUdp.h" -#include "dabInputPrbs.h" -#include "dabInputRawFile.h" -#include "dabInputRawFifo.h" -#include "dabInputDmbFile.h" -#include "dabInputDmbUdp.h" - +#include "input/Prbs.h" +#include "input/Zmq.h" #include "dabOutput/dabOutput.h" #include "dabOutput/edi/TagItems.h" diff --git a/src/DabMux.h b/src/DabMux.h index 5dda759..80b4881 100644 --- a/src/DabMux.h +++ b/src/DabMux.h @@ -25,8 +25,7 @@ You should have received a copy of the GNU General Public License along with ODR-DabMux. If not, see . */ -#ifndef _DABMUX_H -#define _DABMUX_H +#pragma once #include #include @@ -34,7 +33,7 @@ #include "DabMultiplexer.h" #include "RemoteControl.h" #include "dabOutput/dabOutput.h" -#include "dabInput.h" +#include "input/inputs.h" #include "Eti.h" #include "MuxElements.h" @@ -44,5 +43,3 @@ # include #endif -#endif - diff --git a/src/Makefile.am b/src/Makefile.am index 408c86e..b8de4e8 100644 --- a/src/Makefile.am +++ b/src/Makefile.am @@ -47,23 +47,9 @@ odr_dabmux_LDADD =$(ZMQ_LIBS) $(CURL_LIBS) \ odr_dabmux_SOURCES =DabMux.cpp DabMux.h \ DabMultiplexer.cpp DabMultiplexer.h \ - dabInput.h dabInput.cpp \ - dabInputDabplusFifo.h dabInputDabplusFifo.cpp \ - dabInputDabplusFile.h dabInputDabplusFile.cpp \ - dabInputDmbFile.h dabInputDmbFile.cpp \ - dabInputDmbUdp.h dabInputDmbUdp.cpp \ - dabInputEnhancedFifo.h dabInputEnhancedFifo.cpp \ - dabInputEnhancedPacketFile.h dabInputEnhancedPacketFile.cpp \ - dabInputFifo.h dabInputFifo.cpp \ - dabInputFile.h dabInputFile.cpp \ - dabInputMpegFifo.h dabInputMpegFifo.cpp \ - dabInputMpegFile.h dabInputMpegFile.cpp \ - dabInputPacketFile.h dabInputPacketFile.cpp \ - dabInputPrbs.h dabInputPrbs.cpp \ - dabInputRawFile.h dabInputRawFile.cpp \ - dabInputRawFifo.h dabInputRawFifo.cpp \ - dabInputUdp.h dabInputUdp.cpp \ - dabInputZmq.h dabInputZmq.cpp \ + input/inputs.h \ + input/Prbs.cpp input/Prbs.h \ + input/Zmq.cpp input/Zmq.h \ dabOutput/dabOutput.h \ dabOutput/dabOutputFile.cpp \ dabOutput/dabOutputFifo.cpp \ diff --git a/src/MuxElements.h b/src/MuxElements.h index 7056121..7324cdc 100644 --- a/src/MuxElements.h +++ b/src/MuxElements.h @@ -40,7 +40,7 @@ #include #include #include "dabOutput/dabOutput.h" -#include "dabInput.h" +#include "input/inputs.h" #include "RemoteControl.h" #include "Eti.h" @@ -295,7 +295,7 @@ public: std::string uid; std::string inputUri; - DabInputBase* input; + std::shared_ptr input; unsigned char id; subchannel_type_t type; uint16_t startAddress; diff --git a/src/dabInput.h b/src/dabInput.h index d5444cd..0accddb 100644 --- a/src/dabInput.h +++ b/src/dabInput.h @@ -29,8 +29,6 @@ #include "RemoteControl.h" #include -extern Logger etiLog; - // TODO replace usage of dabInputOperations by a // class hierarchy struct dabInputOperations { @@ -48,19 +46,6 @@ struct dabInputOperations { bool operator==(const dabInputOperations&); }; -/* New input object base */ -class DabInputBase { - public: - virtual int open(const std::string& name) = 0; - virtual int readFrame(void* buffer, int size) = 0; - virtual int setBitrate(int bitrate) = 0; - virtual int close() = 0; - - virtual ~DabInputBase() {} - protected: - DabInputBase() {} -}; - /* Wrapper class for old-style dabInputOperations inputs */ class DabInputCompatible : public DabInputBase { public: diff --git a/src/dabInputPrbs.cpp b/src/dabInputPrbs.cpp deleted file mode 100644 index 2678668..0000000 --- a/src/dabInputPrbs.cpp +++ /dev/null @@ -1,98 +0,0 @@ -/* - Copyright (C) 2009 Her Majesty the Queen in Right of Canada (Communications - Research Center Canada) - - Copyright (C) 2016 - Matthias P. Braendli, matthias.braendli@mpb.li - - http://www.opendigitalradio.org - - Pseudo-Random Bit Sequence generator for test purposes. - */ -/* - This file is part of ODR-DabMux. - - ODR-DabMux is free software: you can redistribute it and/or modify - it under the terms of the GNU General Public License as - published by the Free Software Foundation, either version 3 of the - License, or (at your option) any later version. - - ODR-DabMux is distributed in the hope that it will be useful, - but WITHOUT ANY WARRANTY; without even the implied warranty of - MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the - GNU General Public License for more details. - - You should have received a copy of the GNU General Public License - along with ODR-DabMux. If not, see . - */ - -#include "dabInputPrbs.h" - -#include -#include -#include -#include -#include -#include -#include "utils.h" - -using namespace std; - -// ETS 300 799 Clause G.2.1 -// Preferred polynomial is G(x) = x^20 + x^17 + 1 -const uint32_t PRBS_DEFAULT_POLY = (1 << 19) | (1 << 16) | 1; - -int DabInputPrbs::open(const string& name) -{ - if (name.empty()) { - m_prbs.setup(PRBS_DEFAULT_POLY); - } - else { - if (name[0] != ':') { - throw invalid_argument( - "Invalid PRBS address format. " - "Must be prbs://:polynomial."); - } - - const string poly_str = name.substr(1); - - long polynomial = hexparse(poly_str); - - if (polynomial == 0) { - throw invalid_argument("No polynomial given for PRBS input"); - } - - m_prbs.setup(polynomial); - } - rewind(); - - return 0; -} - -int DabInputPrbs::readFrame(void* buffer, int size) -{ - unsigned char* cbuffer = reinterpret_cast(buffer); - - for (int i = 0; i < size; ++i) { - cbuffer[i] = m_prbs.step(); - } - - return size; -} - -int DabInputPrbs::setBitrate(int bitrate) -{ - return bitrate; -} - -int DabInputPrbs::close() -{ - return 0; -} - -int DabInputPrbs::rewind() -{ - m_prbs.rewind(); - return 0; -} - diff --git a/src/dabInputPrbs.h b/src/dabInputPrbs.h deleted file mode 100644 index 95c5e25..0000000 --- a/src/dabInputPrbs.h +++ /dev/null @@ -1,52 +0,0 @@ -/* - Copyright (C) 2009 Her Majesty the Queen in Right of Canada (Communications - Research Center Canada) - - Copyright (C) 2016 - Matthias P. Braendli, matthias.braendli@mpb.li - - http://www.opendigitalradio.org - - Pseudo-Random Bit Sequence generator for test purposes. - */ -/* - This file is part of ODR-DabMux. - - ODR-DabMux is free software: you can redistribute it and/or modify - it under the terms of the GNU General Public License as - published by the Free Software Foundation, either version 3 of the - License, or (at your option) any later version. - - ODR-DabMux is distributed in the hope that it will be useful, - but WITHOUT ANY WARRANTY; without even the implied warranty of - MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the - GNU General Public License for more details. - - You should have received a copy of the GNU General Public License - along with ODR-DabMux. If not, see . - */ - -#pragma once - -#ifdef HAVE_CONFIG_H -# include "config.h" -#endif - -#include - -#include "dabInput.h" -#include "prbs.h" - -class DabInputPrbs : public DabInputBase { - public: - virtual int open(const std::string& name); - virtual int readFrame(void* buffer, int size); - virtual int setBitrate(int bitrate); - virtual int close(); - - private: - virtual int rewind(); - - PrbsGenerator m_prbs; -}; - diff --git a/src/dabInputZmq.cpp b/src/dabInputZmq.cpp deleted file mode 100644 index 93f1ea3..0000000 --- a/src/dabInputZmq.cpp +++ /dev/null @@ -1,619 +0,0 @@ -/* - Copyright (C) 2009 Her Majesty the Queen in Right of Canada (Communications - Research Center Canada) - - Copyright (C) 2013, 2014 Matthias P. Braendli - http://www.opendigitalradio.org - - ZeroMQ input. see www.zeromq.org for more info - - For the AAC+ input, each zeromq message must contain one superframe - or one zmq_frame_header_t followed by a superframe. - - For the MPEG input, each zeromq message must contain one frame. - - Encryption is provided by zmq_curve, see the corresponding manpage. - - From the ZeroMQ manpage 'zmq': - - The 0MQ lightweight messaging kernel is a library which extends the standard - socket interfaces with features traditionally provided by specialised - messaging middleware products. 0MQ sockets provide an abstraction of - asynchronous message queues, multiple messaging patterns, message filtering - (subscriptions), seamless access to multiple transport protocols and more. - */ -/* - This file is part of ODR-DabMux. - - ODR-DabMux is free software: you can redistribute it and/or modify - it under the terms of the GNU General Public License as - published by the Free Software Foundation, either version 3 of the - License, or (at your option) any later version. - - ODR-DabMux is distributed in the hope that it will be useful, - but WITHOUT ANY WARRANTY; without even the implied warranty of - MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the - GNU General Public License for more details. - - You should have received a copy of the GNU General Public License - along with ODR-DabMux. If not, see . - */ - -#include "dabInput.h" -#include "dabInputZmq.h" -#include "PcDebug.h" - -#ifdef HAVE_CONFIG_H -# include "config.h" -#endif - -#ifdef HAVE_INPUT_ZEROMQ - -#include "zmq.hpp" -#include -#include -#include -#include -#include -#include -#include -#include - -#ifdef __MINGW32__ -# define bzero(s, n) memset(s, 0, n) -#endif - -using namespace std; - -int readkey(string& keyfile, char* key) -{ - int fd = open(keyfile.c_str(), O_RDONLY); - if (fd < 0) - return fd; - int ret = read(fd, key, CURVE_KEYLEN); - close(fd); - if (ret < 0) { - return ret; - } - - /* It needs to be zero-terminated */ - key[CURVE_KEYLEN] = '\0'; - - return 0; -} - -/***** Common functions (MPEG and AAC) ******/ - -/* If necessary, unbind the socket, then check the keys, - * if they are ok and encryption is required, set the - * keys to the socket, and finally bind the socket - * to the new address - */ -void DabInputZmqBase::rebind() -{ - if (! m_zmq_sock_bound_to.empty()) { - try { - m_zmq_sock.unbind(m_zmq_sock_bound_to.c_str()); - } - catch (zmq::error_t& err) { - etiLog.level(warn) << "ZMQ unbind for input " << m_name << " failed"; - } - } - - m_zmq_sock_bound_to = ""; - - /* Load each key independently */ - if (! m_config.curve_public_keyfile.empty()) { - int rc = readkey(m_config.curve_public_keyfile, m_curve_public_key); - - if (rc < 0) { - etiLog.level(warn) << "Invalid public key for input " << - m_name; - - INVALIDATE_KEY(m_curve_public_key); - } - } - - if (! m_config.curve_secret_keyfile.empty()) { - int rc = readkey(m_config.curve_secret_keyfile, m_curve_secret_key); - - if (rc < 0) { - etiLog.level(warn) << "Invalid secret key for input " << - m_name; - - INVALIDATE_KEY(m_curve_secret_key); - } - } - - if (! m_config.curve_encoder_keyfile.empty()) { - int rc = readkey(m_config.curve_encoder_keyfile, m_curve_encoder_key); - - if (rc < 0) { - etiLog.level(warn) << "Invalid encoder key for input " << - m_name; - - INVALIDATE_KEY(m_curve_encoder_key); - } - } - - /* If you want encryption, you need to have defined all - * key files - */ - if ( m_config.enable_encryption && - ( ! (KEY_VALID(m_curve_public_key) && - KEY_VALID(m_curve_secret_key) && - KEY_VALID(m_curve_encoder_key) ) ) ) { - throw std::runtime_error("When enabling encryption, all three " - "keyfiles must be valid!"); - } - - if (m_config.enable_encryption) { - try { - /* We want to check that the encoder is the right one, - * so the encoder is the CURVE server. - */ - m_zmq_sock.setsockopt(ZMQ_CURVE_SERVERKEY, - m_curve_encoder_key, CURVE_KEYLEN); - } - catch (zmq::error_t& err) { - std::ostringstream os; - os << "ZMQ set encoder key for input " << m_name << " failed" << - err.what(); - throw std::runtime_error(os.str()); - } - - try { - m_zmq_sock.setsockopt(ZMQ_CURVE_PUBLICKEY, - m_curve_public_key, CURVE_KEYLEN); - } - catch (zmq::error_t& err) { - std::ostringstream os; - os << "ZMQ set public key for input " << m_name << " failed" << - err.what(); - throw std::runtime_error(os.str()); - } - - try { - m_zmq_sock.setsockopt(ZMQ_CURVE_SECRETKEY, - m_curve_secret_key, CURVE_KEYLEN); - } - catch (zmq::error_t& err) { - std::ostringstream os; - os << "ZMQ set secret key for input " << m_name << " failed" << - err.what(); - throw std::runtime_error(os.str()); - } - } - else { - try { - /* This forces the socket to go to the ZMQ_NULL auth - * mechanism - */ - const int no = 0; - m_zmq_sock.setsockopt(ZMQ_CURVE_SERVER, &no, sizeof(no)); - } - catch (zmq::error_t& err) { - etiLog.level(warn) << "ZMQ disable encryption keys for input " << - m_name << " failed: " << err.what(); - } - - } - - // Prepare the ZMQ socket to accept connections - try { - m_zmq_sock.bind(m_inputUri.c_str()); - } - catch (zmq::error_t& err) { - std::ostringstream os; - os << "ZMQ bind for input " << m_name << " failed" << - err.what(); - throw std::runtime_error(os.str()); - } - - m_zmq_sock_bound_to = m_inputUri; - - try { - m_zmq_sock.setsockopt(ZMQ_SUBSCRIBE, nullptr, 0); - } - catch (zmq::error_t& err) { - std::ostringstream os; - os << "ZMQ set socket options for input " << m_name << " failed" << - err.what(); - throw std::runtime_error(os.str()); - } -} - -int DabInputZmqBase::open(const std::string& inputUri) -{ - m_inputUri = inputUri; - - /* Let caller handle exceptions when we open() */ - rebind(); - - // We want to appear in the statistics ! - m_stats.registerAtServer(); - - return 0; -} - -int DabInputZmqBase::close() -{ - m_zmq_sock.close(); - return 0; -} - -int DabInputZmqBase::setBitrate(int bitrate) -{ - m_bitrate = bitrate; - return bitrate; // TODO do a nice check here -} - -// size corresponds to a frame size. It is constant for a given bitrate -int DabInputZmqBase::readFrame(void* buffer, int size) -{ - int rc; - - /* We must *always* read data from the ZMQ socket, - * to make sure that ZMQ internal buffers are emptied - * quickly. It's the only way to control the buffers - * of the whole path from encoder to our frame_buffer. - */ - rc = readFromSocket(size); - - /* Notify of a buffer overrun, and drop some frames */ - if (m_frame_buffer.size() >= m_config.buffer_size) { - m_stats.notifyOverrun(); - - /* If the buffer is really too full, we drop as many frames as needed - * to get down to the prebuffering size. We would like to have our buffer - * filled to the prebuffering length. - */ - if (m_frame_buffer.size() >= 1.5*m_config.buffer_size) { - size_t over_max = m_frame_buffer.size() - m_config.prebuffering; - - while (over_max--) { - delete[] m_frame_buffer.front(); - m_frame_buffer.pop_front(); - } - } - else { - /* Our frame_buffer contains DAB logical frames. Five of these make one - * AAC superframe. - * - * Dropping this superframe amounts to dropping 120ms of audio. - * - * We're actually not sure to drop five DAB logical frames - * belonging to the same AAC superframe. It is assumed that no - * receiver will crash because of this. At least, the DAB logical frame - * vs. AAC superframe alignment is preserved. - * - * TODO: of course this assumption probably doesn't hold. Fix this ! - * TODO: also, with MPEG, the above doesn't hold, so we drop five - * frames even though we could drop less. - * */ - for (int frame_del_count = 0; frame_del_count < 5; frame_del_count++) { - delete[] m_frame_buffer.front(); - m_frame_buffer.pop_front(); - } - } - } - - if (m_prebuf_current > 0) { - if (rc > 0) - m_prebuf_current--; - if (m_prebuf_current == 0) - etiLog.log(info, "inputZMQ %s input pre-buffering complete\n", - m_name.c_str()); - - /* During prebuffering, give a zeroed frame to the mux */ - m_stats.notifyUnderrun(); - memset(buffer, 0, size); - return size; - } - - // Save stats data in bytes, not in frames - m_stats.notifyBuffer(m_frame_buffer.size() * size); - - if (m_frame_buffer.empty()) { - etiLog.log(warn, "inputZMQ %s input empty, re-enabling pre-buffering\n", - m_name.c_str()); - // reset prebuffering - m_prebuf_current = m_config.prebuffering; - - /* We have no data to give, we give a zeroed frame */ - m_stats.notifyUnderrun(); - memset(buffer, 0, size); - return size; - } - else - { - /* Normal situation, give a frame from the frame_buffer */ - uint8_t* newframe = m_frame_buffer.front(); - memcpy(buffer, newframe, size); - delete[] newframe; - m_frame_buffer.pop_front(); - return size; - } -} - - -/******** MPEG input *******/ - -// Read a MPEG frame from the socket, and push to list -int DabInputZmqMPEG::readFromSocket(size_t framesize) -{ - bool messageReceived = false; - zmq::message_t msg; - - try { - messageReceived = m_zmq_sock.recv(&msg, ZMQ_DONTWAIT); - if (!messageReceived) { - return 0; - } - - } - catch (zmq::error_t& err) - { - etiLog.level(error) << "Failed to receive MPEG frame from zmq socket " << - m_name << ": " << err.what(); - } - - /* This is the old 'one superframe per ZMQ message' format */ - uint8_t* data = (uint8_t*)msg.data(); - size_t datalen = msg.size(); - - /* Look for the new zmq_frame_header_t format */ - zmq_frame_header_t* frame = (zmq_frame_header_t*)msg.data(); - - if (msg.size() == ZMQ_FRAME_SIZE(frame) && - frame->version == 1 && - frame->encoder == ZMQ_ENCODER_TOOLAME) { - datalen = frame->datasize; - data = ZMQ_FRAME_DATA(frame); - - m_stats.notifyPeakLevels(frame->audiolevel_left, - frame->audiolevel_right); - } - - - if (datalen == framesize) - { - if (m_frame_buffer.size() > m_config.buffer_size) { - etiLog.level(warn) << - "inputZMQ " << m_name << - " buffer full (" << m_frame_buffer.size() << ")," - " dropping incoming frame !"; - messageReceived = false; - } - else if (m_enable_input) { - // copy the input frame blockwise into the frame_buffer - auto framedata = new uint8_t[framesize]; - memcpy(framedata, data, framesize); - m_frame_buffer.push_back(framedata); - } - else { - return 0; - } - } - else { - etiLog.level(error) << - "inputZMQ " << m_name << - " verify bitrate: recv'd " << msg.size() << " B" << - ", need " << framesize << "."; - messageReceived = false; - } - - return messageReceived ? msg.size() : 0; -} - -/******** AAC+ input *******/ - -// Read a AAC+ superframe from the socket, cut it into five frames, -// and push to list -int DabInputZmqAAC::readFromSocket(size_t framesize) -{ - bool messageReceived; - zmq::message_t msg; - - try { - messageReceived = m_zmq_sock.recv(&msg, ZMQ_DONTWAIT); - if (!messageReceived) { - return 0; - } - - } - catch (zmq::error_t& err) - { - etiLog.level(error) << - "Failed to receive AAC superframe from zmq socket " << - m_name << ": " << err.what(); - } - - /* This is the old 'one superframe per ZMQ message' format */ - uint8_t* data = (uint8_t*)msg.data(); - size_t datalen = msg.size(); - - /* Look for the new zmq_frame_header_t format */ - zmq_frame_header_t* frame = (zmq_frame_header_t*)msg.data(); - - if (msg.size() == ZMQ_FRAME_SIZE(frame) && - frame->version == 1 && - frame->encoder == ZMQ_ENCODER_FDK) { - datalen = frame->datasize; - data = ZMQ_FRAME_DATA(frame); - - m_stats.notifyPeakLevels(frame->audiolevel_left, - frame->audiolevel_right); - } - - - /* TS 102 563, Section 6: - * Audio super frames are transported in five successive DAB logical frames - * with additional error protection. - */ - if (datalen) - { - if (datalen == 5*framesize) - { - if (m_frame_buffer.size() > m_config.buffer_size) { - etiLog.level(warn) << - "inputZMQ " << m_name << - " buffer full (" << m_frame_buffer.size() << ")," - " dropping incoming superframe !"; - datalen = 0; - } - else if (m_enable_input) { - // copy the input frame blockwise into the frame_buffer - for (uint8_t* framestart = data; - framestart < &data[5*framesize]; - framestart += framesize) { - auto audioframe = new uint8_t[framesize]; - memcpy(audioframe, framestart, framesize); - m_frame_buffer.push_back(audioframe); - } - } - else { - datalen = 0; - } - } - else { - etiLog.level(error) << - "inputZMQ " << m_name << - " verify bitrate: recv'd " << msg.size() << " B" << - ", need " << 5*framesize << "."; - - datalen = 0; - } - } - else { - etiLog.level(error) << - "inputZMQ " << m_name << - " invalid frame received"; - } - - return datalen; -} - -/********* REMOTE CONTROL ***********/ - -void DabInputZmqBase::set_parameter(const string& parameter, - const string& value) -{ - if (parameter == "buffer") { - size_t new_limit = atol(value.c_str()); - - if (new_limit < INPUT_ZMQ_MIN_BUFFER_SIZE) { - throw ParameterError("Desired buffer size too small." - " Minimum " STRINGIFY(INPUT_ZMQ_MIN_BUFFER_SIZE) ); - } - else if (new_limit > INPUT_ZMQ_MAX_BUFFER_SIZE) { - throw ParameterError("Desired buffer size too large." - " Maximum " STRINGIFY(INPUT_ZMQ_MAX_BUFFER_SIZE) ); - } - - m_config.buffer_size = new_limit; - } - else if (parameter == "prebuffering") { - size_t new_prebuf = atol(value.c_str()); - - if (new_prebuf < INPUT_ZMQ_MIN_BUFFER_SIZE) { - throw ParameterError("Desired prebuffering too small." - " Minimum " STRINGIFY(INPUT_ZMQ_MIN_BUFFER_SIZE) ); - } - else if (new_prebuf > INPUT_ZMQ_MAX_BUFFER_SIZE) { - throw ParameterError("Desired prebuffering too large." - " Maximum " STRINGIFY(INPUT_ZMQ_MAX_BUFFER_SIZE) ); - } - - m_config.prebuffering = new_prebuf; - } - else if (parameter == "enable") { - if (value == "1") { - m_enable_input = true; - } - else if (value == "0") { - m_enable_input = false; - } - else { - throw ParameterError("Value not understood, specify 0 or 1."); - } - } - else if (parameter == "encryption") { - if (value == "1") { - m_config.enable_encryption = true; - } - else if (value == "0") { - m_config.enable_encryption = false; - } - else { - throw ParameterError("Value not understood, specify 0 or 1."); - } - - try { - rebind(); - } - catch (std::runtime_error &e) { - stringstream ss; - ss << "Could not bind socket again with new keys." << - e.what(); - throw ParameterError(ss.str()); - } - } - else if (parameter == "secretkey") { - m_config.curve_secret_keyfile = value; - } - else if (parameter == "publickey") { - m_config.curve_public_keyfile = value; - } - else if (parameter == "encoderkey") { - m_config.curve_encoder_keyfile = value; - } - else { - stringstream ss; - ss << "Parameter '" << parameter << - "' is not exported by controllable " << get_rc_name(); - throw ParameterError(ss.str()); - } -} - -const string DabInputZmqBase::get_parameter(const string& parameter) const -{ - stringstream ss; - if (parameter == "buffer") { - ss << m_config.buffer_size; - } - else if (parameter == "prebuffering") { - ss << m_config.prebuffering; - } - else if (parameter == "enable") { - if (m_enable_input) - ss << "true"; - else - ss << "false"; - } - else if (parameter == "encryption") { - if (m_config.enable_encryption) - ss << "true"; - else - ss << "false"; - } - else if (parameter == "secretkey") { - ss << m_config.curve_secret_keyfile; - } - else if (parameter == "publickey") { - ss << m_config.curve_public_keyfile; - } - else if (parameter == "encoderkey") { - ss << m_config.curve_encoder_keyfile; - } - else { - ss << "Parameter '" << parameter << - "' is not exported by controllable " << get_rc_name(); - throw ParameterError(ss.str()); - } - return ss.str(); - -} - -#endif - diff --git a/src/dabInputZmq.h b/src/dabInputZmq.h deleted file mode 100644 index 351fb07..0000000 --- a/src/dabInputZmq.h +++ /dev/null @@ -1,276 +0,0 @@ -/* - Copyright (C) 2009 Her Majesty the Queen in Right of Canada (Communications - Research Center Canada) - - Copyright (C) 2013, 2014 Matthias P. Braendli - http://www.opendigitalradio.org - - ZeroMQ input. see www.zeromq.org for more info - - For the AAC+ input, each zeromq message must contain one superframe, - or one zmq_frame_header_t followed by a superframe. - - For the MPEG input, each zeromq message must contain one frame. - - Encryption is provided by zmq_curve, see the corresponding manpage. - - From the ZeroMQ manpage 'zmq': - - The 0MQ lightweight messaging kernel is a library which extends the standard - socket interfaces with features traditionally provided by specialised - messaging middleware products. 0MQ sockets provide an abstraction of - asynchronous message queues, multiple messaging patterns, message filtering - (subscriptions), seamless access to multiple transport protocols and more. - */ -/* - This file is part of ODR-DabMux. - - It defines a ZeroMQ input for dabplus data. - - ODR-DabMux is free software: you can redistribute it and/or modify - it under the terms of the GNU General Public License as - published by the Free Software Foundation, either version 3 of the - License, or (at your option) any later version. - - ODR-DabMux is distributed in the hope that it will be useful, - but WITHOUT ANY WARRANTY; without even the implied warranty of - MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the - GNU General Public License for more details. - - You should have received a copy of the GNU General Public License - along with ODR-DabMux. If not, see . - */ - -#ifndef DAB_INPUT_ZMQ_H -#define DAB_INPUT_ZMQ_H - -#ifdef HAVE_CONFIG_H -# include "config.h" -#endif - -#ifdef HAVE_INPUT_ZEROMQ - -#include -#include -#include -#include "zmq.hpp" -#include "dabInput.h" -#include "ManagementServer.h" - -/* The frame_buffer contains DAB logical frames as defined in - * TS 102 563, section 6. - * Five elements of this buffer make one AAC superframe (120ms audio) - */ - -// Number of elements to prebuffer before starting the pipeline -#define INPUT_ZMQ_DEF_PREBUFFERING (5*4) // 480ms - -// Default frame_buffer size in number of elements -#define INPUT_ZMQ_DEF_BUFFER_SIZE (5*8) // 960ms - -// Minimum frame_buffer size in number of elements -// This is one AAC superframe, but you probably don't want to -// go that low anyway. -#define INPUT_ZMQ_MIN_BUFFER_SIZE (5*1) // 120ms - -// Maximum frame_buffer size in number of elements -// One minute is clearly way over what everybody would -// want. -#define INPUT_ZMQ_MAX_BUFFER_SIZE (5*500) // 60s - -/* The ZeroMQ Curve key is 40 bytes long in Z85 representation - * - * But we need to store it as zero-terminated string. - */ -#define CURVE_KEYLEN 40 - -/* helper to invalidate a key */ -#define INVALIDATE_KEY(k) memset(k, 0, CURVE_KEYLEN+1) - -/* Verification for key validity */ -#define KEY_VALID(k) (k[0] != '\0') - -/* Read a key from file into key - * - * Returns 0 on success, negative value on failure - */ -int readkey(std::string& keyfile, char* key); - -struct dab_input_zmq_config_t -{ - /* The size of the internal buffer, measured in number - * of elements. - * - * Each element corresponds to five frames, - * or one AAC superframe. - */ - size_t buffer_size; - - /* The amount of prebuffering to do before we start streaming - * - * Same units as buffer_size - */ - size_t prebuffering; - - /* Whether to enforce encryption or not - */ - bool enable_encryption; - - /* Full path to file containing public key. - */ - std::string curve_public_keyfile; - - /* Full path to file containing secret key. - */ - std::string curve_secret_keyfile; - - /* Full path to file containing encoder public key. - */ - std::string curve_encoder_keyfile; -}; - -#define ZMQ_ENCODER_FDK 1 -#define ZMQ_ENCODER_TOOLAME 2 - -/* This defines the on-wire representation of a ZMQ message header. - * - * The data follows right after this header */ -struct zmq_frame_header_t -{ - uint16_t version; // we support version=1 now - uint16_t encoder; // see ZMQ_ENCODER_XYZ - - /* length of the 'data' field */ - uint32_t datasize; - - /* Audio level, peak, linear PCM */ - int16_t audiolevel_left; - int16_t audiolevel_right; - - /* Data follows this header */ -} __attribute__ ((packed)); - -/* The expected frame size incl data of the given frame */ -#define ZMQ_FRAME_SIZE(f) (sizeof(zmq_frame_header_t) + f->datasize) - -#define ZMQ_FRAME_DATA(f) ( ((uint8_t*)f)+sizeof(zmq_frame_header_t) ) - - -class DabInputZmqBase : public DabInputBase, public RemoteControllable { - public: - DabInputZmqBase(const std::string name, - dab_input_zmq_config_t config) - : RemoteControllable(name), - m_zmq_context(1), - m_zmq_sock(m_zmq_context, ZMQ_SUB), - m_zmq_sock_bound_to(""), - m_bitrate(0), - m_enable_input(true), - m_config(config), - m_stats(m_name), - m_prebuf_current(config.prebuffering) { - RC_ADD_PARAMETER(enable, - "If the input is enabled. Set to zero to empty the buffer."); - - RC_ADD_PARAMETER(encryption, - "If encryption is enabled or disabled [1 or 0]." - " If 1 is written, the keys are reloaded."); - - RC_ADD_PARAMETER(publickey, - "The multiplexer's public key file."); - - RC_ADD_PARAMETER(secretkey, - "The multiplexer's secret key file."); - - RC_ADD_PARAMETER(encoderkey, - "The encoder's public key file."); - - /* Set all keys to zero */ - INVALIDATE_KEY(m_curve_public_key); - INVALIDATE_KEY(m_curve_secret_key); - INVALIDATE_KEY(m_curve_encoder_key); - } - - virtual int open(const std::string& inputUri); - virtual int readFrame(void* buffer, int size); - virtual int setBitrate(int bitrate); - virtual int close(); - - /* Remote control */ - virtual void set_parameter(const std::string& parameter, - const std::string& value); - - /* Getting a parameter always returns a string. */ - virtual const std::string get_parameter(const std::string& parameter) const; - - protected: - virtual int readFromSocket(size_t framesize) = 0; - - virtual void rebind(); - - zmq::context_t m_zmq_context; - zmq::socket_t m_zmq_sock; // handle for the zmq socket - - /* If the socket is bound, this saves the endpoint, - * otherwise, it's an empty string - */ - std::string m_zmq_sock_bound_to; - int m_bitrate; - - /* set this to zero to empty the input buffer */ - bool m_enable_input; - - /* stores elements of type char[] */ - std::list m_frame_buffer; - - dab_input_zmq_config_t m_config; - - /* Key management, keys need to be zero-terminated */ - char m_curve_public_key[CURVE_KEYLEN+1]; - char m_curve_secret_key[CURVE_KEYLEN+1]; - char m_curve_encoder_key[CURVE_KEYLEN+1]; - - std::string m_inputUri; - - InputStat m_stats; - - private: - size_t m_prebuf_current; -}; - -class DabInputZmqMPEG : public DabInputZmqBase { - public: - DabInputZmqMPEG(const std::string name, - dab_input_zmq_config_t config) - : DabInputZmqBase(name, config) { - RC_ADD_PARAMETER(buffer, - "Size of the input buffer [mpeg frames]"); - - RC_ADD_PARAMETER(prebuffering, - "Min buffer level before streaming starts [mpeg frames]"); - } - - private: - virtual int readFromSocket(size_t framesize); -}; - -class DabInputZmqAAC : public DabInputZmqBase { - public: - DabInputZmqAAC(const std::string name, - dab_input_zmq_config_t config) - : DabInputZmqBase(name, config) { - RC_ADD_PARAMETER(buffer, - "Size of the input buffer [aac superframes]"); - - RC_ADD_PARAMETER(prebuffering, - "Min buffer level before streaming starts [aac superframes]"); - } - - private: - virtual int readFromSocket(size_t framesize); -}; - -#endif // HAVE_INPUT_ZMQ - -#endif // DAB_INPUT_ZMQ_H - diff --git a/src/input/Prbs.cpp b/src/input/Prbs.cpp new file mode 100644 index 0000000..b9e244b --- /dev/null +++ b/src/input/Prbs.cpp @@ -0,0 +1,101 @@ +/* + Copyright (C) 2009 Her Majesty the Queen in Right of Canada (Communications + Research Center Canada) + + Copyright (C) 2016 + Matthias P. Braendli, matthias.braendli@mpb.li + + http://www.opendigitalradio.org + + Pseudo-Random Bit Sequence generator for test purposes. + */ +/* + This file is part of ODR-DabMux. + + ODR-DabMux is free software: you can redistribute it and/or modify + it under the terms of the GNU General Public License as + published by the Free Software Foundation, either version 3 of the + License, or (at your option) any later version. + + ODR-DabMux is distributed in the hope that it will be useful, + but WITHOUT ANY WARRANTY; without even the implied warranty of + MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + GNU General Public License for more details. + + You should have received a copy of the GNU General Public License + along with ODR-DabMux. If not, see . + */ + +#include "input/Prbs.h" + +#include +#include +#include +#include +#include +#include +#include "utils.h" + +using namespace std; + +namespace Inputs { + +// ETS 300 799 Clause G.2.1 +// Preferred polynomial is G(x) = x^20 + x^17 + 1 +const uint32_t PRBS_DEFAULT_POLY = (1 << 19) | (1 << 16) | 1; + +int Prbs::open(const string& name) +{ + if (name.empty()) { + m_prbs.setup(PRBS_DEFAULT_POLY); + } + else { + if (name[0] != ':') { + throw invalid_argument( + "Invalid PRBS address format. " + "Must be prbs://:polynomial."); + } + + const string poly_str = name.substr(1); + + long polynomial = hexparse(poly_str); + + if (polynomial == 0) { + throw invalid_argument("No polynomial given for PRBS input"); + } + + m_prbs.setup(polynomial); + } + rewind(); + + return 0; +} + +int Prbs::readFrame(void* buffer, int size) +{ + unsigned char* cbuffer = reinterpret_cast(buffer); + + for (int i = 0; i < size; ++i) { + cbuffer[i] = m_prbs.step(); + } + + return size; +} + +int Prbs::setBitrate(int bitrate) +{ + return bitrate; +} + +int Prbs::close() +{ + return 0; +} + +int Prbs::rewind() +{ + m_prbs.rewind(); + return 0; +} + +}; diff --git a/src/input/Prbs.h b/src/input/Prbs.h new file mode 100644 index 0000000..47b52ad --- /dev/null +++ b/src/input/Prbs.h @@ -0,0 +1,56 @@ +/* + Copyright (C) 2009 Her Majesty the Queen in Right of Canada (Communications + Research Center Canada) + + Copyright (C) 2016 + Matthias P. Braendli, matthias.braendli@mpb.li + + http://www.opendigitalradio.org + + Pseudo-Random Bit Sequence generator for test purposes. + */ +/* + This file is part of ODR-DabMux. + + ODR-DabMux is free software: you can redistribute it and/or modify + it under the terms of the GNU General Public License as + published by the Free Software Foundation, either version 3 of the + License, or (at your option) any later version. + + ODR-DabMux is distributed in the hope that it will be useful, + but WITHOUT ANY WARRANTY; without even the implied warranty of + MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + GNU General Public License for more details. + + You should have received a copy of the GNU General Public License + along with ODR-DabMux. If not, see . + */ + +#pragma once + +#ifdef HAVE_CONFIG_H +# include "config.h" +#endif + +#include + +#include "input/inputs.h" +#include "prbs.h" + +namespace Inputs { + +class Prbs : public InputBase { + public: + virtual int open(const std::string& name); + virtual int readFrame(void* buffer, int size); + virtual int setBitrate(int bitrate); + virtual int close(); + + private: + virtual int rewind(); + + PrbsGenerator m_prbs; +}; + +}; + diff --git a/src/input/Zmq.cpp b/src/input/Zmq.cpp new file mode 100644 index 0000000..6ef5fce --- /dev/null +++ b/src/input/Zmq.cpp @@ -0,0 +1,625 @@ +/* + Copyright (C) 2009 Her Majesty the Queen in Right of Canada (Communications + Research Center Canada) + + Copyright (C) 2016 Matthias P. Braendli + http://www.opendigitalradio.org + + ZeroMQ input. see www.zeromq.org for more info + + For the AAC+ input, each zeromq message must contain one superframe + or one zmq_frame_header_t followed by a superframe. + + For the MPEG input, each zeromq message must contain one frame. + + Encryption is provided by zmq_curve, see the corresponding manpage. + + From the ZeroMQ manpage 'zmq': + + The 0MQ lightweight messaging kernel is a library which extends the standard + socket interfaces with features traditionally provided by specialised + messaging middleware products. 0MQ sockets provide an abstraction of + asynchronous message queues, multiple messaging patterns, message filtering + (subscriptions), seamless access to multiple transport protocols and more. + */ +/* + This file is part of ODR-DabMux. + + ODR-DabMux is free software: you can redistribute it and/or modify + it under the terms of the GNU General Public License as + published by the Free Software Foundation, either version 3 of the + License, or (at your option) any later version. + + ODR-DabMux is distributed in the hope that it will be useful, + but WITHOUT ANY WARRANTY; without even the implied warranty of + MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + GNU General Public License for more details. + + You should have received a copy of the GNU General Public License + along with ODR-DabMux. If not, see . + */ + +#include "input/Zmq.h" + +#ifdef HAVE_CONFIG_H +# include "config.h" +#endif + +#ifdef HAVE_INPUT_ZEROMQ + +#include "zmq.hpp" +#include +#include +#include +#include +#include +#include +#include +#include +#include "PcDebug.h" +#include "Log.h" + +#ifdef __MINGW32__ +# define bzero(s, n) memset(s, 0, n) +#endif + +namespace Inputs { + +using namespace std; + +int readkey(string& keyfile, char* key) +{ + FILE* fd = fopen(keyfile.c_str(), "r"); + if (fd == nullptr) { + return -1; + } + + int ret = fread(key, CURVE_KEYLEN, 1, fd); + fclose(fd); + if (ret == 0) { + return -1; + } + + /* It needs to be zero-terminated */ + key[CURVE_KEYLEN] = '\0'; + + return 0; +} + +/***** Common functions (MPEG and AAC) ******/ + +/* If necessary, unbind the socket, then check the keys, + * if they are ok and encryption is required, set the + * keys to the socket, and finally bind the socket + * to the new address + */ +void ZmqBase::rebind() +{ + if (! m_zmq_sock_bound_to.empty()) { + try { + m_zmq_sock.unbind(m_zmq_sock_bound_to.c_str()); + } + catch (zmq::error_t& err) { + etiLog.level(warn) << "ZMQ unbind for input " << m_name << " failed"; + } + } + + m_zmq_sock_bound_to = ""; + + /* Load each key independently */ + if (! m_config.curve_public_keyfile.empty()) { + int rc = readkey(m_config.curve_public_keyfile, m_curve_public_key); + + if (rc < 0) { + etiLog.level(warn) << "Invalid public key for input " << + m_name; + + INVALIDATE_KEY(m_curve_public_key); + } + } + + if (! m_config.curve_secret_keyfile.empty()) { + int rc = readkey(m_config.curve_secret_keyfile, m_curve_secret_key); + + if (rc < 0) { + etiLog.level(warn) << "Invalid secret key for input " << + m_name; + + INVALIDATE_KEY(m_curve_secret_key); + } + } + + if (! m_config.curve_encoder_keyfile.empty()) { + int rc = readkey(m_config.curve_encoder_keyfile, m_curve_encoder_key); + + if (rc < 0) { + etiLog.level(warn) << "Invalid encoder key for input " << + m_name; + + INVALIDATE_KEY(m_curve_encoder_key); + } + } + + /* If you want encryption, you need to have defined all + * key files + */ + if ( m_config.enable_encryption && + ( ! (KEY_VALID(m_curve_public_key) && + KEY_VALID(m_curve_secret_key) && + KEY_VALID(m_curve_encoder_key) ) ) ) { + throw std::runtime_error("When enabling encryption, all three " + "keyfiles must be valid!"); + } + + if (m_config.enable_encryption) { + try { + /* We want to check that the encoder is the right one, + * so the encoder is the CURVE server. + */ + m_zmq_sock.setsockopt(ZMQ_CURVE_SERVERKEY, + m_curve_encoder_key, CURVE_KEYLEN); + } + catch (zmq::error_t& err) { + std::ostringstream os; + os << "ZMQ set encoder key for input " << m_name << " failed" << + err.what(); + throw std::runtime_error(os.str()); + } + + try { + m_zmq_sock.setsockopt(ZMQ_CURVE_PUBLICKEY, + m_curve_public_key, CURVE_KEYLEN); + } + catch (zmq::error_t& err) { + std::ostringstream os; + os << "ZMQ set public key for input " << m_name << " failed" << + err.what(); + throw std::runtime_error(os.str()); + } + + try { + m_zmq_sock.setsockopt(ZMQ_CURVE_SECRETKEY, + m_curve_secret_key, CURVE_KEYLEN); + } + catch (zmq::error_t& err) { + std::ostringstream os; + os << "ZMQ set secret key for input " << m_name << " failed" << + err.what(); + throw std::runtime_error(os.str()); + } + } + else { + try { + /* This forces the socket to go to the ZMQ_NULL auth + * mechanism + */ + const int no = 0; + m_zmq_sock.setsockopt(ZMQ_CURVE_SERVER, &no, sizeof(no)); + } + catch (zmq::error_t& err) { + etiLog.level(warn) << "ZMQ disable encryption keys for input " << + m_name << " failed: " << err.what(); + } + + } + + // Prepare the ZMQ socket to accept connections + try { + m_zmq_sock.bind(m_inputUri.c_str()); + } + catch (zmq::error_t& err) { + std::ostringstream os; + os << "ZMQ bind for input " << m_name << " failed" << + err.what(); + throw std::runtime_error(os.str()); + } + + m_zmq_sock_bound_to = m_inputUri; + + try { + m_zmq_sock.setsockopt(ZMQ_SUBSCRIBE, nullptr, 0); + } + catch (zmq::error_t& err) { + std::ostringstream os; + os << "ZMQ set socket options for input " << m_name << " failed" << + err.what(); + throw std::runtime_error(os.str()); + } +} + +int ZmqBase::open(const std::string& inputUri) +{ + m_inputUri = inputUri; + + /* Let caller handle exceptions when we open() */ + rebind(); + + // We want to appear in the statistics ! + m_stats.registerAtServer(); + + return 0; +} + +int ZmqBase::close() +{ + m_zmq_sock.close(); + return 0; +} + +int ZmqBase::setBitrate(int bitrate) +{ + m_bitrate = bitrate; + return bitrate; // TODO do a nice check here +} + +// size corresponds to a frame size. It is constant for a given bitrate +int ZmqBase::readFrame(void* buffer, int size) +{ + int rc; + + /* We must *always* read data from the ZMQ socket, + * to make sure that ZMQ internal buffers are emptied + * quickly. It's the only way to control the buffers + * of the whole path from encoder to our frame_buffer. + */ + rc = readFromSocket(size); + + /* Notify of a buffer overrun, and drop some frames */ + if (m_frame_buffer.size() >= m_config.buffer_size) { + m_stats.notifyOverrun(); + + /* If the buffer is really too full, we drop as many frames as needed + * to get down to the prebuffering size. We would like to have our buffer + * filled to the prebuffering length. + */ + if (m_frame_buffer.size() >= 1.5*m_config.buffer_size) { + size_t over_max = m_frame_buffer.size() - m_config.prebuffering; + + while (over_max--) { + delete[] m_frame_buffer.front(); + m_frame_buffer.pop_front(); + } + } + else { + /* Our frame_buffer contains DAB logical frames. Five of these make one + * AAC superframe. + * + * Dropping this superframe amounts to dropping 120ms of audio. + * + * We're actually not sure to drop five DAB logical frames + * belonging to the same AAC superframe. It is assumed that no + * receiver will crash because of this. At least, the DAB logical frame + * vs. AAC superframe alignment is preserved. + * + * TODO: of course this assumption probably doesn't hold. Fix this ! + * TODO: also, with MPEG, the above doesn't hold, so we drop five + * frames even though we could drop less. + * */ + for (int frame_del_count = 0; frame_del_count < 5; frame_del_count++) { + delete[] m_frame_buffer.front(); + m_frame_buffer.pop_front(); + } + } + } + + if (m_prebuf_current > 0) { + if (rc > 0) + m_prebuf_current--; + if (m_prebuf_current == 0) + etiLog.log(info, "inputZMQ %s input pre-buffering complete\n", + m_name.c_str()); + + /* During prebuffering, give a zeroed frame to the mux */ + m_stats.notifyUnderrun(); + memset(buffer, 0, size); + return size; + } + + // Save stats data in bytes, not in frames + m_stats.notifyBuffer(m_frame_buffer.size() * size); + + if (m_frame_buffer.empty()) { + etiLog.log(warn, "inputZMQ %s input empty, re-enabling pre-buffering\n", + m_name.c_str()); + // reset prebuffering + m_prebuf_current = m_config.prebuffering; + + /* We have no data to give, we give a zeroed frame */ + m_stats.notifyUnderrun(); + memset(buffer, 0, size); + return size; + } + else + { + /* Normal situation, give a frame from the frame_buffer */ + uint8_t* newframe = m_frame_buffer.front(); + memcpy(buffer, newframe, size); + delete[] newframe; + m_frame_buffer.pop_front(); + return size; + } +} + + +/******** MPEG input *******/ + +// Read a MPEG frame from the socket, and push to list +int ZmqMPEG::readFromSocket(size_t framesize) +{ + bool messageReceived = false; + zmq::message_t msg; + + try { + messageReceived = m_zmq_sock.recv(&msg, ZMQ_DONTWAIT); + if (!messageReceived) { + return 0; + } + + } + catch (zmq::error_t& err) + { + etiLog.level(error) << "Failed to receive MPEG frame from zmq socket " << + m_name << ": " << err.what(); + } + + /* This is the old 'one superframe per ZMQ message' format */ + uint8_t* data = (uint8_t*)msg.data(); + size_t datalen = msg.size(); + + /* Look for the new zmq_frame_header_t format */ + zmq_frame_header_t* frame = (zmq_frame_header_t*)msg.data(); + + if (msg.size() == ZMQ_FRAME_SIZE(frame) && + frame->version == 1 && + frame->encoder == ZMQ_ENCODER_TOOLAME) { + datalen = frame->datasize; + data = ZMQ_FRAME_DATA(frame); + + m_stats.notifyPeakLevels(frame->audiolevel_left, + frame->audiolevel_right); + } + + + if (datalen == framesize) + { + if (m_frame_buffer.size() > m_config.buffer_size) { + etiLog.level(warn) << + "inputZMQ " << m_name << + " buffer full (" << m_frame_buffer.size() << ")," + " dropping incoming frame !"; + messageReceived = false; + } + else if (m_enable_input) { + // copy the input frame blockwise into the frame_buffer + auto framedata = new uint8_t[framesize]; + memcpy(framedata, data, framesize); + m_frame_buffer.push_back(framedata); + } + else { + return 0; + } + } + else { + etiLog.level(error) << + "inputZMQ " << m_name << + " verify bitrate: recv'd " << msg.size() << " B" << + ", need " << framesize << "."; + messageReceived = false; + } + + return messageReceived ? msg.size() : 0; +} + +/******** AAC+ input *******/ + +// Read a AAC+ superframe from the socket, cut it into five frames, +// and push to list +int ZmqAAC::readFromSocket(size_t framesize) +{ + bool messageReceived; + zmq::message_t msg; + + try { + messageReceived = m_zmq_sock.recv(&msg, ZMQ_DONTWAIT); + if (!messageReceived) { + return 0; + } + + } + catch (zmq::error_t& err) + { + etiLog.level(error) << + "Failed to receive AAC superframe from zmq socket " << + m_name << ": " << err.what(); + } + + /* This is the old 'one superframe per ZMQ message' format */ + uint8_t* data = (uint8_t*)msg.data(); + size_t datalen = msg.size(); + + /* Look for the new zmq_frame_header_t format */ + zmq_frame_header_t* frame = (zmq_frame_header_t*)msg.data(); + + if (msg.size() == ZMQ_FRAME_SIZE(frame) && + frame->version == 1 && + frame->encoder == ZMQ_ENCODER_FDK) { + datalen = frame->datasize; + data = ZMQ_FRAME_DATA(frame); + + m_stats.notifyPeakLevels(frame->audiolevel_left, + frame->audiolevel_right); + } + + + /* TS 102 563, Section 6: + * Audio super frames are transported in five successive DAB logical frames + * with additional error protection. + */ + if (datalen) + { + if (datalen == 5*framesize) + { + if (m_frame_buffer.size() > m_config.buffer_size) { + etiLog.level(warn) << + "inputZMQ " << m_name << + " buffer full (" << m_frame_buffer.size() << ")," + " dropping incoming superframe !"; + datalen = 0; + } + else if (m_enable_input) { + // copy the input frame blockwise into the frame_buffer + for (uint8_t* framestart = data; + framestart < &data[5*framesize]; + framestart += framesize) { + auto audioframe = new uint8_t[framesize]; + memcpy(audioframe, framestart, framesize); + m_frame_buffer.push_back(audioframe); + } + } + else { + datalen = 0; + } + } + else { + etiLog.level(error) << + "inputZMQ " << m_name << + " verify bitrate: recv'd " << msg.size() << " B" << + ", need " << 5*framesize << "."; + + datalen = 0; + } + } + else { + etiLog.level(error) << + "inputZMQ " << m_name << + " invalid frame received"; + } + + return datalen; +} + +/********* REMOTE CONTROL ***********/ + +void ZmqBase::set_parameter(const string& parameter, + const string& value) +{ + if (parameter == "buffer") { + size_t new_limit = atol(value.c_str()); + + if (new_limit < INPUT_ZMQ_MIN_BUFFER_SIZE) { + throw ParameterError("Desired buffer size too small." + " Minimum " STRINGIFY(INPUT_ZMQ_MIN_BUFFER_SIZE) ); + } + else if (new_limit > INPUT_ZMQ_MAX_BUFFER_SIZE) { + throw ParameterError("Desired buffer size too large." + " Maximum " STRINGIFY(INPUT_ZMQ_MAX_BUFFER_SIZE) ); + } + + m_config.buffer_size = new_limit; + } + else if (parameter == "prebuffering") { + size_t new_prebuf = atol(value.c_str()); + + if (new_prebuf < INPUT_ZMQ_MIN_BUFFER_SIZE) { + throw ParameterError("Desired prebuffering too small." + " Minimum " STRINGIFY(INPUT_ZMQ_MIN_BUFFER_SIZE) ); + } + else if (new_prebuf > INPUT_ZMQ_MAX_BUFFER_SIZE) { + throw ParameterError("Desired prebuffering too large." + " Maximum " STRINGIFY(INPUT_ZMQ_MAX_BUFFER_SIZE) ); + } + + m_config.prebuffering = new_prebuf; + } + else if (parameter == "enable") { + if (value == "1") { + m_enable_input = true; + } + else if (value == "0") { + m_enable_input = false; + } + else { + throw ParameterError("Value not understood, specify 0 or 1."); + } + } + else if (parameter == "encryption") { + if (value == "1") { + m_config.enable_encryption = true; + } + else if (value == "0") { + m_config.enable_encryption = false; + } + else { + throw ParameterError("Value not understood, specify 0 or 1."); + } + + try { + rebind(); + } + catch (std::runtime_error &e) { + stringstream ss; + ss << "Could not bind socket again with new keys." << + e.what(); + throw ParameterError(ss.str()); + } + } + else if (parameter == "secretkey") { + m_config.curve_secret_keyfile = value; + } + else if (parameter == "publickey") { + m_config.curve_public_keyfile = value; + } + else if (parameter == "encoderkey") { + m_config.curve_encoder_keyfile = value; + } + else { + stringstream ss; + ss << "Parameter '" << parameter << + "' is not exported by controllable " << get_rc_name(); + throw ParameterError(ss.str()); + } +} + +const string ZmqBase::get_parameter(const string& parameter) const +{ + stringstream ss; + if (parameter == "buffer") { + ss << m_config.buffer_size; + } + else if (parameter == "prebuffering") { + ss << m_config.prebuffering; + } + else if (parameter == "enable") { + if (m_enable_input) + ss << "true"; + else + ss << "false"; + } + else if (parameter == "encryption") { + if (m_config.enable_encryption) + ss << "true"; + else + ss << "false"; + } + else if (parameter == "secretkey") { + ss << m_config.curve_secret_keyfile; + } + else if (parameter == "publickey") { + ss << m_config.curve_public_keyfile; + } + else if (parameter == "encoderkey") { + ss << m_config.curve_encoder_keyfile; + } + else { + ss << "Parameter '" << parameter << + "' is not exported by controllable " << get_rc_name(); + throw ParameterError(ss.str()); + } + return ss.str(); + +} + +}; + +#endif + diff --git a/src/input/Zmq.h b/src/input/Zmq.h new file mode 100644 index 0000000..d1dd2d5 --- /dev/null +++ b/src/input/Zmq.h @@ -0,0 +1,271 @@ +/* + Copyright (C) 2009 Her Majesty the Queen in Right of Canada (Communications + Research Center Canada) + + Copyright (C) 2016 Matthias P. Braendli + http://www.opendigitalradio.org + + ZeroMQ input. see www.zeromq.org for more info + + For the AAC+ input, each zeromq message must contain one superframe, + or one zmq_frame_header_t followed by a superframe. + + For the MPEG input, each zeromq message must contain one frame. + + Encryption is provided by zmq_curve, see the corresponding manpage. + + From the ZeroMQ manpage 'zmq': + + The 0MQ lightweight messaging kernel is a library which extends the standard + socket interfaces with features traditionally provided by specialised + messaging middleware products. 0MQ sockets provide an abstraction of + asynchronous message queues, multiple messaging patterns, message filtering + (subscriptions), seamless access to multiple transport protocols and more. + */ +/* + This file is part of ODR-DabMux. + + It defines a ZeroMQ input for dabplus data. + + ODR-DabMux is free software: you can redistribute it and/or modify + it under the terms of the GNU General Public License as + published by the Free Software Foundation, either version 3 of the + License, or (at your option) any later version. + + ODR-DabMux is distributed in the hope that it will be useful, + but WITHOUT ANY WARRANTY; without even the implied warranty of + MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + GNU General Public License for more details. + + You should have received a copy of the GNU General Public License + along with ODR-DabMux. If not, see . + */ + +#pragma once + +#ifdef HAVE_CONFIG_H +# include "config.h" +#endif + +#ifdef HAVE_INPUT_ZEROMQ + +#include +#include +#include +#include "zmq.hpp" +#include "input/inputs.h" +#include "ManagementServer.h" + +namespace Inputs { + +/* The frame_buffer contains DAB logical frames as defined in + * TS 102 563, section 6. + * Five elements of this buffer make one AAC superframe (120ms audio) + */ + +// Minimum frame_buffer size in number of elements +// This is one AAC superframe, but you probably don't want to +// go that low anyway. +const size_t INPUT_ZMQ_MIN_BUFFER_SIZE = 5*1; // 120ms + +// Maximum frame_buffer size in number of elements +// One minute is clearly way over what everybody would +// want. +const size_t INPUT_ZMQ_MAX_BUFFER_SIZE = 5*500; // 60s + +/* The ZeroMQ Curve key is 40 bytes long in Z85 representation + * + * But we need to store it as zero-terminated string. + */ +const size_t CURVE_KEYLEN = 40; + +/* helper to invalidate a key */ +#define INVALIDATE_KEY(k) memset(k, 0, CURVE_KEYLEN+1) + +/* Verification for key validity */ +#define KEY_VALID(k) (k[0] != '\0') + +/* Read a key from file into key + * + * Returns 0 on success, negative value on failure + */ +int readkey(std::string& keyfile, char* key); + +struct dab_input_zmq_config_t +{ + /* The size of the internal buffer, measured in number + * of elements. + * + * Each element corresponds to five frames, + * or one AAC superframe. + */ + size_t buffer_size; + + /* The amount of prebuffering to do before we start streaming + * + * Same units as buffer_size + */ + size_t prebuffering; + + /* Whether to enforce encryption or not + */ + bool enable_encryption; + + /* Full path to file containing public key. + */ + std::string curve_public_keyfile; + + /* Full path to file containing secret key. + */ + std::string curve_secret_keyfile; + + /* Full path to file containing encoder public key. + */ + std::string curve_encoder_keyfile; +}; + +#define ZMQ_ENCODER_FDK 1 +#define ZMQ_ENCODER_TOOLAME 2 + +/* This defines the on-wire representation of a ZMQ message header. + * + * The data follows right after this header */ +struct zmq_frame_header_t +{ + uint16_t version; // we support version=1 now + uint16_t encoder; // see ZMQ_ENCODER_XYZ + + /* length of the 'data' field */ + uint32_t datasize; + + /* Audio level, peak, linear PCM */ + int16_t audiolevel_left; + int16_t audiolevel_right; + + /* Data follows this header */ +} __attribute__ ((packed)); + +/* The expected frame size incl data of the given frame */ +#define ZMQ_FRAME_SIZE(f) (sizeof(zmq_frame_header_t) + f->datasize) + +#define ZMQ_FRAME_DATA(f) ( ((uint8_t*)f)+sizeof(zmq_frame_header_t) ) + + +class ZmqBase : public InputBase, public RemoteControllable { + public: + ZmqBase(const std::string name, + dab_input_zmq_config_t config) + : RemoteControllable(name), + m_zmq_context(1), + m_zmq_sock(m_zmq_context, ZMQ_SUB), + m_zmq_sock_bound_to(""), + m_bitrate(0), + m_enable_input(true), + m_config(config), + m_stats(m_name), + m_prebuf_current(config.prebuffering) { + RC_ADD_PARAMETER(enable, + "If the input is enabled. Set to zero to empty the buffer."); + + RC_ADD_PARAMETER(encryption, + "If encryption is enabled or disabled [1 or 0]." + " If 1 is written, the keys are reloaded."); + + RC_ADD_PARAMETER(publickey, + "The multiplexer's public key file."); + + RC_ADD_PARAMETER(secretkey, + "The multiplexer's secret key file."); + + RC_ADD_PARAMETER(encoderkey, + "The encoder's public key file."); + + /* Set all keys to zero */ + INVALIDATE_KEY(m_curve_public_key); + INVALIDATE_KEY(m_curve_secret_key); + INVALIDATE_KEY(m_curve_encoder_key); + } + + virtual int open(const std::string& inputUri); + virtual int readFrame(void* buffer, int size); + virtual int setBitrate(int bitrate); + virtual int close(); + + /* Remote control */ + virtual void set_parameter(const std::string& parameter, + const std::string& value); + + /* Getting a parameter always returns a string. */ + virtual const std::string get_parameter(const std::string& parameter) const; + + protected: + virtual int readFromSocket(size_t framesize) = 0; + + virtual void rebind(); + + zmq::context_t m_zmq_context; + zmq::socket_t m_zmq_sock; // handle for the zmq socket + + /* If the socket is bound, this saves the endpoint, + * otherwise, it's an empty string + */ + std::string m_zmq_sock_bound_to; + int m_bitrate; + + /* set this to zero to empty the input buffer */ + bool m_enable_input; + + /* stores elements of type char[] */ + std::list m_frame_buffer; + + dab_input_zmq_config_t m_config; + + /* Key management, keys need to be zero-terminated */ + char m_curve_public_key[CURVE_KEYLEN+1]; + char m_curve_secret_key[CURVE_KEYLEN+1]; + char m_curve_encoder_key[CURVE_KEYLEN+1]; + + std::string m_inputUri; + + InputStat m_stats; + + private: + size_t m_prebuf_current; +}; + +class ZmqMPEG : public ZmqBase { + public: + ZmqMPEG(const std::string name, + dab_input_zmq_config_t config) + : ZmqBase(name, config) { + RC_ADD_PARAMETER(buffer, + "Size of the input buffer [mpeg frames]"); + + RC_ADD_PARAMETER(prebuffering, + "Min buffer level before streaming starts [mpeg frames]"); + } + + private: + virtual int readFromSocket(size_t framesize); +}; + +class ZmqAAC : public ZmqBase { + public: + ZmqAAC(const std::string name, + dab_input_zmq_config_t config) + : ZmqBase(name, config) { + RC_ADD_PARAMETER(buffer, + "Size of the input buffer [aac superframes]"); + + RC_ADD_PARAMETER(prebuffering, + "Min buffer level before streaming starts [aac superframes]"); + } + + private: + virtual int readFromSocket(size_t framesize); +}; + +}; +#endif // HAVE_INPUT_ZMQ + + -- cgit v1.2.3