From c3aebce2ba6201a33bdc1f4f65d7ade0f42b7168 Mon Sep 17 00:00:00 2001 From: "Matthias P. Braendli" Date: Sun, 11 Sep 2016 23:24:20 +0200 Subject: Add ETI-over-TCP input --- src/DabMod.cpp | 14 ++++++ src/InputReader.h | 44 ++++++++++++++++-- src/InputTcpReader.cpp | 122 +++++++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 175 insertions(+), 5 deletions(-) create mode 100644 src/InputTcpReader.cpp (limited to 'src') diff --git a/src/DabMod.cpp b/src/DabMod.cpp index 6f35e22..904c3c8 100644 --- a/src/DabMod.cpp +++ b/src/DabMod.cpp @@ -178,6 +178,8 @@ int launch_modulator(int argc, char* argv[]) auto inputZeroMQReader = make_shared(); #endif + auto inputTcpReader = make_shared(); + struct sigaction sa; memset(&sa, 0, sizeof(struct sigaction)); sa.sa_handler = &signalHandler; @@ -616,6 +618,9 @@ int launch_modulator(int argc, char* argv[]) // if the name starts with zmq+XYZ://somewhere:port inputTransport = "zeromq"; } + else if (inputName.substr(0, 6) == "tcp://") { + inputTransport = "tcp"; + } } else { inputName = "/dev/stdin"; @@ -705,6 +710,10 @@ int launch_modulator(int argc, char* argv[]) m.inputReader = inputZeroMQReader.get(); #endif } + else if (inputTransport == "tcp") { + inputTcpReader->Open(inputName); + m.inputReader = inputTcpReader.get(); + } else { fprintf(stderr, "Error, invalid input transport %s selected!\n", inputTransport.c_str()); @@ -816,6 +825,11 @@ int launch_modulator(int argc, char* argv[]) m.inputReader = inputZeroMQReader.get(); #endif } + else if (inputTransport == "tcp") { + inputTcpReader = make_shared(); + inputTcpReader->Open(inputName); + m.inputReader = inputTcpReader.get(); + } break; case run_modulator_state_t::reconfigure: etiLog.level(warn) << "Detected change in ensemble configuration."; diff --git a/src/InputReader.h b/src/InputReader.h index daacc9e..4d0792b 100644 --- a/src/InputReader.h +++ b/src/InputReader.h @@ -3,8 +3,10 @@ Her Majesty the Queen in Right of Canada (Communications Research Center Canada) - Copyrigth (C) 2013, 2015 + Copyright (C) 2016 Matthias P. Braendli, matthias.braendli@mpb.li + + http://www.opendigitalradio.org */ /* This file is part of ODR-DabMod. @@ -39,6 +41,14 @@ #endif #include "porting.h" #include "Log.h" +#include +#include +#include +#include +#include +#define SOCKET int +#define INVALID_SOCKET -1 +#define SOCKET_ERROR -1 /* Known types of input streams. Description taken from the CRC mmbTools forum. @@ -115,8 +125,8 @@ class InputFileReader : public InputReader } private: - InputFileReader(const InputFileReader& other); - InputFileReader& operator=(const InputFileReader& other); + InputFileReader(const InputFileReader& other) = delete; + InputFileReader& operator=(const InputFileReader& other) = delete; int IdentifyType(); @@ -134,6 +144,30 @@ class InputFileReader : public InputReader // after 2**32 * 24ms ~= 3.3 years }; +class InputTcpReader : public InputReader +{ + public: + InputTcpReader(); + ~InputTcpReader(); + + // Endpoint is either host:port or tcp://host:port + void Open(const std::string& endpoint); + + // Put next frame into buffer. This function will never write more than + // 6144 bytes into buffer. + // returns number of bytes written to buffer, 0 on eof, -1 on error + virtual int GetNextFrame(void* buffer); + + // Print some information + virtual void PrintInfo(); + + private: + InputTcpReader(const InputTcpReader& other) = delete; + InputTcpReader& operator=(const InputTcpReader& other) = delete; + SOCKET m_sock; + std::string m_uri; +}; + struct zmq_input_overflow : public std::exception { const char* what () const throw () @@ -201,8 +235,8 @@ class InputZeroMQReader : public InputReader void PrintInfo(); private: - InputZeroMQReader(const InputZeroMQReader& other); - InputZeroMQReader& operator=(const InputZeroMQReader& other); + InputZeroMQReader(const InputZeroMQReader& other) = delete; + InputZeroMQReader& operator=(const InputZeroMQReader& other) = delete; std::string uri_; InputZeroMQWorker worker_; diff --git a/src/InputTcpReader.cpp b/src/InputTcpReader.cpp new file mode 100644 index 0000000..34ccb2e --- /dev/null +++ b/src/InputTcpReader.cpp @@ -0,0 +1,122 @@ +/* + Copyright (C) 2005, 2006, 2007, 2008, 2009, 2010, 2011, 2012 + 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-DabMod. + + ODR-DabMod 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-DabMod 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-DabMod. If not, see . + */ + +#ifdef HAVE_CONFIG_H +# include "config.h" +#endif + +#include "InputReader.h" +#include "PcDebug.h" +#include "Utils.h" + +InputTcpReader::InputTcpReader() +{ + if ((m_sock = socket(PF_INET, SOCK_STREAM, 0)) == INVALID_SOCKET) { + throw std::runtime_error("Can't create TCP socket"); + } +} + +InputTcpReader::~InputTcpReader() +{ + if (m_sock != INVALID_SOCKET) { + close(m_sock); + } +} + +void InputTcpReader::Open(const std::string& endpoint) +{ + std::string hostname; + if (endpoint.compare(0, 6, "tcp://") == 0) { + hostname = endpoint.substr(6, std::string::npos); + } + else { + hostname = endpoint; + } + + size_t colon_pos = hostname.find(":"); + if (colon_pos == std::string::npos) { + std::stringstream ss; + ss << "Could not parse TCP endpoint " << endpoint; + throw std::runtime_error(ss.str()); + } + + long port = strtol(hostname.c_str() + colon_pos + 1, NULL, 10); + if (errno == ERANGE) { + std::stringstream ss; + ss << "Could not parse port in TCP endpoint " << endpoint; + throw std::runtime_error(ss.str()); + } + + hostname = hostname.substr(0, colon_pos); + + struct sockaddr_in addr; + addr.sin_family = PF_INET; + addr.sin_addr.s_addr = htons(INADDR_ANY); + addr.sin_port = htons(port); + + hostent *host = gethostbyname(hostname.c_str()); + if (host) { + addr.sin_addr = *(in_addr *)(host->h_addr); + } + else { + std::stringstream ss; + ss << "Could not resolve hostname " << hostname << ": " << strerror(errno); + throw std::runtime_error(ss.str()); + } + + if (connect(m_sock, (struct sockaddr*)&addr, sizeof(addr)) == -1) { + std::stringstream ss; + ss << "Could not connect to " << hostname << ":" << port << " :" << strerror(errno); + throw std::runtime_error(ss.str()); + } + + m_uri = endpoint; +} + +int InputTcpReader::GetNextFrame(void* buffer) +{ + uint8_t* buf = (uint8_t*)buffer; + + const size_t framesize = 6144; + + ssize_t r = recv(m_sock, buf, framesize, MSG_WAITALL); + + if (r == -1) { + std::stringstream ss; + ss << "Could not receive from socket :" << strerror(errno); + throw std::runtime_error(ss.str()); + } + + return r; +} + +void InputTcpReader::PrintInfo() +{ + fprintf(stderr, "Input TCP:\n"); + fprintf(stderr, " Receiving from %s\n\n", m_uri.c_str()); +} + -- cgit v1.2.3