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/input/Zmq.cpp | 625 ++++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 625 insertions(+) create mode 100644 src/input/Zmq.cpp (limited to 'src/input/Zmq.cpp') 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 + -- cgit v1.2.3 From 804fe1979f9ed7bef7badaf0aa08e35e09874772 Mon Sep 17 00:00:00 2001 From: "Matthias P. Braendli" Date: Sun, 30 Oct 2016 16:28:27 +0100 Subject: Add rudimentary file input No nonblock support yet --- src/ConfigParser.cpp | 11 +++ src/Makefile.am | 1 + src/dabInputMpegFile.cpp | 31 ------- src/input/File.cpp | 227 +++++++++++++++++++++++++++++++++++++++++++++++ src/input/File.h | 69 ++++++++++++++ src/input/Prbs.h | 4 - src/input/Zmq.cpp | 9 -- src/input/Zmq.h | 7 -- src/input/inputs.h | 52 +++++++++++ src/mpeg.c | 26 ++++++ src/mpeg.h | 7 ++ 11 files changed, 393 insertions(+), 51 deletions(-) create mode 100644 src/input/File.cpp create mode 100644 src/input/File.h create mode 100644 src/input/inputs.h (limited to 'src/input/Zmq.cpp') diff --git a/src/ConfigParser.cpp b/src/ConfigParser.cpp index bdc2099..2a8d3da 100644 --- a/src/ConfigParser.cpp +++ b/src/ConfigParser.cpp @@ -55,6 +55,7 @@ #include "input/Prbs.h" #include "input/Zmq.h" +#include "input/File.h" #ifdef _WIN32 @@ -930,6 +931,16 @@ static void setup_subchannel_from_ptree(DabSubchannel* subchan, if (nonblock) { // TODO } + + if (type == "audio") { + subchan->input = make_shared(); + } + else if (type == "dabplus") { + subchan->input = make_shared(); + } + else { + throw logic_error("Incomplete handling of file input"); + } } else if (proto == "tcp" || proto == "epgm" || diff --git a/src/Makefile.am b/src/Makefile.am index b8de4e8..b356566 100644 --- a/src/Makefile.am +++ b/src/Makefile.am @@ -50,6 +50,7 @@ odr_dabmux_SOURCES =DabMux.cpp DabMux.h \ input/inputs.h \ input/Prbs.cpp input/Prbs.h \ input/Zmq.cpp input/Zmq.h \ + input/File.cpp input/File.h \ dabOutput/dabOutput.h \ dabOutput/dabOutputFile.cpp \ dabOutput/dabOutputFifo.cpp \ diff --git a/src/dabInputMpegFile.cpp b/src/dabInputMpegFile.cpp index 804ea29..6f24f32 100644 --- a/src/dabInputMpegFile.cpp +++ b/src/dabInputMpegFile.cpp @@ -47,37 +47,6 @@ struct dabInputOperations dabInputMpegFileOperations = { }; -#define MPEG_FREQUENCY -2 -#define MPEG_PADDING -3 -#define MPEG_COPYRIGHT -4 -#define MPEG_ORIGINAL -5 -#define MPEG_EMPHASIS -6 -int checkDabMpegFrame(void* data) { - mpegHeader* header = (mpegHeader*)data; - unsigned long* headerData = (unsigned long*)data; - if ((*headerData & 0x0f0ffcff) == 0x0004fcff) return 0; - if ((*headerData & 0x0f0ffcff) == 0x0004f4ff) return 0; - if (getMpegFrequency(header) != 48000) { - if (getMpegFrequency(header) != 24000) { - return MPEG_FREQUENCY; - } - } - if (header->padding != 0) { - return MPEG_PADDING; - } - if (header->copyright != 0) { - return MPEG_COPYRIGHT; - } - if (header->original != 0) { - return MPEG_ORIGINAL; - } - if (header->emphasis != 0) { - return MPEG_EMPHASIS; - } - return -1; -} - - int dabInputMpegFileRead(dabInputOperations* ops, void* args, void* buffer, int size) { dabInputFileData* data = (dabInputFileData*)args; diff --git a/src/input/File.cpp b/src/input/File.cpp new file mode 100644 index 0000000..9721b97 --- /dev/null +++ b/src/input/File.cpp @@ -0,0 +1,227 @@ +/* + 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 + + */ +/* + 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 +#include +#include +#include +#include +#include +#ifndef _WIN32 +# define O_BINARY 0 +#endif + +#include "input/File.h" +#include "mpeg.h" + +namespace Inputs { + +int FileBase::open(const std::string& name) +{ + m_fd = ::open(name.c_str(), O_RDONLY | O_BINARY); + if (m_fd == -1) { + std::stringstream ss; + ss << "Could not open input file " << name << ": " << + strerror(errno); + throw std::runtime_error(ss.str()); + } + return 0; +} + +int FileBase::close() +{ + if (m_fd != -1) { + ::close(m_fd); + m_fd = -1; + } + return 0; +} + +int FileBase::rewind() +{ + return ::lseek(m_fd, 0, SEEK_SET); +} + +int MPEGFile::readFrame(void* buffer, int size) +{ + int result; + bool do_rewind = false; +READ_SUBCHANNEL: + if (m_parity) { + result = readData(m_fd, buffer, size, 2); + m_parity = false; + return 0; + } else { + result = readMpegHeader(m_fd, buffer, size); + if (result > 0) { + result = readMpegFrame(m_fd, buffer, size); + if (result < 0 && getMpegFrequency(buffer) == 24000) { + m_parity = true; + result = size; + } + } + } + switch (result) { + case MPEG_BUFFER_UNDERFLOW: + etiLog.log(warn, "data underflow -> frame muted\n"); + goto MUTE_SUBCHANNEL; + case MPEG_BUFFER_OVERFLOW: + etiLog.log(warn, "bitrate too high -> frame muted\n"); + goto MUTE_SUBCHANNEL; + case MPEG_FILE_EMPTY: + if (do_rewind) { + etiLog.log(error, "file rewinded and still empty " + "-> frame muted\n"); + goto MUTE_SUBCHANNEL; + } + else { + etiLog.log(info, "reach end of file -> rewinding\n"); + rewind(); + goto READ_SUBCHANNEL; + } + case MPEG_FILE_ERROR: + etiLog.log(alert, "can't read file (%i) -> frame muted\n", errno); + perror(""); + goto MUTE_SUBCHANNEL; + case MPEG_SYNC_NOT_FOUND: + etiLog.log(alert, "mpeg sync not found, maybe is not a valid file " + "-> frame muted\n"); + goto MUTE_SUBCHANNEL; + case MPEG_INVALID_FRAME: + etiLog.log(alert, "file is not a valid mpeg file " + "-> frame muted\n"); + goto MUTE_SUBCHANNEL; + default: + if (result < 0) { + etiLog.log(alert, + "unknown error (code = %i) -> frame muted\n", + result); +MUTE_SUBCHANNEL: + memset(buffer, 0, size); + } + else { + if (result < size) { + etiLog.log(warn, "bitrate too low from file " + "-> frame padded\n"); + memset((char*)buffer + result, 0, size - result); + } + + result = checkDabMpegFrame(buffer); + switch (result) { + case MPEG_FREQUENCY: + etiLog.log(error, "file has a frame with an invalid " + "frequency: %i, should be 48000 or 24000\n", + getMpegFrequency(buffer)); + break; + case MPEG_PADDING: + etiLog.log(warn, + "file has a frame with padding bit set\n"); + break; + case MPEG_COPYRIGHT: + result = 0; + break; + case MPEG_ORIGINAL: + result = 0; + break; + case MPEG_EMPHASIS: + etiLog.log(warn, + "file has a frame with emphasis bits set\n"); + break; + default: + if (result < 0) { + etiLog.log(alert, "mpeg file has an invalid DAB " + "mpeg frame (unknown reason: %i)\n", result); + } + break; + } + } + } + return result; +} + +int MPEGFile::setBitrate(int bitrate) +{ + if (bitrate == 0) { + char buffer[4]; + + if (readFrame(buffer, 4) == 0) { + bitrate = getMpegBitrate(buffer); + } + else { + bitrate = -1; + } + rewind(); + } + return bitrate; +} + + +int DABPlusFile::readFrame(void* buffer, int size) +{ + uint8_t* dataOut = reinterpret_cast(buffer); + + ssize_t ret = read(m_fd, dataOut, size); + + if (ret == -1) { + etiLog.log(alert, "ERROR: Can't read file\n"); + perror(""); + return -1; + } + + if (ret < size) { + ssize_t sizeOut = ret; + etiLog.log(info, "reach end of file -> rewinding\n"); + if (rewind() == -1) { + etiLog.log(alert, "ERROR: Can't rewind file\n"); + return -1; + } + + ret = read(m_fd, dataOut + sizeOut, size - sizeOut); + if (ret == -1) { + etiLog.log(alert, "ERROR: Can't read file\n"); + perror(""); + return -1; + } + + if (ret < size) { + etiLog.log(alert, "ERROR: Not enough data in file\n"); + return -1; + } + } + + return size; +} + +int DABPlusFile::setBitrate(int bitrate) +{ + if (bitrate <= 0) { + etiLog.log(error, "Invalid bitrate (%i)\n", bitrate); + return -1; + } + + return bitrate; +} + +}; diff --git a/src/input/File.h b/src/input/File.h new file mode 100644 index 0000000..61be8b1 --- /dev/null +++ b/src/input/File.h @@ -0,0 +1,69 @@ +/* + 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 + + */ +/* + 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 + +#include +#include +#include +#include "input/inputs.h" +#include "ManagementServer.h" + +namespace Inputs { + +class FileBase : public InputBase { + public: + virtual int open(const std::string& name); + virtual int readFrame(void* buffer, int size) = 0; + virtual int setBitrate(int bitrate) = 0; + virtual int close(); + + /* Rewind the file + * Returns -1 on failure, 0 on success + */ + virtual int rewind(); + protected: + // We use unix open() instead of fopen() because + // we want to do non-blocking I/O + int m_fd = -1; +}; + +class MPEGFile : public FileBase { + public: + virtual int readFrame(void* buffer, int size); + virtual int setBitrate(int bitrate); + + private: + bool m_parity = false; +}; + +class DABPlusFile : public FileBase { + public: + virtual int readFrame(void* buffer, int size); + virtual int setBitrate(int bitrate); +}; + + +}; diff --git a/src/input/Prbs.h b/src/input/Prbs.h index 47b52ad..1ad5047 100644 --- a/src/input/Prbs.h +++ b/src/input/Prbs.h @@ -28,10 +28,6 @@ #pragma once -#ifdef HAVE_CONFIG_H -# include "config.h" -#endif - #include #include "input/inputs.h" diff --git a/src/input/Zmq.cpp b/src/input/Zmq.cpp index 6ef5fce..985fad3 100644 --- a/src/input/Zmq.cpp +++ b/src/input/Zmq.cpp @@ -41,13 +41,6 @@ #include "input/Zmq.h" -#ifdef HAVE_CONFIG_H -# include "config.h" -#endif - -#ifdef HAVE_INPUT_ZEROMQ - -#include "zmq.hpp" #include #include #include @@ -621,5 +614,3 @@ const string ZmqBase::get_parameter(const string& parameter) const }; -#endif - diff --git a/src/input/Zmq.h b/src/input/Zmq.h index d1dd2d5..02fce3a 100644 --- a/src/input/Zmq.h +++ b/src/input/Zmq.h @@ -43,12 +43,6 @@ #pragma once -#ifdef HAVE_CONFIG_H -# include "config.h" -#endif - -#ifdef HAVE_INPUT_ZEROMQ - #include #include #include @@ -266,6 +260,5 @@ class ZmqAAC : public ZmqBase { }; }; -#endif // HAVE_INPUT_ZMQ diff --git a/src/input/inputs.h b/src/input/inputs.h new file mode 100644 index 0000000..3bc1fa4 --- /dev/null +++ b/src/input/inputs.h @@ -0,0 +1,52 @@ +/* + Copyright (C) 2004, 2005, 2006, 2007, 2008, 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 + */ +/* + 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 "Log.h" +#include "RemoteControl.h" +#include + +namespace Inputs { + +/* New input object base */ +class InputBase { + 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 ~InputBase() {} + protected: + InputBase() {} +}; + +}; + diff --git a/src/mpeg.c b/src/mpeg.c index f7aed34..fb6591a 100644 --- a/src/mpeg.c +++ b/src/mpeg.c @@ -219,3 +219,29 @@ int readMpegFrame(int file, void* data, int size) } return framelength; } + +int checkDabMpegFrame(void* data) { + mpegHeader* header = (mpegHeader*)data; + unsigned long* headerData = (unsigned long*)data; + if ((*headerData & 0x0f0ffcff) == 0x0004fcff) return 0; + if ((*headerData & 0x0f0ffcff) == 0x0004f4ff) return 0; + if (getMpegFrequency(header) != 48000) { + if (getMpegFrequency(header) != 24000) { + return MPEG_FREQUENCY; + } + } + if (header->padding != 0) { + return MPEG_PADDING; + } + if (header->copyright != 0) { + return MPEG_COPYRIGHT; + } + if (header->original != 0) { + return MPEG_ORIGINAL; + } + if (header->emphasis != 0) { + return MPEG_EMPHASIS; + } + return -1; +} + diff --git a/src/mpeg.h b/src/mpeg.h index aa7cfb6..15b9b80 100644 --- a/src/mpeg.h +++ b/src/mpeg.h @@ -75,6 +75,13 @@ ssize_t readData(int file, void* data, size_t size, unsigned int tries); int readMpegHeader(int file, void* data, int size); int readMpegFrame(int file, void* data, int size); +#define MPEG_FREQUENCY -2 +#define MPEG_PADDING -3 +#define MPEG_COPYRIGHT -4 +#define MPEG_ORIGINAL -5 +#define MPEG_EMPHASIS -6 +int checkDabMpegFrame(void* data); + #ifdef __cplusplus } #endif -- cgit v1.2.3 From 8db328c61832a92bf3f7641061b68767141104f3 Mon Sep 17 00:00:00 2001 From: "Matthias P. Braendli" Date: Fri, 4 Nov 2016 14:49:11 +0100 Subject: Change readFrame argument types --- src/dabInput.h | 2 +- src/input/File.cpp | 18 ++++++++---------- src/input/File.h | 6 +++--- src/input/Prbs.cpp | 8 +++----- src/input/Prbs.h | 2 +- src/input/Udp.cpp | 2 +- src/input/Udp.h | 2 +- src/input/Zmq.cpp | 2 +- src/input/Zmq.h | 2 +- src/input/inputs.h | 2 +- 10 files changed, 21 insertions(+), 25 deletions(-) (limited to 'src/input/Zmq.cpp') diff --git a/src/dabInput.h b/src/dabInput.h index 0accddb..d2c5f49 100644 --- a/src/dabInput.h +++ b/src/dabInput.h @@ -62,7 +62,7 @@ class DabInputCompatible : public DabInputBase { virtual int setbuf(int size) { return m_ops.setbuf(args, size); } - virtual int readFrame(void* buffer, int size) + virtual int readFrame(uint8_t* buffer, size_t size) { if (m_ops.lock) { m_ops.lock(args); diff --git a/src/input/File.cpp b/src/input/File.cpp index 9721b97..eb26136 100644 --- a/src/input/File.cpp +++ b/src/input/File.cpp @@ -64,7 +64,7 @@ int FileBase::rewind() return ::lseek(m_fd, 0, SEEK_SET); } -int MPEGFile::readFrame(void* buffer, int size) +int MPEGFile::readFrame(uint8_t* buffer, size_t size) { int result; bool do_rewind = false; @@ -122,7 +122,7 @@ MUTE_SUBCHANNEL: memset(buffer, 0, size); } else { - if (result < size) { + if (result < (ssize_t)size) { etiLog.log(warn, "bitrate too low from file " "-> frame padded\n"); memset((char*)buffer + result, 0, size - result); @@ -164,7 +164,7 @@ MUTE_SUBCHANNEL: int MPEGFile::setBitrate(int bitrate) { if (bitrate == 0) { - char buffer[4]; + uint8_t buffer[4]; if (readFrame(buffer, 4) == 0) { bitrate = getMpegBitrate(buffer); @@ -178,11 +178,9 @@ int MPEGFile::setBitrate(int bitrate) } -int DABPlusFile::readFrame(void* buffer, int size) +int DABPlusFile::readFrame(uint8_t* buffer, size_t size) { - uint8_t* dataOut = reinterpret_cast(buffer); - - ssize_t ret = read(m_fd, dataOut, size); + ssize_t ret = read(m_fd, buffer, size); if (ret == -1) { etiLog.log(alert, "ERROR: Can't read file\n"); @@ -190,7 +188,7 @@ int DABPlusFile::readFrame(void* buffer, int size) return -1; } - if (ret < size) { + if (ret < (ssize_t)size) { ssize_t sizeOut = ret; etiLog.log(info, "reach end of file -> rewinding\n"); if (rewind() == -1) { @@ -198,14 +196,14 @@ int DABPlusFile::readFrame(void* buffer, int size) return -1; } - ret = read(m_fd, dataOut + sizeOut, size - sizeOut); + ret = read(m_fd, buffer + sizeOut, size - sizeOut); if (ret == -1) { etiLog.log(alert, "ERROR: Can't read file\n"); perror(""); return -1; } - if (ret < size) { + if (ret < (ssize_t)size) { etiLog.log(alert, "ERROR: Not enough data in file\n"); return -1; } diff --git a/src/input/File.h b/src/input/File.h index 61be8b1..01f4f21 100644 --- a/src/input/File.h +++ b/src/input/File.h @@ -36,7 +36,7 @@ namespace Inputs { class FileBase : public InputBase { public: virtual int open(const std::string& name); - virtual int readFrame(void* buffer, int size) = 0; + virtual int readFrame(uint8_t* buffer, size_t size) = 0; virtual int setBitrate(int bitrate) = 0; virtual int close(); @@ -52,7 +52,7 @@ class FileBase : public InputBase { class MPEGFile : public FileBase { public: - virtual int readFrame(void* buffer, int size); + virtual int readFrame(uint8_t* buffer, size_t size); virtual int setBitrate(int bitrate); private: @@ -61,7 +61,7 @@ class MPEGFile : public FileBase { class DABPlusFile : public FileBase { public: - virtual int readFrame(void* buffer, int size); + virtual int readFrame(uint8_t* buffer, size_t size); virtual int setBitrate(int bitrate); }; diff --git a/src/input/Prbs.cpp b/src/input/Prbs.cpp index 8e5a9ae..607ce9f 100644 --- a/src/input/Prbs.cpp +++ b/src/input/Prbs.cpp @@ -77,12 +77,10 @@ int Prbs::open(const string& name) return 0; } -int Prbs::readFrame(void* buffer, int size) +int Prbs::readFrame(uint8_t* buffer, size_t size) { - unsigned char* cbuffer = reinterpret_cast(buffer); - - for (int i = 0; i < size; ++i) { - cbuffer[i] = m_prbs.step(); + for (size_t i = 0; i < size; ++i) { + buffer[i] = m_prbs.step(); } return size; diff --git a/src/input/Prbs.h b/src/input/Prbs.h index 1ad5047..3b2b7d4 100644 --- a/src/input/Prbs.h +++ b/src/input/Prbs.h @@ -38,7 +38,7 @@ namespace Inputs { class Prbs : public InputBase { public: virtual int open(const std::string& name); - virtual int readFrame(void* buffer, int size); + virtual int readFrame(uint8_t* buffer, size_t size); virtual int setBitrate(int bitrate); virtual int close(); diff --git a/src/input/Udp.cpp b/src/input/Udp.cpp index e534a06..a238d9b 100644 --- a/src/input/Udp.cpp +++ b/src/input/Udp.cpp @@ -87,7 +87,7 @@ int Udp::open(const std::string& name) return 0; } -int Udp::readFrame(void* buffer, int size) +int Udp::readFrame(uint8_t* buffer, size_t size) { uint8_t* data = reinterpret_cast(buffer); diff --git a/src/input/Udp.h b/src/input/Udp.h index b6705e9..379dbf3 100644 --- a/src/input/Udp.h +++ b/src/input/Udp.h @@ -36,7 +36,7 @@ namespace Inputs { class Udp : public InputBase { public: virtual int open(const std::string& name); - virtual int readFrame(void* buffer, int size); + virtual int readFrame(uint8_t* buffer, size_t size); virtual int setBitrate(int bitrate); virtual int close(); diff --git a/src/input/Zmq.cpp b/src/input/Zmq.cpp index 985fad3..a5601fa 100644 --- a/src/input/Zmq.cpp +++ b/src/input/Zmq.cpp @@ -246,7 +246,7 @@ int ZmqBase::setBitrate(int bitrate) } // size corresponds to a frame size. It is constant for a given bitrate -int ZmqBase::readFrame(void* buffer, int size) +int ZmqBase::readFrame(uint8_t* buffer, size_t size) { int rc; diff --git a/src/input/Zmq.h b/src/input/Zmq.h index 02fce3a..8d729e0 100644 --- a/src/input/Zmq.h +++ b/src/input/Zmq.h @@ -181,7 +181,7 @@ class ZmqBase : public InputBase, public RemoteControllable { } virtual int open(const std::string& inputUri); - virtual int readFrame(void* buffer, int size); + virtual int readFrame(uint8_t* buffer, size_t size); virtual int setBitrate(int bitrate); virtual int close(); diff --git a/src/input/inputs.h b/src/input/inputs.h index 3bc1fa4..bfb1fb6 100644 --- a/src/input/inputs.h +++ b/src/input/inputs.h @@ -39,7 +39,7 @@ namespace Inputs { class InputBase { public: virtual int open(const std::string& name) = 0; - virtual int readFrame(void* buffer, int size) = 0; + virtual int readFrame(uint8_t* buffer, size_t size) = 0; virtual int setBitrate(int bitrate) = 0; virtual int close() = 0; -- cgit v1.2.3