diff options
| -rw-r--r-- | Makefile.am | 23 | ||||
| -rw-r--r-- | README.md | 3 | ||||
| -rw-r--r-- | contrib/Socket.cpp | 894 | ||||
| -rw-r--r-- | contrib/Socket.h | 294 | ||||
| -rw-r--r-- | contrib/ThreadsafeQueue.h | 176 | ||||
| -rw-r--r-- | contrib/edi/AFPacket.cpp | 96 | ||||
| -rw-r--r-- | contrib/edi/AFPacket.h | 61 | ||||
| -rw-r--r-- | contrib/edi/Config.h | 84 | ||||
| -rw-r--r-- | contrib/edi/Interleaver.cpp | 122 | ||||
| -rw-r--r-- | contrib/edi/Interleaver.h | 74 | ||||
| -rw-r--r-- | contrib/edi/PFT.cpp | 325 | ||||
| -rw-r--r-- | contrib/edi/PFT.h | 77 | ||||
| -rw-r--r-- | contrib/edi/ReedSolomon.cpp | 116 | ||||
| -rw-r--r-- | contrib/edi/ReedSolomon.h | 56 | ||||
| -rw-r--r-- | contrib/edi/TagItems.cpp | 215 | ||||
| -rw-r--r-- | contrib/edi/TagItems.h | 139 | ||||
| -rw-r--r-- | contrib/edi/TagPacket.cpp | 75 | ||||
| -rw-r--r-- | contrib/edi/TagPacket.h | 56 | ||||
| -rw-r--r-- | contrib/edi/Transport.cpp | 190 | ||||
| -rw-r--r-- | contrib/edi/Transport.h | 71 | ||||
| -rw-r--r-- | contrib/edi/crc.c | 266 | ||||
| -rw-r--r-- | contrib/edi/crc.h | 59 | ||||
| -rw-r--r-- | src/Outputs.cpp | 101 | ||||
| -rw-r--r-- | src/Outputs.h | 30 | ||||
| -rw-r--r-- | src/SampleQueue.h | 5 | ||||
| -rw-r--r-- | src/odr-audioenc.cpp | 752 |
26 files changed, 3997 insertions, 363 deletions
diff --git a/Makefile.am b/Makefile.am index d60df2f..aea695c 100644 --- a/Makefile.am +++ b/Makefile.am @@ -79,7 +79,8 @@ odr_audioenc_LDADD = libtoolame-dab.la \ $(odr_audioenc_LDADD_ALSA) \ $(LIBVLC_LIBS) $(LIBFDKAAC_LIBS) odr_audioenc_CPPFLAGS = $(AM_CPPFLAGS) $(LIBFDKAAC_CFLAGS) $(GITVERSION_FLAGS) \ - -ggdb -O2 + -ggdb -O2 -Isrc -Icontrib + odr_audioenc_SOURCES = src/odr-audioenc.cpp \ src/FileInput.cpp \ src/FileInput.h \ @@ -101,6 +102,26 @@ odr_audioenc_SOURCES = src/odr-audioenc.cpp \ src/utils.h \ src/wavfile.cpp \ src/common.h \ + contrib/Socket.cpp \ + contrib/Socket.h \ + contrib/edi/AFPacket.cpp \ + contrib/edi/AFPacket.h \ + contrib/edi/Config.h \ + contrib/edi/crc.c \ + contrib/edi/crc.h \ + contrib/edi/Interleaver.cpp \ + contrib/edi/Interleaver.h \ + contrib/edi/PFT.cpp \ + contrib/edi/PFT.h \ + contrib/edi/ReedSolomon.cpp \ + contrib/edi/ReedSolomon.h \ + contrib/edi/TagItems.cpp \ + contrib/edi/TagItems.h \ + contrib/edi/TagPacket.cpp \ + contrib/edi/TagPacket.h \ + contrib/edi/ThreadsafeQueue.h \ + contrib/edi/Transport.cpp \ + contrib/edi/Transport.h \ $(FEC_SOURCES) bin_PROGRAMS = odr-audioenc$(EXEEXT) @@ -11,7 +11,7 @@ in the [repository](https://github.com/Opendigitalradio/fdk-aac.git). The main tool is the *odr-audioenc* encoder, which can read audio from a file (raw or wav), from an ALSA source, from JACK or using libVLC, -and encode to a file, a pipe, or to a ZeroMQ output compatible with ODR-DabMux. +and encode to a file, a pipe, to a ZeroMQ or EDI output compatible with ODR-DabMux. The libVLC input allows the encoder to use all inputs supported by VLC, and therefore also webstreams and other network sources. @@ -282,6 +282,7 @@ The ODR-AudioEnc project contains http://www.apache.org/licenses/LICENSE-2.0 - libtoolame-dab, derived from TooLAME, licensed under LGPL v2.1 or later. See libtoolame-dab/LGPL.txt. This is built into a shared library. + - EDI output (files in src/edi) are GPLv3+ The odr-audioenc binary is linked against the libtoolame-dab and fdk-aac shared libraries. diff --git a/contrib/Socket.cpp b/contrib/Socket.cpp new file mode 100644 index 0000000..d14902e --- /dev/null +++ b/contrib/Socket.cpp @@ -0,0 +1,894 @@ +/* + Copyright (C) 2003, 2004, 2005, 2006, 2007, 2008, 2009 Her Majesty the + Queen in Right of Canada (Communications Research Center Canada) + + Copyright (C) 2019 + Matthias P. Braendli, matthias.braendli@mpb.li + + http://www.opendigitalradio.org + */ +/* + This program 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. + + This program 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 this program. If not, see <https://www.gnu.org/licenses/>. +*/ + +#include "Socket.h" + +#include <iostream> +#include <cstdio> +#include <cstring> +#include <cerrno> +#include <fcntl.h> +#include <poll.h> + +namespace Socket { + +using namespace std; + +void InetAddress::resolveUdpDestination(const std::string& destination, int port) +{ + char service[NI_MAXSERV]; + snprintf(service, NI_MAXSERV-1, "%d", port); + + struct addrinfo hints; + memset(&hints, 0, sizeof(struct addrinfo)); + hints.ai_family = AF_INET; + hints.ai_socktype = SOCK_DGRAM; /* Datagram socket */ + hints.ai_flags = 0; + hints.ai_protocol = 0; + + struct addrinfo *result, *rp; + int s = getaddrinfo(destination.c_str(), service, &hints, &result); + if (s != 0) { + throw runtime_error(string("getaddrinfo failed: ") + gai_strerror(s)); + } + + for (rp = result; rp != nullptr; rp = rp->ai_next) { + // Take the first result + memcpy(&addr, rp->ai_addr, rp->ai_addrlen); + break; + } + + freeaddrinfo(result); + + if (rp == nullptr) { + throw runtime_error("Could not resolve"); + } +} + +UDPPacket::UDPPacket() { } + +UDPPacket::UDPPacket(size_t initSize) : + buffer(initSize) +{ } + + +UDPSocket::UDPSocket() : + m_sock(INVALID_SOCKET) +{ + reinit(0, ""); +} + +UDPSocket::UDPSocket(int port) : + m_sock(INVALID_SOCKET) +{ + reinit(port, ""); +} + +UDPSocket::UDPSocket(int port, const std::string& name) : + m_sock(INVALID_SOCKET) +{ + reinit(port, name); +} + + +void UDPSocket::setBlocking(bool block) +{ + int res = fcntl(m_sock, F_SETFL, block ? 0 : O_NONBLOCK); + if (res == -1) { + throw runtime_error(string("Can't change blocking state of socket: ") + strerror(errno)); + } +} + +void UDPSocket::reinit(int port) +{ + return reinit(port, ""); +} + +void UDPSocket::reinit(int port, const std::string& name) +{ + if (m_sock != INVALID_SOCKET) { + ::close(m_sock); + } + + if (port == 0) { + // No need to bind to a given port, creating the + // socket is enough + m_sock = ::socket(AF_INET, SOCK_DGRAM, 0); + return; + } + + char service[NI_MAXSERV]; + snprintf(service, NI_MAXSERV-1, "%d", port); + + struct addrinfo hints; + memset(&hints, 0, sizeof(struct addrinfo)); + hints.ai_family = AF_INET; + hints.ai_socktype = SOCK_DGRAM; /* Datagram socket */ + hints.ai_flags = AI_PASSIVE; /* For wildcard IP address */ + hints.ai_protocol = 0; /* Any protocol */ + hints.ai_canonname = nullptr; + hints.ai_addr = nullptr; + hints.ai_next = nullptr; + + struct addrinfo *result, *rp; + int s = getaddrinfo(name.empty() ? nullptr : name.c_str(), + port == 0 ? nullptr : service, + &hints, &result); + if (s != 0) { + throw runtime_error(string("getaddrinfo failed: ") + gai_strerror(s)); + } + + /* getaddrinfo() returns a list of address structures. + Try each address until we successfully bind(2). + If socket(2) (or bind(2)) fails, we (close the socket + and) try the next address. */ + for (rp = result; rp != nullptr; rp = rp->ai_next) { + int sfd = ::socket(rp->ai_family, rp->ai_socktype, rp->ai_protocol); + if (sfd == -1) { + continue; + } + + if (::bind(sfd, rp->ai_addr, rp->ai_addrlen) == 0) { + m_sock = sfd; + break; + } + + ::close(sfd); + } + + freeaddrinfo(result); + + if (rp == nullptr) { + throw runtime_error("Could not bind"); + } +} + +void UDPSocket::close() +{ + if (m_sock != INVALID_SOCKET) { + ::close(m_sock); + } + + m_sock = INVALID_SOCKET; +} + +UDPSocket::~UDPSocket() +{ + if (m_sock != INVALID_SOCKET) { + ::close(m_sock); + } +} + + +UDPPacket UDPSocket::receive(size_t max_size) +{ + UDPPacket packet(max_size); + socklen_t addrSize; + addrSize = sizeof(*packet.address.as_sockaddr()); + ssize_t ret = recvfrom(m_sock, + packet.buffer.data(), + packet.buffer.size(), + 0, + packet.address.as_sockaddr(), + &addrSize); + + if (ret == SOCKET_ERROR) { + packet.buffer.resize(0); + + // This suppresses the -Wlogical-op warning +#if EAGAIN == EWOULDBLOCK + if (errno == EAGAIN) { +#else + if (errno == EAGAIN or errno == EWOULDBLOCK) { +#endif + return 0; + } + throw runtime_error(string("Can't receive data: ") + strerror(errno)); + } + + packet.buffer.resize(ret); + return packet; +} + +void UDPSocket::send(UDPPacket& packet) +{ + const int ret = sendto(m_sock, packet.buffer.data(), packet.buffer.size(), 0, + packet.address.as_sockaddr(), sizeof(*packet.address.as_sockaddr())); + if (ret == SOCKET_ERROR && errno != ECONNREFUSED) { + throw runtime_error(string("Can't send UDP packet: ") + strerror(errno)); + } +} + + +void UDPSocket::send(const std::vector<uint8_t>& data, InetAddress destination) +{ + const int ret = sendto(m_sock, data.data(), data.size(), 0, + destination.as_sockaddr(), sizeof(*destination.as_sockaddr())); + if (ret == SOCKET_ERROR && errno != ECONNREFUSED) { + throw runtime_error(string("Can't send UDP packet: ") + strerror(errno)); + } +} + +void UDPSocket::joinGroup(const char* groupname, const char* if_addr) +{ + ip_mreqn group; + if ((group.imr_multiaddr.s_addr = inet_addr(groupname)) == INADDR_NONE) { + throw runtime_error("Cannot convert multicast group name"); + } + if (!IN_MULTICAST(ntohl(group.imr_multiaddr.s_addr))) { + throw runtime_error("Group name is not a multicast address"); + } + + if (if_addr) { + group.imr_address.s_addr = inet_addr(if_addr); + } + else { + group.imr_address.s_addr = htons(INADDR_ANY); + } + group.imr_ifindex = 0; + if (setsockopt(m_sock, IPPROTO_IP, IP_ADD_MEMBERSHIP, &group, sizeof(group)) + == SOCKET_ERROR) { + throw runtime_error(string("Can't join multicast group") + strerror(errno)); + } +} + +void UDPSocket::setMulticastSource(const char* source_addr) +{ + struct in_addr addr; + if (inet_aton(source_addr, &addr) == 0) { + throw runtime_error(string("Can't parse source address") + strerror(errno)); + } + + if (setsockopt(m_sock, IPPROTO_IP, IP_MULTICAST_IF, &addr, sizeof(addr)) + == SOCKET_ERROR) { + throw runtime_error(string("Can't set source address") + strerror(errno)); + } +} + +void UDPSocket::setMulticastTTL(int ttl) +{ + if (setsockopt(m_sock, IPPROTO_IP, IP_MULTICAST_TTL, &ttl, sizeof(ttl)) + == SOCKET_ERROR) { + throw runtime_error(string("Can't set multicast ttl") + strerror(errno)); + } +} + +UDPReceiver::UDPReceiver() { } + +UDPReceiver::~UDPReceiver() { + m_stop = true; + m_sock.close(); + if (m_thread.joinable()) { + m_thread.join(); + } +} + +void UDPReceiver::start(int port, const string& bindto, const string& mcastaddr, size_t max_packets_queued) { + m_port = port; + m_bindto = bindto; + m_mcastaddr = mcastaddr; + m_max_packets_queued = max_packets_queued; + m_thread = std::thread(&UDPReceiver::m_run, this); +} + +std::vector<uint8_t> UDPReceiver::get_packet_buffer() +{ + if (m_stop) { + throw runtime_error("UDP Receiver not running"); + } + + UDPPacket p; + m_packets.wait_and_pop(p); + + return p.buffer; +} + +void UDPReceiver::m_run() +{ + // Ensure that stop is set to true in case of exception or return + struct SetStopOnDestruct { + SetStopOnDestruct(atomic<bool>& stop) : m_stop(stop) {} + ~SetStopOnDestruct() { m_stop = true; } + private: atomic<bool>& m_stop; + } autoSetStop(m_stop); + + if (IN_MULTICAST(ntohl(inet_addr(m_mcastaddr.c_str())))) { + m_sock.reinit(m_port, m_mcastaddr); + m_sock.setMulticastSource(m_bindto.c_str()); + m_sock.joinGroup(m_mcastaddr.c_str(), m_bindto.c_str()); + } + else { + m_sock.reinit(m_port, m_bindto); + } + + while (not m_stop) { + constexpr size_t packsize = 8192; + try { + auto packet = m_sock.receive(packsize); + if (packet.buffer.size() == packsize) { + // TODO replace fprintf + fprintf(stderr, "Warning, possible UDP truncation\n"); + } + + // If this blocks, the UDP socket will lose incoming packets + m_packets.push_wait_if_full(packet, m_max_packets_queued); + } + catch (const std::runtime_error& e) { + // TODO replace fprintf + // TODO handle intr + fprintf(stderr, "Socket error: %s\n", e.what()); + m_stop = true; + } + } +} + + +TCPSocket::TCPSocket() +{ +} + +TCPSocket::~TCPSocket() +{ + if (m_sock != -1) { + ::close(m_sock); + } +} + +TCPSocket::TCPSocket(TCPSocket&& other) : + m_sock(other.m_sock), + m_remote_address(move(other.m_remote_address)) +{ + if (other.m_sock != -1) { + other.m_sock = -1; + } +} + +TCPSocket& TCPSocket::operator=(TCPSocket&& other) +{ + swap(m_remote_address, other.m_remote_address); + + m_sock = other.m_sock; + if (other.m_sock != -1) { + other.m_sock = -1; + } + + return *this; +} + +bool TCPSocket::valid() const +{ + return m_sock != -1; +} + +void TCPSocket::connect(const std::string& hostname, int port) +{ + if (m_sock != INVALID_SOCKET) { + throw std::logic_error("You may only connect an invalid TCPSocket"); + } + + char service[NI_MAXSERV]; + snprintf(service, NI_MAXSERV-1, "%d", port); + + /* Obtain address(es) matching host/port */ + struct addrinfo hints; + memset(&hints, 0, sizeof(struct addrinfo)); + hints.ai_family = AF_INET; + hints.ai_socktype = SOCK_STREAM; + hints.ai_flags = 0; + hints.ai_protocol = 0; + + struct addrinfo *result, *rp; + int s = getaddrinfo(hostname.c_str(), service, &hints, &result); + if (s != 0) { + throw runtime_error(string("getaddrinfo failed: ") + gai_strerror(s)); + } + + /* getaddrinfo() returns a list of address structures. + Try each address until we successfully connect(2). + If socket(2) (or connect(2)) fails, we (close the socket + and) try the next address. */ + + for (rp = result; rp != nullptr; rp = rp->ai_next) { + int sfd = ::socket(rp->ai_family, rp->ai_socktype, + rp->ai_protocol); + if (sfd == -1) + continue; + + int ret = ::connect(sfd, rp->ai_addr, rp->ai_addrlen); + if (ret != -1 or (ret == -1 and errno == EINPROGRESS)) { + // As the TCPClient could set the socket to nonblocking, we + // must handle EINPROGRESS here + m_sock = sfd; + break; + } + + ::close(sfd); + } + + if (m_sock != INVALID_SOCKET) { +#if defined(HAVE_SO_NOSIGPIPE) + int val = 1; + if (setsockopt(m_sock, SOL_SOCKET, SO_NOSIGPIPE, &val, sizeof(val)) + == SOCKET_ERROR) { + throw std::runtime_error("Can't set SO_NOSIGPIPE"); + } +#endif + } + + freeaddrinfo(result); /* No longer needed */ + + if (rp == nullptr) { + throw runtime_error("Could not connect"); + } + +} + +void TCPSocket::listen(int port, const string& name) +{ + if (m_sock != INVALID_SOCKET) { + throw std::logic_error("You may only listen with an invalid TCPSocket"); + } + + char service[NI_MAXSERV]; + snprintf(service, NI_MAXSERV-1, "%d", port); + + struct addrinfo hints; + memset(&hints, 0, sizeof(struct addrinfo)); + hints.ai_family = AF_INET; + hints.ai_socktype = SOCK_STREAM; + hints.ai_flags = AI_PASSIVE; /* For wildcard IP address */ + hints.ai_protocol = 0; + hints.ai_canonname = nullptr; + hints.ai_addr = nullptr; + hints.ai_next = nullptr; + + struct addrinfo *result, *rp; + int s = getaddrinfo(name.empty() ? nullptr : name.c_str(), service, &hints, &result); + if (s != 0) { + throw runtime_error(string("getaddrinfo failed: ") + gai_strerror(s)); + } + + /* getaddrinfo() returns a list of address structures. + Try each address until we successfully bind(2). + If socket(2) (or bind(2)) fails, we (close the socket + and) try the next address. */ + for (rp = result; rp != nullptr; rp = rp->ai_next) { + int sfd = ::socket(rp->ai_family, rp->ai_socktype, rp->ai_protocol); + if (sfd == -1) { + continue; + } + + if (::bind(sfd, rp->ai_addr, rp->ai_addrlen) == 0) { + m_sock = sfd; + break; + } + + ::close(sfd); + } + + freeaddrinfo(result); + + if (m_sock != INVALID_SOCKET) { +#if defined(HAVE_SO_NOSIGPIPE) + int val = 1; + if (setsockopt(m_sock, SOL_SOCKET, SO_NOSIGPIPE, + &val, sizeof(val)) < 0) { + throw std::runtime_error("Can't set SO_NOSIGPIPE"); + } +#endif + + int ret = ::listen(m_sock, 0); + if (ret == -1) { + throw std::runtime_error(string("Could not listen: ") + strerror(errno)); + } + } + + if (rp == nullptr) { + throw runtime_error("Could not bind"); + } +} + +void TCPSocket::close() +{ + ::close(m_sock); + m_sock = -1; +} + +TCPSocket TCPSocket::accept(int timeout_ms) +{ + if (timeout_ms == 0) { + InetAddress remote_addr; + socklen_t client_len = sizeof(remote_addr.addr); + int sockfd = ::accept(m_sock, remote_addr.as_sockaddr(), &client_len); + TCPSocket s(sockfd, remote_addr); + return s; + } + else { + struct pollfd fds[1]; + fds[0].fd = m_sock; + fds[0].events = POLLIN; + + int retval = poll(fds, 1, timeout_ms); + + if (retval == -1) { + std::string errstr(strerror(errno)); + throw std::runtime_error("TCP Socket accept error: " + errstr); + } + else if (retval > 0) { + InetAddress remote_addr; + socklen_t client_len = sizeof(remote_addr.addr); + int sockfd = ::accept(m_sock, remote_addr.as_sockaddr(), &client_len); + TCPSocket s(sockfd, remote_addr); + return s; + } + else { + TCPSocket s(-1); + return s; + } + } +} + +ssize_t TCPSocket::sendall(const void *buffer, size_t buflen) +{ + uint8_t *buf = (uint8_t*)buffer; + while (buflen > 0) { + /* On Linux, the MSG_NOSIGNAL flag ensures that the process + * would not receive a SIGPIPE and die. + * Other systems have SO_NOSIGPIPE set on the socket for the + * same effect. */ +#if defined(HAVE_MSG_NOSIGNAL) + const int flags = MSG_NOSIGNAL; +#else + const int flags = 0; +#endif + ssize_t sent = ::send(m_sock, buf, buflen, flags); + if (sent < 0) { + return -1; + } + else { + buf += sent; + buflen -= sent; + } + } + return buflen; +} + +ssize_t TCPSocket::send(const void* data, size_t size, int timeout_ms) +{ + if (timeout_ms) { + struct pollfd fds[1]; + fds[0].fd = m_sock; + fds[0].events = POLLOUT; + + const int retval = poll(fds, 1, timeout_ms); + + if (retval == -1) { + throw std::runtime_error(string("TCP Socket send error on poll(): ") + strerror(errno)); + } + else if (retval == 0) { + // Timed out + return 0; + } + } + + /* On Linux, the MSG_NOSIGNAL flag ensures that the process would not + * receive a SIGPIPE and die. + * Other systems have SO_NOSIGPIPE set on the socket for the same effect. */ +#if defined(HAVE_MSG_NOSIGNAL) + const int flags = MSG_NOSIGNAL; +#else + const int flags = 0; +#endif + const ssize_t ret = ::send(m_sock, (const char*)data, size, flags); + + if (ret == SOCKET_ERROR) { + throw std::runtime_error(string("TCP Socket send error: ") + strerror(errno)); + } + return ret; +} + +ssize_t TCPSocket::recv(void *buffer, size_t length, int flags) +{ + ssize_t ret = ::recv(m_sock, buffer, length, flags); + if (ret == -1) { + std::string errstr(strerror(errno)); + throw std::runtime_error("TCP receive error: " + errstr); + } + return ret; +} + +ssize_t TCPSocket::recv(void *buffer, size_t length, int flags, int timeout_ms) +{ + struct pollfd fds[1]; + fds[0].fd = m_sock; + fds[0].events = POLLIN; + + int retval = poll(fds, 1, timeout_ms); + + if (retval == -1 and errno == EINTR) { + throw Interrupted(); + } + else if (retval == -1) { + std::string errstr(strerror(errno)); + throw std::runtime_error("TCP receive with poll() error: " + errstr); + } + else if (retval > 0 and (fds[0].revents | POLLIN)) { + ssize_t ret = ::recv(m_sock, buffer, length, flags); + if (ret == -1) { + if (errno == ECONNREFUSED) { + return 0; + } + std::string errstr(strerror(errno)); + throw std::runtime_error("TCP receive after poll() error: " + errstr); + } + return ret; + } + else { + throw Timeout(); + } +} + +TCPSocket::TCPSocket(int sockfd) : + m_sock(sockfd), + m_remote_address() +{ } + +TCPSocket::TCPSocket(int sockfd, InetAddress remote_address) : + m_sock(sockfd), + m_remote_address(remote_address) +{ } + +void TCPClient::connect(const std::string& hostname, int port) +{ + m_hostname = hostname; + m_port = port; + reconnect(); +} + +ssize_t TCPClient::recv(void *buffer, size_t length, int flags, int timeout_ms) +{ + try { + ssize_t ret = m_sock.recv(buffer, length, flags, timeout_ms); + + if (ret == 0) { + m_sock.close(); + + TCPSocket newsock; + m_sock = std::move(newsock); + reconnect(); + } + + return ret; + } + catch (const TCPSocket::Interrupted&) { + return -1; + } + catch (const TCPSocket::Timeout&) { + return 0; + } + + return 0; +} + +void TCPClient::reconnect() +{ + int flags = fcntl(m_sock.m_sock, F_GETFL); + if (fcntl(m_sock.m_sock, F_SETFL, flags | O_NONBLOCK) == -1) { + std::string errstr(strerror(errno)); + throw std::runtime_error("TCP: Could not set O_NONBLOCK: " + errstr); + } + + m_sock.connect(m_hostname, m_port); +} + +TCPConnection::TCPConnection(TCPSocket&& sock) : + queue(), + m_running(true), + m_sender_thread(), + m_sock(move(sock)) +{ +#if MISSING_OWN_ADDR + auto own_addr = m_sock.getOwnAddress(); + auto addr = m_sock.getRemoteAddress(); + etiLog.level(debug) << "New TCP Connection on port " << + own_addr.getPort() << " from " << + addr.getHostAddress() << ":" << addr.getPort(); +#endif + m_sender_thread = std::thread(&TCPConnection::process, this); +} + +TCPConnection::~TCPConnection() +{ + m_running = false; + vector<uint8_t> termination_marker; + queue.push(termination_marker); + m_sender_thread.join(); +} + +void TCPConnection::process() +{ + while (m_running) { + vector<uint8_t> data; + queue.wait_and_pop(data); + + if (data.empty()) { + // empty vector is the termination marker + m_running = false; + break; + } + + try { + ssize_t remaining = data.size(); + const uint8_t *buf = reinterpret_cast<const uint8_t*>(data.data()); + const int timeout_ms = 10; // Less than one ETI frame + + while (m_running and remaining > 0) { + const ssize_t sent = m_sock.send(buf, remaining, timeout_ms); + if (sent < 0 or sent > remaining) { + throw std::logic_error("Invalid TCPSocket::send() return value"); + } + remaining -= sent; + buf += sent; + } + } + catch (const std::runtime_error& e) { + m_running = false; + } + } + +#if MISSING_OWN_ADDR + auto own_addr = m_sock.getOwnAddress(); + auto addr = m_sock.getRemoteAddress(); + etiLog.level(debug) << "Dropping TCP Connection on port " << + own_addr.getPort() << " from " << + addr.getHostAddress() << ":" << addr.getPort(); +#endif +} + + +TCPDataDispatcher::TCPDataDispatcher(size_t max_queue_size) : + m_max_queue_size(max_queue_size) +{ +} + +TCPDataDispatcher::~TCPDataDispatcher() +{ + m_running = false; + m_connections.clear(); + m_listener_socket.close(); + if (m_listener_thread.joinable()) { + m_listener_thread.join(); + } +} + +void TCPDataDispatcher::start(int port, const string& address) +{ + m_listener_socket.listen(port, address); + + m_running = true; + m_listener_thread = std::thread(&TCPDataDispatcher::process, this); +} + +void TCPDataDispatcher::write(const vector<uint8_t>& data) +{ + if (not m_running) { + throw runtime_error(m_exception_data); + } + + for (auto& connection : m_connections) { + connection.queue.push(data); + } + + m_connections.remove_if( + [&](const TCPConnection& conn){ return conn.queue.size() > m_max_queue_size; }); +} + +void TCPDataDispatcher::process() +{ + try { + const int timeout_ms = 1000; + + while (m_running) { + // Add a new TCPConnection to the list, constructing it from the client socket + auto sock = m_listener_socket.accept(timeout_ms); + if (sock.valid()) { + m_connections.emplace(m_connections.begin(), move(sock)); + } + } + } + catch (const std::runtime_error& e) { + m_exception_data = string("TCPDataDispatcher error: ") + e.what(); + m_running = false; + } +} + +TCPReceiveServer::TCPReceiveServer(size_t blocksize) : + m_blocksize(blocksize) +{ +} + +void TCPReceiveServer::start(int listen_port, const std::string& address) +{ + m_listener_socket.listen(listen_port, address); + + m_running = true; + m_listener_thread = std::thread(&TCPReceiveServer::process, this); +} + +TCPReceiveServer::~TCPReceiveServer() +{ + m_running = false; + if (m_listener_thread.joinable()) { + m_listener_thread.join(); + } +} + +vector<uint8_t> TCPReceiveServer::receive() +{ + vector<uint8_t> buffer; + m_queue.try_pop(buffer); + + // we can ignore try_pop()'s return value, because + // if it is unsuccessful the buffer is not touched. + return buffer; +} + +void TCPReceiveServer::process() +{ + constexpr int timeout_ms = 1000; + constexpr int disconnect_timeout_ms = 10000; + constexpr int max_num_timeouts = disconnect_timeout_ms / timeout_ms; + + while (m_running) { + auto sock = m_listener_socket.accept(timeout_ms); + + int num_timeouts = 0; + + while (m_running and sock.valid()) { + try { + vector<uint8_t> buf(m_blocksize); + ssize_t r = sock.recv(buf.data(), buf.size(), 0, timeout_ms); + if (r < 0) { + throw logic_error("Invalid recv return value"); + } + else { + buf.resize(r); + m_queue.push(move(buf)); + } + } + catch (const TCPSocket::Interrupted&) { + break; + } + catch (const TCPSocket::Timeout&) { + num_timeouts++; + } + + if (num_timeouts > max_num_timeouts) { + sock.close(); + } + } + } +} + +} diff --git a/contrib/Socket.h b/contrib/Socket.h new file mode 100644 index 0000000..8bb7fe1 --- /dev/null +++ b/contrib/Socket.h @@ -0,0 +1,294 @@ +/* + Copyright (C) 2003, 2004, 2005, 2006, 2007, 2008, 2009 Her Majesty the + Queen in Right of Canada (Communications Research Center Canada) + + Copyright (C) 2019 + Matthias P. Braendli, matthias.braendli@mpb.li + + http://www.opendigitalradio.org + */ +/* + This program 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. + + This program 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 this program. If not, see <https://www.gnu.org/licenses/>. +*/ + +#pragma once + +#ifdef HAVE_CONFIG_H +# include "config.h" +#endif + +#include "ThreadsafeQueue.h" +#include <cstdlib> +#include <iostream> +#include <vector> +#include <atomic> +#include <thread> +#include <list> + +#include <sys/socket.h> +#include <netinet/in.h> +#include <unistd.h> +#include <netdb.h> +#include <arpa/inet.h> +#include <pthread.h> +#define SOCKET int +#define INVALID_SOCKET -1 +#define SOCKET_ERROR -1 + + +namespace Socket { + +struct InetAddress { + struct sockaddr_storage addr; + + struct sockaddr *as_sockaddr() { return reinterpret_cast<sockaddr*>(&addr); }; + + void resolveUdpDestination(const std::string& destination, int port); +}; + +/** This class represents a UDP packet. + * + * A UDP packet contains a payload (sequence of bytes) and an address. For + * outgoing packets, the address is the destination address. For incoming + * packets, the address tells the user from what source the packet arrived from. + */ +class UDPPacket +{ + public: + UDPPacket(); + UDPPacket(size_t initSize); + + std::vector<uint8_t> buffer; + InetAddress address; +}; + +/** + * This class represents a socket for sending and receiving UDP packets. + * + * A UDP socket is the sending or receiving point for a packet delivery service. + * Each packet sent or received on a datagram socket is individually + * addressed and routed. Multiple packets sent from one machine to another may + * be routed differently, and may arrive in any order. + */ +class UDPSocket +{ + public: + /** Create a new socket that will not be bound to any port. To be used + * for data output. + */ + UDPSocket(); + /** Create a new socket. + * @param port The port number on which the socket will be bound + */ + UDPSocket(int port); + /** Create a new socket. + * @param port The port number on which the socket will be bound + * @param name The IP address on which the socket will be bound. + * It is used to bind the socket on a specific interface if + * the computer have many NICs. + */ + UDPSocket(int port, const std::string& name); + ~UDPSocket(); + UDPSocket(const UDPSocket& other) = delete; + const UDPSocket& operator=(const UDPSocket& other) = delete; + + /** Close the already open socket, and create a new one. Throws a runtime_error on error. */ + void reinit(int port); + void reinit(int port, const std::string& name); + + void close(void); + void send(UDPPacket& packet); + void send(const std::vector<uint8_t>& data, InetAddress destination); + UDPPacket receive(size_t max_size); + void joinGroup(const char* groupname, const char* if_addr = nullptr); + void setMulticastSource(const char* source_addr); + void setMulticastTTL(int ttl); + + /** Set blocking mode. By default, the socket is blocking. + * throws a runtime_error on error. + */ + void setBlocking(bool block); + + protected: + SOCKET m_sock; +}; + +/* Threaded UDP receiver */ +class UDPReceiver { + public: + UDPReceiver(); + ~UDPReceiver(); + UDPReceiver(const UDPReceiver&) = delete; + UDPReceiver operator=(const UDPReceiver&) = delete; + + // Start the receiver in a separate thread + void start(int port, const std::string& bindto, const std::string& mcastaddr, size_t max_packets_queued); + + // Get the data contained in a UDP packet, blocks if none available + // In case of error, throws a runtime_error + std::vector<uint8_t> get_packet_buffer(void); + + private: + void m_run(void); + + int m_port = 0; + std::string m_bindto; + std::string m_mcastaddr; + size_t m_max_packets_queued = 1; + std::thread m_thread; + std::atomic<bool> m_stop = ATOMIC_VAR_INIT(false); + ThreadsafeQueue<UDPPacket> m_packets; + UDPSocket m_sock; +}; + +class TCPSocket { + public: + TCPSocket(); + ~TCPSocket(); + TCPSocket(const TCPSocket& other) = delete; + TCPSocket& operator=(const TCPSocket& other) = delete; + TCPSocket(TCPSocket&& other); + TCPSocket& operator=(TCPSocket&& other); + + bool valid(void) const; + void connect(const std::string& hostname, int port); + void listen(int port, const std::string& name); + void close(void); + + /* throws a runtime_error on failure, an invalid socket on timeout */ + TCPSocket accept(int timeout_ms); + + /* returns -1 on error, doesn't work on nonblocking sockets */ + ssize_t sendall(const void *buffer, size_t buflen); + + /** Send data over the TCP connection. + * @param data The buffer that will be sent. + * @param size Number of bytes to send. + * @param timeout_ms number of milliseconds before timeout, or 0 for infinite timeout + * return number of bytes sent, 0 on timeout, or throws runtime_error. + */ + ssize_t send(const void* data, size_t size, int timeout_ms=0); + + /* Returns number of bytes read, 0 on disconnect. Throws a + * runtime_error on error */ + ssize_t recv(void *buffer, size_t length, int flags); + + class Timeout {}; + class Interrupted {}; + /* Returns number of bytes read, 0 on disconnect or refused connection. + * Throws a Timeout on timeout, Interrupted on EINTR, a runtime_error + * on error + */ + ssize_t recv(void *buffer, size_t length, int flags, int timeout_ms); + + private: + explicit TCPSocket(int sockfd); + explicit TCPSocket(int sockfd, InetAddress remote_address); + SOCKET m_sock = -1; + + InetAddress m_remote_address; + + friend class TCPClient; +}; + +/* Implements a TCP receiver that auto-reconnects on errors */ +class TCPClient { + public: + void connect(const std::string& hostname, int port); + + /* Returns numer of bytes read, 0 on auto-reconnect, -1 + * on interruption. + * Throws a runtime_error on error */ + ssize_t recv(void *buffer, size_t length, int flags, int timeout_ms); + + private: + void reconnect(void); + TCPSocket m_sock; + std::string m_hostname; + int m_port; +}; + +/* Helper class for TCPDataDispatcher, contains a queue of pending data and + * a sender thread. */ +class TCPConnection +{ + public: + TCPConnection(TCPSocket&& sock); + TCPConnection(const TCPConnection&) = delete; + TCPConnection& operator=(const TCPConnection&) = delete; + ~TCPConnection(); + + ThreadsafeQueue<std::vector<uint8_t> > queue; + + private: + std::atomic<bool> m_running; + std::thread m_sender_thread; + TCPSocket m_sock; + + void process(void); +}; + +/* Send a TCP stream to several destinations, and automatically disconnect destinations + * whose buffer overflows. + */ +class TCPDataDispatcher +{ + public: + TCPDataDispatcher(size_t max_queue_size); + ~TCPDataDispatcher(); + TCPDataDispatcher(const TCPDataDispatcher&) = delete; + TCPDataDispatcher& operator=(const TCPDataDispatcher&) = delete; + + void start(int port, const std::string& address); + void write(const std::vector<uint8_t>& data); + + private: + void process(); + + size_t m_max_queue_size; + + std::atomic<bool> m_running; + std::string m_exception_data; + std::thread m_listener_thread; + TCPSocket m_listener_socket; + std::list<TCPConnection> m_connections; +}; + +/* A TCP Server to receive data, which abstracts the handling of connects and disconnects. + */ +class TCPReceiveServer { + public: + TCPReceiveServer(size_t blocksize); + ~TCPReceiveServer(); + TCPReceiveServer(const TCPReceiveServer&) = delete; + TCPReceiveServer& operator=(const TCPReceiveServer&) = delete; + + void start(int listen_port, const std::string& address); + + // Return a vector that contains up to blocksize bytes of data, or + // and empty vector if no data is available. + std::vector<uint8_t> receive(); + + private: + void process(); + + size_t m_blocksize = 0; + ThreadsafeQueue<std::vector<uint8_t> > m_queue; + std::atomic<bool> m_running; + std::string m_exception_data; + std::thread m_listener_thread; + TCPSocket m_listener_socket; +}; + +} diff --git a/contrib/ThreadsafeQueue.h b/contrib/ThreadsafeQueue.h new file mode 100644 index 0000000..62f4c96 --- /dev/null +++ b/contrib/ThreadsafeQueue.h @@ -0,0 +1,176 @@ +/* + Copyright (C) 2007, 2008, 2009, 2010, 2011 Her Majesty the Queen in + Right of Canada (Communications Research Center Canada) + + Copyright (C) 2018 + Matthias P. Braendli, matthias.braendli@mpb.li + + An implementation for a threadsafe queue, depends on C++11 + + When creating a ThreadsafeQueue, one can specify the minimal number + of elements it must contain before it is possible to take one + element out. + */ +/* + This program 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. + + This program 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 this program. If not, see <https://www.gnu.org/licenses/>. + */ + +#pragma once + +#include <mutex> +#include <condition_variable> +#include <queue> +#include <utility> + +/* This queue is meant to be used by two threads. One producer + * that pushes elements into the queue, and one consumer that + * retrieves the elements. + * + * The queue can make the consumer block until an element + * is available, or a wakeup requested. + */ + +/* Class thrown by blocking pop to tell the consumer + * that there's a wakeup requested. */ +class ThreadsafeQueueWakeup {}; + +template<typename T> +class ThreadsafeQueue +{ +public: + /* Push one element into the queue, and notify another thread that + * might be waiting. + * + * returns the new queue size. + */ + size_t push(T const& val) + { + std::unique_lock<std::mutex> lock(the_mutex); + the_queue.push(val); + size_t queue_size = the_queue.size(); + lock.unlock(); + + the_rx_notification.notify_one(); + + return queue_size; + } + + size_t push(T&& val) + { + std::unique_lock<std::mutex> lock(the_mutex); + the_queue.emplace(std::move(val)); + size_t queue_size = the_queue.size(); + lock.unlock(); + + the_rx_notification.notify_one(); + + return queue_size; + } + + /* Push one element into the queue, but wait until the + * queue size goes below the threshold. + * + * Notify waiting thread. + * + * returns the new queue size. + */ + size_t push_wait_if_full(T const& val, size_t threshold) + { + std::unique_lock<std::mutex> lock(the_mutex); + while (the_queue.size() >= threshold) { + the_tx_notification.wait(lock); + } + the_queue.push(val); + size_t queue_size = the_queue.size(); + lock.unlock(); + + the_rx_notification.notify_one(); + + return queue_size; + } + + /* Trigger a wakeup event on a blocking consumer, which + * will receive a ThreadsafeQueueWakeup exception. + */ + void trigger_wakeup(void) + { + std::unique_lock<std::mutex> lock(the_mutex); + wakeup_requested = true; + lock.unlock(); + the_rx_notification.notify_one(); + } + + /* Send a notification for the receiver thread */ + void notify(void) + { + the_rx_notification.notify_one(); + } + + bool empty() const + { + std::unique_lock<std::mutex> lock(the_mutex); + return the_queue.empty(); + } + + size_t size() const + { + std::unique_lock<std::mutex> lock(the_mutex); + return the_queue.size(); + } + + bool try_pop(T& popped_value) + { + std::unique_lock<std::mutex> lock(the_mutex); + if (the_queue.empty()) { + return false; + } + + popped_value = the_queue.front(); + the_queue.pop(); + + lock.unlock(); + the_tx_notification.notify_one(); + + return true; + } + + void wait_and_pop(T& popped_value, size_t prebuffering = 1) + { + std::unique_lock<std::mutex> lock(the_mutex); + while (the_queue.size() < prebuffering and + not wakeup_requested) { + the_rx_notification.wait(lock); + } + + if (wakeup_requested) { + wakeup_requested = false; + throw ThreadsafeQueueWakeup(); + } + else { + std::swap(popped_value, the_queue.front()); + the_queue.pop(); + + lock.unlock(); + the_tx_notification.notify_one(); + } + } + +private: + std::queue<T> the_queue; + mutable std::mutex the_mutex; + std::condition_variable the_rx_notification; + std::condition_variable the_tx_notification; + bool wakeup_requested = false; +}; + diff --git a/contrib/edi/AFPacket.cpp b/contrib/edi/AFPacket.cpp new file mode 100644 index 0000000..a58a980 --- /dev/null +++ b/contrib/edi/AFPacket.cpp @@ -0,0 +1,96 @@ +/* + Copyright (C) 2014 + Matthias P. Braendli, matthias.braendli@mpb.li + + http://www.opendigitalradio.org + + EDI output. + This implements an AF Packet as defined ETSI TS 102 821. + Also see ETSI TS 102 693 + + */ +/* + 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 <http://www.gnu.org/licenses/>. + */ +#include "config.h" +#include "crc.h" +#include "AFPacket.h" +#include "TagItems.h" +#include "TagPacket.h" +#include <vector> +#include <string> +#include <iostream> +#include <cstdio> +#include <stdint.h> +#include <arpa/inet.h> + +namespace edi { + +// Header PT field. AF packet contains TAG payload +const uint8_t AFHEADER_PT_TAG = 'T'; + +// AF Packet Major (3 bits) and Minor (4 bits) version +const uint8_t AFHEADER_VERSION = 0x10; // MAJ=1, MIN=0 + +AFPacket AFPacketiser::Assemble(TagPacket tag_packet) +{ + std::vector<uint8_t> payload = tag_packet.Assemble(); + + if (m_verbose) + std::cerr << "Assemble AFPacket " << seq << std::endl; + + std::string pack_data("AF"); // SYNC + std::vector<uint8_t> packet(pack_data.begin(), pack_data.end()); + + uint32_t taglength = payload.size(); + + if (m_verbose) + std::cerr << " AFPacket payload size " << payload.size() << std::endl; + + // write length into packet + packet.push_back((taglength >> 24) & 0xFF); + packet.push_back((taglength >> 16) & 0xFF); + packet.push_back((taglength >> 8) & 0xFF); + packet.push_back(taglength & 0xFF); + + // fill rest of header + packet.push_back(seq >> 8); + packet.push_back(seq & 0xFF); + seq++; + packet.push_back((have_crc ? 0x80 : 0) | AFHEADER_VERSION); // ar_cf: CRC=1 + packet.push_back(AFHEADER_PT_TAG); + + // insert payload, must have a length multiple of 8 bytes + packet.insert(packet.end(), payload.begin(), payload.end()); + + // calculate CRC over AF Header and payload + uint16_t crc = 0xffff; + crc = crc16(crc, &(packet.front()), packet.size()); + crc ^= 0xffff; + + if (m_verbose) + fprintf(stderr, " AFPacket crc %x\n", crc); + + packet.push_back((crc >> 8) & 0xFF); + packet.push_back(crc & 0xFF); + + if (m_verbose) + std::cerr << " AFPacket length " << packet.size() << std::endl; + + return packet; +} + +} diff --git a/contrib/edi/AFPacket.h b/contrib/edi/AFPacket.h new file mode 100644 index 0000000..b4ccef1 --- /dev/null +++ b/contrib/edi/AFPacket.h @@ -0,0 +1,61 @@ +/* + Copyright (C) 2014 + Matthias P. Braendli, matthias.braendli@mpb.li + + http://www.opendigitalradio.org + + EDI output. + This implements an AF Packet as defined ETSI TS 102 821. + Also see ETSI TS 102 693 + + */ +/* + 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 <http://www.gnu.org/licenses/>. + */ + +#pragma once + +#include "config.h" +#include <vector> +#include <stdint.h> +#include "TagItems.h" +#include "TagPacket.h" + +namespace edi { + +typedef std::vector<uint8_t> AFPacket; + +// ETSI TS 102 821, 6.1 AF packet structure +class AFPacketiser +{ + public: + AFPacketiser() : + m_verbose(false) {}; + AFPacketiser(bool verbose) : + m_verbose(verbose) {}; + + AFPacket Assemble(TagPacket tag_packet); + + private: + static const bool have_crc = true; + + uint16_t seq = 0; //counter that overflows at 0xFFFF + + bool m_verbose; +}; + +} + diff --git a/contrib/edi/Config.h b/contrib/edi/Config.h new file mode 100644 index 0000000..0c7dce8 --- /dev/null +++ b/contrib/edi/Config.h @@ -0,0 +1,84 @@ +/* + Copyright (C) 2019 + Matthias P. Braendli, matthias.braendli@mpb.li + + http://www.opendigitalradio.org + + EDI output, + UDP and TCP transports and their configuration + + */ +/* + 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 <http://www.gnu.org/licenses/>. + */ + +#pragma once + +#include "config.h" +#include <vector> +#include <string> +#include <memory> +#include <cstdint> + +namespace edi { + +/** Configuration for EDI output */ + +struct destination_t { + virtual ~destination_t() {}; +}; + +// Can represent both unicast and multicast destinations +struct udp_destination_t : public destination_t { + std::string dest_addr; + std::string source_addr; + unsigned int source_port = 0; + unsigned int ttl = 10; +}; + +// TCP server that can accept multiple connections +struct tcp_server_t : public destination_t { + unsigned int listen_port = 0; + size_t max_frames_queued = 1024; +}; + +// TCP client that connects to one endpoint +struct tcp_client_t : public destination_t { + std::string dest_addr; + unsigned int dest_port = 0; + size_t max_frames_queued = 1024; +}; + +struct configuration_t { + unsigned chunk_len = 207; // RSk, data length of each chunk + unsigned fec = 0; // number of fragments that can be recovered + bool dump = false; // dump a file with the EDI packets + bool verbose = false; + bool enable_pft = false; // Enable protection and fragmentation + unsigned int tagpacket_alignment = 0; + std::vector<std::shared_ptr<destination_t> > destinations; + unsigned int dest_port = 0; // common destination port, because it's encoded in the transport layer + unsigned int latency_frames = 0; // if nonzero, enable interleaver with a latency of latency_frames * 24ms + + bool enabled() const { return destinations.size() > 0; } + bool interleaver_enabled() const { return latency_frames > 0; } + + void print() const; +}; + +} + + diff --git a/contrib/edi/Interleaver.cpp b/contrib/edi/Interleaver.cpp new file mode 100644 index 0000000..50c5be2 --- /dev/null +++ b/contrib/edi/Interleaver.cpp @@ -0,0 +1,122 @@ +/* + Copyright (C) 2017 + Matthias P. Braendli, matthias.braendli@mpb.li + + http://www.opendigitalradio.org + + EDI output, + Interleaving of PFT fragments to increase robustness against + burst packet loss. + + This is possible because EDI has to assume that fragments may reach + the receiver out of order. + + */ +/* + 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 <http://www.gnu.org/licenses/>. + */ + +#include "edi/Interleaver.h" +#include <cassert> + +namespace edi { + +void Interleaver::SetLatency(size_t latency_frames) +{ + m_latency = latency_frames; +} + +Interleaver::fragment_vec Interleaver::Interleave(fragment_vec &fragments) +{ + m_fragment_count = fragments.size(); + + // Create vectors containing Fcount*latency fragments in total + // and store them into the deque + if (m_buffer.empty()) { + m_buffer.emplace_back(); + } + + auto& last_buffer = m_buffer.back(); + + for (auto& fragment : fragments) { + const bool last_buffer_is_complete = + (last_buffer.size() >= m_fragment_count * m_latency); + + if (last_buffer_is_complete) { + m_buffer.emplace_back(); + last_buffer = m_buffer.back(); + } + + last_buffer.push_back(std::move(fragment)); + } + + fragments.clear(); + + while ( not m_buffer.empty() and + (m_buffer.front().size() >= m_fragment_count * m_latency)) { + + auto& first_buffer = m_buffer.front(); + + assert(first_buffer.size() == m_fragment_count * m_latency); + + /* Assume we have 5 fragments per AF frame, and latency of 3. + * This will give the following strides: + * 0 1 2 + * +-------+-------+---+ + * | 0 1 | 2 3 | 4 | + * | | +---+ | + * | 5 6 | 7 | 8 9 | + * | +---+ | | + * |10 |11 12 |13 14 | + * +---+-------+-------+ + * + * ix will be 0, 5, 10, 1, 6 in the first loop + */ + + for (size_t i = 0; i < m_fragment_count; i++) { + const size_t ix = m_interleave_offset + m_fragment_count * m_stride; + m_interleaved_fragments.push_back(first_buffer.at(ix)); + + m_stride += 1; + if (m_stride >= m_latency) { + m_interleave_offset++; + m_stride = 0; + } + } + + if (m_interleave_offset >= m_fragment_count) { + m_interleave_offset = 0; + m_stride = 0; + m_buffer.pop_front(); + } + } + + std::vector<PFTFragment> interleaved_frags; + + const size_t n = std::min(m_fragment_count, m_interleaved_fragments.size()); + std::move(m_interleaved_fragments.begin(), + m_interleaved_fragments.begin() + n, + std::back_inserter(interleaved_frags)); + m_interleaved_fragments.erase( + m_interleaved_fragments.begin(), + m_interleaved_fragments.begin() + n); + + return interleaved_frags; +} + +} + + diff --git a/contrib/edi/Interleaver.h b/contrib/edi/Interleaver.h new file mode 100644 index 0000000..23aebf8 --- /dev/null +++ b/contrib/edi/Interleaver.h @@ -0,0 +1,74 @@ +/* + Copyright (C) 2017 + Matthias P. Braendli, matthias.braendli@mpb.li + + http://www.opendigitalradio.org + + EDI output, + Interleaving of PFT fragments to increase robustness against + burst packet loss. + + This is possible because EDI has to assume that fragments may reach + the receiver out of order. + + */ +/* + 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 <http://www.gnu.org/licenses/>. + */ + +#pragma once + +#include "config.h" +#include <vector> +#include <deque> +#include <stdexcept> +#include <stdint.h> +#include "edi/PFT.h" + +namespace edi { + +class Interleaver { + public: + using fragment_vec = std::vector<PFTFragment>; + + /* Configure the interleaver to use latency_frames number of AF + * packets for interleaving. Total delay through the interleaver + * will be latency_frames * 24ms + */ + void SetLatency(size_t latency_frames); + + /* Move the fragments for an AF Packet into the interleaver and + * return interleaved fragments to be transmitted. + */ + fragment_vec Interleave(fragment_vec &fragments); + + private: + size_t m_latency = 0; + size_t m_fragment_count = 0; + size_t m_interleave_offset = 0; + size_t m_stride = 0; + + /* Buffer that accumulates enough fragments to interleave */ + std::deque<fragment_vec> m_buffer; + + /* Buffer that contains fragments that have been interleaved, + * to avoid that the interleaver output is too bursty + */ + std::deque<PFTFragment> m_interleaved_fragments; +}; + +} + diff --git a/contrib/edi/PFT.cpp b/contrib/edi/PFT.cpp new file mode 100644 index 0000000..0692914 --- /dev/null +++ b/contrib/edi/PFT.cpp @@ -0,0 +1,325 @@ +/* + Copyright (C) 2019 + Matthias P. Braendli, matthias.braendli@mpb.li + + http://www.opendigitalradio.org + + EDI output, + Protection, Fragmentation and Transport. (PFT) + + Are supported: + Reed-Solomon and Fragmentation + + This implements part of PFT as defined ETSI TS 102 821. + + */ +/* + 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 <http://www.gnu.org/licenses/>. + */ + +#include "config.h" +#include <vector> +#include <list> +#include <cstdio> +#include <cstring> +#include <stdint.h> +#include <arpa/inet.h> +#include <stdexcept> +#include <sstream> +#include <iostream> +#include "edi/PFT.h" +#include "crc.h" + +namespace edi { + +using namespace std; + +// An integer division that rounds up, i.e. ceil(a/b) +#define CEIL_DIV(a, b) (a % b == 0 ? a / b : a / b + 1) + +PFT::PFT() { } + +PFT::PFT(const configuration_t &conf) : + m_k(conf.chunk_len), + m_m(conf.fec), + m_dest_port(conf.dest_port), + m_pseq(0), + m_num_chunks(0), + m_verbose(conf.verbose) + { + if (m_k > 207) { + throw std::out_of_range("EDI PFT Chunk size too large."); + } + + if (m_m > 5) { + clog << + "EDI PFT: high number of recoverable fragments" + " may lead to large overhead" << endl; + // See TS 102 821, 7.2.1 Known values, list entry for 'm' + } + } + +RSBlock PFT::Protect(AFPacket af_packet) +{ + RSBlock rs_block; + + // number of chunks is ceil(afpacketsize / m_k) + // TS 102 821 7.2.2: c = ceil(l / k_max) + m_num_chunks = CEIL_DIV(af_packet.size(), m_k); + + if (m_verbose) { + fprintf(stderr, "Protect %zu chunks of size %zu\n", + m_num_chunks, af_packet.size()); + } + + // calculate size of chunk: + // TS 102 821 7.2.2: k = ceil(l / c) + // chunk_len does not include the 48 bytes of protection. + const size_t chunk_len = CEIL_DIV(af_packet.size(), m_num_chunks); + if (chunk_len > 207) { + std::stringstream ss; + ss << "Chunk length " << chunk_len << " too large (>207)"; + throw std::runtime_error(ss.str()); + } + + // The last RS chunk is zero padded + // TS 102 821 7.2.2: z = c*k - l + const size_t zero_pad = m_num_chunks * chunk_len - af_packet.size(); + + // Create the RS(k+p,k) encoder + const int firstRoot = 1; // Discovered by analysing EDI dump + const int gfPoly = 0x11d; + const bool reverse = false; + // The encoding has to be 255, 207 always, because the chunk has to + // be padded at the end, and not at the beginning as libfec would + // do + ReedSolomon rs_encoder(255, 207, reverse, gfPoly, firstRoot); + + // add zero padding to last chunk + for (size_t i = 0; i < zero_pad; i++) { + af_packet.push_back(0); + } + + if (m_verbose) { + fprintf(stderr, " add %zu zero padding\n", zero_pad); + } + + // Calculate RS for each chunk and assemble RS block + for (size_t i = 0; i < af_packet.size(); i+= chunk_len) { + vector<uint8_t> chunk(207); + vector<uint8_t> protection(PARITYBYTES); + + // copy chunk_len bytes into new chunk + memcpy(&chunk.front(), &af_packet[i], chunk_len); + + // calculate RS for chunk with padding + rs_encoder.encode(&chunk.front(), &protection.front(), 207); + + // Drop the padding + chunk.resize(chunk_len); + + // append new chunk and protection to the RS Packet + rs_block.insert(rs_block.end(), chunk.begin(), chunk.end()); + rs_block.insert(rs_block.end(), protection.begin(), protection.end()); + } + + return rs_block; +} + +vector< vector<uint8_t> > PFT::ProtectAndFragment(AFPacket af_packet) +{ + const bool enable_RS = (m_m > 0); + + if (enable_RS) { + RSBlock rs_block = Protect(af_packet); + +#if 0 + fprintf(stderr, " af_packet (%zu):", af_packet.size()); + for (size_t i = 0; i < af_packet.size(); i++) { + fprintf(stderr, "%02x ", af_packet[i]); + } + fprintf(stderr, "\n"); + + fprintf(stderr, " rs_block (%zu):", rs_block.size()); + for (size_t i = 0; i < rs_block.size(); i++) { + fprintf(stderr, "%02x ", rs_block[i]); + } + fprintf(stderr, "\n"); +#endif + + // TS 102 821 7.2.2: s_max = MIN(floor(c*p/(m+1)), MTU - h)) + const size_t max_payload_size = ( m_num_chunks * PARITYBYTES ) / (m_m + 1); + + // Calculate fragment count and size + // TS 102 821 7.2.2: ceil((l + c*p + z) / s_max) + // l + c*p + z = length of RS block + const size_t num_fragments = CEIL_DIV(rs_block.size(), max_payload_size); + + // TS 102 821 7.2.2: ceil((l + c*p + z) / f) + const size_t fragment_size = CEIL_DIV(rs_block.size(), num_fragments); + + if (m_verbose) + fprintf(stderr, " PnF fragment_size %zu, num frag %zu\n", + fragment_size, num_fragments); + + vector< vector<uint8_t> > fragments(num_fragments); + + for (size_t i = 0; i < num_fragments; i++) { + fragments[i].resize(fragment_size); + for (size_t j = 0; j < fragment_size; j++) { + const size_t ix = j*num_fragments + i; + if (ix < rs_block.size()) { + fragments[i][j] = rs_block[ix]; + } + else { + fragments[i][j] = 0; + } + } + } + + return fragments; + } + else { // No RS, only fragmentation + // TS 102 821 7.2.2: s_max = MTU - h + // Ethernet MTU is 1500, but maybe you are routing over a network which + // has some sort of packet encapsulation. Add some margin. + const size_t max_payload_size = 1400; + + // Calculate fragment count and size + // TS 102 821 7.2.2: ceil((l + c*p + z) / s_max) + // l + c*p + z = length of AF packet + const size_t num_fragments = CEIL_DIV(af_packet.size(), max_payload_size); + + // TS 102 821 7.2.2: ceil((l + c*p + z) / f) + const size_t fragment_size = CEIL_DIV(af_packet.size(), num_fragments); + vector< vector<uint8_t> > fragments(num_fragments); + + for (size_t i = 0; i < num_fragments; i++) { + fragments[i].reserve(fragment_size); + + for (size_t j = 0; j < fragment_size; j++) { + const size_t ix = i*fragment_size + j; + if (ix < af_packet.size()) { + fragments[i].push_back(af_packet.at(ix)); + } + else { + break; + } + } + } + + return fragments; + } +} + +std::vector< PFTFragment > PFT::Assemble(AFPacket af_packet) +{ + vector< vector<uint8_t> > fragments = ProtectAndFragment(af_packet); + vector< vector<uint8_t> > pft_fragments; // These contain PF headers + + const bool enable_RS = (m_m > 0); + const bool enable_transport = true; + + unsigned int findex = 0; + + unsigned fcount = fragments.size(); + + // calculate size of chunk: + // TS 102 821 7.2.2: k = ceil(l / c) + // chunk_len does not include the 48 bytes of protection. + const size_t chunk_len = enable_RS ? + CEIL_DIV(af_packet.size(), m_num_chunks) : 0; + + // The last RS chunk is zero padded + // TS 102 821 7.2.2: z = c*k - l + const size_t zero_pad = enable_RS ? + m_num_chunks * chunk_len - af_packet.size() : 0; + + for (const auto &fragment : fragments) { + // Psync + std::string psync("PF"); + std::vector<uint8_t> packet(psync.begin(), psync.end()); + + // Pseq + packet.push_back(m_pseq >> 8); + packet.push_back(m_pseq & 0xFF); + + // Findex + packet.push_back(findex >> 16); + packet.push_back(findex >> 8); + packet.push_back(findex & 0xFF); + findex++; + + // Fcount + packet.push_back(fcount >> 16); + packet.push_back(fcount >> 8); + packet.push_back(fcount & 0xFF); + + // RS (1 bit), transport (1 bit) and Plen (14 bits) + unsigned int plen = fragment.size(); + if (enable_RS) { + plen |= 0x8000; // Set FEC bit + } + + if (enable_transport) { + plen |= 0x4000; // Set ADDR bit + } + + packet.push_back(plen >> 8); + packet.push_back(plen & 0xFF); + + if (enable_RS) { + packet.push_back(chunk_len); // RSk + packet.push_back(zero_pad); // RSz + } + + if (enable_transport) { + // Source (16 bits) + uint16_t addr_source = 0; + packet.push_back(addr_source >> 8); + packet.push_back(addr_source & 0xFF); + + // Dest (16 bits) + packet.push_back(m_dest_port >> 8); + packet.push_back(m_dest_port & 0xFF); + } + + // calculate CRC over AF Header and payload + uint16_t crc = 0xffff; + crc = crc16(crc, &(packet.front()), packet.size()); + crc ^= 0xffff; + + packet.push_back((crc >> 8) & 0xFF); + packet.push_back(crc & 0xFF); + + // insert payload, must have a length multiple of 8 bytes + packet.insert(packet.end(), fragment.begin(), fragment.end()); + + pft_fragments.push_back(packet); + +#if 0 + fprintf(stderr, "* PFT pseq %d, findex %d, fcount %d, plen %d\n", + m_pseq, findex, fcount, plen & ~0xC000); +#endif + } + + m_pseq++; + + return pft_fragments; +} + +} + diff --git a/contrib/edi/PFT.h b/contrib/edi/PFT.h new file mode 100644 index 0000000..6d41781 --- /dev/null +++ b/contrib/edi/PFT.h @@ -0,0 +1,77 @@ +/* + Copyright (C) 2019 + Matthias P. Braendli, matthias.braendli@mpb.li + + http://www.opendigitalradio.org + + EDI output, + Protection, Fragmentation and Transport. (PFT) + + Are supported: + Reed-Solomon and Fragmentation + + This implements part of PFT as defined ETSI TS 102 821. + + */ +/* + 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 <http://www.gnu.org/licenses/>. + */ + +#pragma once + +#include "config.h" +#include <vector> +#include <list> +#include <stdexcept> +#include <stdint.h> +#include "edi/AFPacket.h" +#include "edi/ReedSolomon.h" +#include "edi/Config.h" + +namespace edi { + +typedef std::vector<uint8_t> RSBlock; +typedef std::vector<uint8_t> PFTFragment; + +class PFT +{ + public: + static constexpr int PARITYBYTES = 48; + + PFT(); + PFT(const configuration_t& conf); + + // return a list of PFT fragments with the correct + // PFT headers + std::vector< PFTFragment > Assemble(AFPacket af_packet); + + // Apply Reed-Solomon FEC to the AF Packet + RSBlock Protect(AFPacket af_packet); + + // Cut a RSBlock into several fragments that can be transmitted + std::vector< std::vector<uint8_t> > ProtectAndFragment(AFPacket af_packet); + + private: + unsigned int m_k = 207; // length of RS data word + unsigned int m_m = 3; // number of fragments that can be recovered if lost + unsigned int m_dest_port = 12000; // Destination port for transport header + uint16_t m_pseq = 0; + size_t m_num_chunks = 0; + bool m_verbose = 0; +}; + +} + diff --git a/contrib/edi/ReedSolomon.cpp b/contrib/edi/ReedSolomon.cpp new file mode 100644 index 0000000..38d8ea8 --- /dev/null +++ b/contrib/edi/ReedSolomon.cpp @@ -0,0 +1,116 @@ +/* + Copyright (C) 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 <http://www.gnu.org/licenses/>. + */ + +#include "ReedSolomon.h" +#include <vector> +#include <algorithm> +#include <stdexcept> +#include <sstream> +#include <stdio.h> // For galois.h ... +#include <string.h> // For memcpy + +extern "C" { +#include "fec/fec.h" +} +#include <assert.h> + +#define SYMSIZE 8 + + +ReedSolomon::ReedSolomon(int N, int K, bool reverse, int gfpoly, int firstRoot, int primElem) +{ + setReverse(reverse); + + m_N = N; + m_K = K; + + const int symsize = SYMSIZE; + const int nroots = N - K; // For EDI PFT, this must be 48 + const int pad = ((1 << symsize) - 1) - N; // is 255-N + + rsData = init_rs_char(symsize, gfpoly, firstRoot, primElem, nroots, pad); + + if (rsData == nullptr) { + std::stringstream ss; + ss << "Invalid Reed-Solomon parameters! " << + "N=" << N << " ; K=" << K << " ; pad=" << pad; + throw std::invalid_argument(ss.str()); + } +} + + +ReedSolomon::~ReedSolomon() +{ + free_rs_char(rsData); +} + + +void ReedSolomon::setReverse(bool state) +{ + reverse = state; +} + + +int ReedSolomon::encode(void* data, void* fec, size_t size) +{ + uint8_t* input = reinterpret_cast<uint8_t*>(data); + uint8_t* output = reinterpret_cast<uint8_t*>(fec); + int ret = 0; + + if (reverse) { + std::vector<uint8_t> buffer(m_N); + + memcpy(&buffer[0], input, m_K); + memcpy(&buffer[m_K], output, m_N - m_K); + + ret = decode_rs_char(rsData, &buffer[0], nullptr, 0); + if ((ret != 0) && (ret != -1)) { + memcpy(input, &buffer[0], m_K); + memcpy(output, &buffer[m_K], m_N - m_K); + } + } + else { + encode_rs_char(rsData, input, output); + } + + return ret; +} + + +int ReedSolomon::encode(void* data, size_t size) +{ + uint8_t* input = reinterpret_cast<uint8_t*>(data); + int ret = 0; + + if (reverse) { + ret = decode_rs_char(rsData, input, nullptr, 0); + } + else { + encode_rs_char(rsData, input, &input[m_K]); + } + + return ret; +} diff --git a/contrib/edi/ReedSolomon.h b/contrib/edi/ReedSolomon.h new file mode 100644 index 0000000..abcef62 --- /dev/null +++ b/contrib/edi/ReedSolomon.h @@ -0,0 +1,56 @@ +/* + Copyright (C) 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 <http://www.gnu.org/licenses/>. + */ + +#pragma once + +#ifdef HAVE_CONFIG_H +# include <config.h> +#endif + +#include <stdlib.h> + +class ReedSolomon +{ +public: + ReedSolomon(int N, int K, + bool reverse = false, + int gfpoly = 0x11d, int firstRoot = 0, int primElem = 1); + ReedSolomon(const ReedSolomon& other) = delete; + ReedSolomon operator=(const ReedSolomon& other) = delete; + ~ReedSolomon(); + + void setReverse(bool state); + int encode(void* data, void* fec, size_t size); + int encode(void* data, size_t size); + +private: + int m_N; + int m_K; + + void* rsData; + bool reverse; +}; + diff --git a/contrib/edi/TagItems.cpp b/contrib/edi/TagItems.cpp new file mode 100644 index 0000000..748f246 --- /dev/null +++ b/contrib/edi/TagItems.cpp @@ -0,0 +1,215 @@ +/* + EDI output. + This defines a few TAG items as defined ETSI TS 102 821 and + ETSI TS 102 693 + + Copyright (C) 2019 + 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 <http://www.gnu.org/licenses/>. + */ + +#include "config.h" +#include "edi/TagItems.h" +#include <vector> +#include <iostream> +#include <string> +#include <stdint.h> +#include <stdexcept> + +namespace edi { + +std::vector<uint8_t> TagStarPTR::Assemble() +{ + //std::cerr << "TagItem *ptr" << std::endl; + std::string pack_data("*ptr"); + std::vector<uint8_t> packet(pack_data.begin(), pack_data.end()); + + packet.push_back(0); + packet.push_back(0); + packet.push_back(0); + packet.push_back(0x40); + + if (protocol.size() != 4) { + throw std::runtime_error("TagStarPTR protocol invalid length"); + } + packet.insert(packet.end(), protocol.begin(), protocol.end()); + + // Major + packet.push_back(0); + packet.push_back(0); + + // Minor + packet.push_back(0); + packet.push_back(0); + return packet; +} + +std::vector<uint8_t> TagDSTI::Assemble() +{ + std::string pack_data("dsti"); + std::vector<uint8_t> packet(pack_data.begin(), pack_data.end()); + packet.reserve(256); + + // Placeholder for length + packet.push_back(0); + packet.push_back(0); + packet.push_back(0); + packet.push_back(0); + + uint8_t dfctl = dflc % 250; + uint8_t dfcth = dflc / 250; + + + uint16_t dstiHeader = dfctl | (dfcth << 8) | (rfadf << 13) | (atstf << 14) | (stihf << 15); + packet.push_back(dstiHeader >> 8); + packet.push_back(dstiHeader & 0xFF); + + if (stihf) { + packet.push_back(stat); + packet.push_back((spid >> 8) & 0xFF); + packet.push_back(spid & 0xFF); + } + + if (atstf) { + packet.push_back(utco); + + packet.push_back((seconds >> 24) & 0xFF); + packet.push_back((seconds >> 16) & 0xFF); + packet.push_back((seconds >> 8) & 0xFF); + packet.push_back(seconds & 0xFF); + + packet.push_back((tsta >> 16) & 0xFF); + packet.push_back((tsta >> 8) & 0xFF); + packet.push_back(tsta & 0xFF); + } + + if (rfadf) { + for (size_t i = 0; i < rfad.size(); i++) { + packet.push_back(rfad[i]); + } + } + // calculate and update size + // remove TAG name and TAG length fields and convert to bits + uint32_t taglength = (packet.size() - 8) * 8; + + // write length into packet + packet[4] = (taglength >> 24) & 0xFF; + packet[5] = (taglength >> 16) & 0xFF; + packet[6] = (taglength >> 8) & 0xFF; + packet[7] = taglength & 0xFF; + + dflc = (dflc+1) % 5000; + + /* + std::cerr << "TagItem dsti, packet.size " << packet.size() << std::endl; + std::cerr << " length " << taglength / 8 << std::endl; + */ + return packet; +} + +void TagDSTI::set_edi_time(const std::time_t t, int tai_utc_offset) +{ + utco = tai_utc_offset - 32; + + const std::time_t posix_timestamp_1_jan_2000 = 946684800; + + seconds = t - posix_timestamp_1_jan_2000 + utco; +} + + +std::vector<uint8_t> TagSSm::Assemble() +{ + std::string pack_data("ss"); + std::vector<uint8_t> packet(pack_data.begin(), pack_data.end()); + packet.reserve(istd_length + 16); + + packet.push_back((id >> 8) & 0xFF); + packet.push_back(id & 0xFF); + + // Placeholder for length + packet.push_back(0); + packet.push_back(0); + packet.push_back(0); + packet.push_back(0); + + if (rfa > 0x1F) { + throw std::runtime_error("TagSSm: invalid RFA value"); + } + + if (tid > 0x7) { + throw std::runtime_error("TagSSm: invalid tid value"); + } + + if (tidext > 0x7) { + throw std::runtime_error("TagSSm: invalid tidext value"); + } + + if (stid > 0x0FFF) { + throw std::runtime_error("TagSSm: invalid stid value"); + } + + uint32_t istc = (rfa << 19) | (tid << 16) | (tidext << 13) | ((crcstf ? 1 : 0) << 12) | stid; + packet.push_back((istc >> 16) & 0xFF); + packet.push_back((istc >> 8) & 0xFF); + packet.push_back(istc & 0xFF); + + for (size_t i = 0; i < istd_length; i++) { + packet.push_back(istd_data[i]); + } + + // calculate and update size + // remove TAG name and TAG length fields and convert to bits + uint32_t taglength = (packet.size() - 8) * 8; + + // write length into packet + packet[4] = (taglength >> 24) & 0xFF; + packet[5] = (taglength >> 16) & 0xFF; + packet[6] = (taglength >> 8) & 0xFF; + packet[7] = taglength & 0xFF; + + /* + std::cerr << "TagItem SSm, length " << packet.size() << std::endl; + std::cerr << " istd_length " << istd_length << std::endl; + */ + return packet; +} + +std::vector<uint8_t> TagStarDMY::Assemble() +{ + std::string pack_data("*dmy"); + std::vector<uint8_t> packet(pack_data.begin(), pack_data.end()); + + packet.resize(4 + 4 + length_); + + const uint32_t length_bits = length_ * 8; + + packet[4] = (length_bits >> 24) & 0xFF; + packet[5] = (length_bits >> 16) & 0xFF; + packet[6] = (length_bits >> 8) & 0xFF; + packet[7] = length_bits & 0xFF; + + // The remaining bytes in the packet are "undefined data" + + return packet; +} + +} + diff --git a/contrib/edi/TagItems.h b/contrib/edi/TagItems.h new file mode 100644 index 0000000..73e745a --- /dev/null +++ b/contrib/edi/TagItems.h @@ -0,0 +1,139 @@ +/* + EDI output. + This defines a few TAG items as defined ETSI TS 102 821 and + ETSI TS 102 693 + + Copyright (C) 2019 + 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 <http://www.gnu.org/licenses/>. + */ + +#pragma once + +#include "config.h" +#include <vector> +#include <array> +#include <chrono> +#include <string> +#include <stdint.h> + +namespace edi { + +class TagItem +{ + public: + virtual std::vector<uint8_t> Assemble() = 0; +}; + +// ETSI TS 102 693, 5.1.1 Protocol type and revision +class TagStarPTR : public TagItem +{ + public: + std::string protocol = ""; + std::vector<uint8_t> Assemble(); +}; + +// ETSI TS 102 693, 5.1.2 DAB STI-D(LI) Management +class TagDSTI : public TagItem +{ + public: + std::vector<uint8_t> Assemble(); + + // dsti Header + bool stihf = false; + bool atstf = false; // presence of atst data + bool rfadf = false; + uint16_t dflc = 0; // modulo 5000 frame counter + + // STI Header (optional) + uint8_t stat = 0; + uint16_t spid = 0; + + /* UTCO: Offset (in seconds) between UTC and the Seconds value. The + * value is expressed as an unsigned 8-bit quantity. As of February + * 2009, the value shall be 2 and shall change as a result of each + * modification of the number of leap seconds, as proscribed by + * International Earth Rotation and Reference Systems Service (IERS). + * + * According to Annex F + * EDI = TAI - 32s (constant) + * EDI = UTC + UTCO + * we derive + * UTCO = TAI-UTC - 32 + * where the TAI-UTC offset is given by the USNO bulletin using + * the ClockTAI module. + */ + uint8_t utco = 0; + + /* Update the EDI time. t is in UTC */ + void set_edi_time(const std::time_t t, int tai_utc_offset); + + /* The number of SI seconds since 2000-01-01 T 00:00:00 UTC as an + * unsigned 32-bit quantity. Contrary to POSIX, this value also + * counts leap seconds. + */ + uint32_t seconds = 0; + + /* TSTA: Shall be the 24 least significant bits of the Time Stamp + * (TIST) field from the STI-D(LI) Frame. The full definition for the + * STI TIST can be found in annex B of EN 300 797 [4]. The most + * significant 8 bits of the TIST field of the incoming STI-D(LI) + * frame, if required, may be carried in the RFAD field. + */ + uint32_t tsta = 0xFFFFFF; + + std::array<uint8_t, 9> rfad; +}; + +// ETSI TS 102 693, 5.1.4 STI-D Payload Stream <m> +class TagSSm : public TagItem +{ + public: + std::vector<uint8_t> Assemble(); + + // SSTCn + uint8_t rfa = 0; + uint8_t tid = 0; // See EN 300 797, 5.4.1.1. Value 0 means "MSC sub-channel" + uint8_t tidext = 0; // EN 300 797, 5.4.1.3, Value 0 means "MSC audio stream" + bool crcstf = false; + uint16_t stid = 0; + + // Pointer to ISTDm data + const uint8_t *istd_data; + size_t istd_length; // bytes + + uint16_t id = 0; +}; + +// ETSI TS 102 821, 5.2.2.2 Dummy padding +class TagStarDMY : public TagItem +{ + public: + /* length is the TAG value length in bytes */ + TagStarDMY(uint32_t length) : length_(length) {} + std::vector<uint8_t> Assemble(); + + private: + uint32_t length_; +}; + +} + diff --git a/contrib/edi/TagPacket.cpp b/contrib/edi/TagPacket.cpp new file mode 100644 index 0000000..01a1ffe --- /dev/null +++ b/contrib/edi/TagPacket.cpp @@ -0,0 +1,75 @@ +/* + Copyright (C) 2014 + Matthias P. Braendli, matthias.braendli@mpb.li + + http://www.opendigitalradio.org + + EDI output. + This defines a TAG Packet. + */ +/* + 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 <http://www.gnu.org/licenses/>. + */ + +#include "config.h" +#include "edi/TagPacket.h" +#include "edi/TagItems.h" +#include <iostream> +#include <string> +#include <cassert> + +namespace edi { + +TagPacket::TagPacket(unsigned int alignment) : m_alignment(alignment) +{ } + +std::vector<uint8_t> TagPacket::Assemble() +{ + std::list<TagItem*>::iterator tag; + + std::vector<uint8_t> packet; + + //std::cerr << "Assemble TAGPacket" << std::endl; + + for (tag = tag_items.begin(); tag != tag_items.end(); ++tag) { + std::vector<uint8_t> tag_data = (*tag)->Assemble(); + packet.insert(packet.end(), tag_data.begin(), tag_data.end()); + + //std::cerr << " Add TAGItem of length " << tag_data.size() << std::endl; + } + + if (m_alignment == 0) { /* no padding */ } + else if (m_alignment == 8) { + // Add padding inside TAG packet + while (packet.size() % 8 > 0) { + packet.push_back(0); // TS 102 821, 5.1, "padding shall be undefined" + } + } + else if (m_alignment > 8) { + TagStarDMY dmy(m_alignment - 8); + auto dmy_data = dmy.Assemble(); + packet.insert(packet.end(), dmy_data.begin(), dmy_data.end()); + } + else { + std::cerr << "Invalid alignment requirement " << m_alignment << + " defined in TagPacket" << std::endl; + } + + return packet; +} + +} + diff --git a/contrib/edi/TagPacket.h b/contrib/edi/TagPacket.h new file mode 100644 index 0000000..a932e89 --- /dev/null +++ b/contrib/edi/TagPacket.h @@ -0,0 +1,56 @@ +/* + Copyright (C) 2014 + Matthias P. Braendli, matthias.braendli@mpb.li + + http://www.opendigitalradio.org + + EDI output. + This defines a TAG Packet. + */ +/* + 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 <http://www.gnu.org/licenses/>. + */ + +#pragma once + +#include "config.h" +#include "edi/TagItems.h" +#include <vector> +#include <string> +#include <list> +#include <cstdint> + +namespace edi { + +// A TagPacket is nothing else than a list of tag items, with an +// Assemble function that puts the bytestream together and adds +// padding such that the total length is a multiple of 8 Bytes. +// +// ETSI TS 102 821, 5.1 Tag Packet +class TagPacket +{ + public: + TagPacket(unsigned int alignment); + std::vector<uint8_t> Assemble(); + + std::list<TagItem*> tag_items; + + private: + unsigned int m_alignment; +}; + +} + diff --git a/contrib/edi/Transport.cpp b/contrib/edi/Transport.cpp new file mode 100644 index 0000000..c2fb2a7 --- /dev/null +++ b/contrib/edi/Transport.cpp @@ -0,0 +1,190 @@ +/* + Copyright (C) 2019 + Matthias P. Braendli, matthias.braendli@mpb.li + + http://www.opendigitalradio.org + + EDI output, + UDP and TCP transports and their configuration + + */ +/* + 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 <http://www.gnu.org/licenses/>. + */ + +#include "edi/Transport.h" +#include <iterator> +#include <iostream> + +using namespace std; + +namespace edi { + +void configuration_t::print() const +{ + clog << "EDI" << endl; + clog << " verbose " << verbose << endl; + for (auto edi_dest : destinations) { + if (auto udp_dest = dynamic_pointer_cast<edi::udp_destination_t>(edi_dest)) { + clog << " UDP to " << udp_dest->dest_addr << ":" << dest_port << endl; + if (not udp_dest->source_addr.empty()) { + clog << " source " << udp_dest->source_addr << endl; + clog << " ttl " << udp_dest->ttl << endl; + } + clog << " source port " << udp_dest->source_port << endl; + } + else if (auto tcp_dest = dynamic_pointer_cast<edi::tcp_server_t>(edi_dest)) { + clog << " TCP listening on port " << tcp_dest->listen_port << endl; + clog << " max frames queued " << tcp_dest->max_frames_queued << endl; + } + else if (auto tcp_dest = dynamic_pointer_cast<edi::tcp_client_t>(edi_dest)) { + clog << " TCP client connecting to " << tcp_dest->dest_addr << ":" << tcp_dest->dest_port << endl; + clog << " max frames queued " << tcp_dest->max_frames_queued << endl; + } + else { + throw logic_error("EDI destination not implemented"); + } + } + if (interleaver_enabled()) { + clog << " interleave " << latency_frames * 24 << " ms" << endl; + } +} + + +Sender::Sender(const configuration_t& conf) : + m_conf(conf), + edi_pft(m_conf) +{ + if (m_conf.verbose) { + clog << "Setup EDI" << endl; + } + + for (const auto& edi_dest : m_conf.destinations) { + if (const auto udp_dest = dynamic_pointer_cast<edi::udp_destination_t>(edi_dest)) { + auto udp_socket = std::make_shared<Socket::UDPSocket>(udp_dest->source_port); + + if (not udp_dest->source_addr.empty()) { + udp_socket->setMulticastSource(udp_dest->source_addr.c_str()); + udp_socket->setMulticastTTL(udp_dest->ttl); + } + + udp_sockets.emplace(udp_dest.get(), udp_socket); + } + else if (auto tcp_dest = dynamic_pointer_cast<edi::tcp_server_t>(edi_dest)) { + auto dispatcher = make_shared<Socket::TCPDataDispatcher>(tcp_dest->max_frames_queued); + dispatcher->start(tcp_dest->listen_port, "0.0.0.0"); + tcp_dispatchers.emplace(tcp_dest.get(), dispatcher); + } + else if (auto tcp_dest = dynamic_pointer_cast<edi::tcp_client_t>(edi_dest)) { + auto tcp_socket = make_shared<Socket::TCPSocket>(); + tcp_socket->connect(tcp_dest->dest_addr, tcp_dest->dest_port); + tcp_senders.emplace(tcp_dest.get(), tcp_socket); + } + else { + throw logic_error("EDI destination not implemented"); + } + } + + if (m_conf.interleaver_enabled()) { + edi_interleaver.SetLatency(m_conf.latency_frames); + } + + if (m_conf.dump) { + edi_debug_file.open("./edi.debug"); + } + + if (m_conf.verbose) { + clog << "EDI set up" << endl; + } +} + +void Sender::write(const TagPacket& tagpacket) +{ + // Assemble into one AF Packet + edi::AFPacket af_packet = edi_afPacketiser.Assemble(tagpacket); + + if (m_conf.enable_pft) { + // Apply PFT layer to AF Packet (Reed Solomon FEC and Fragmentation) + vector<edi::PFTFragment> edi_fragments = edi_pft.Assemble(af_packet); + + if (m_conf.verbose) { + fprintf(stderr, "EDI number of PFT fragment before interleaver %zu\n", + edi_fragments.size()); + } + + if (m_conf.interleaver_enabled()) { + edi_fragments = edi_interleaver.Interleave(edi_fragments); + } + + // Send over ethernet + for (const auto& edi_frag : edi_fragments) { + for (auto& dest : m_conf.destinations) { + if (const auto& udp_dest = dynamic_pointer_cast<edi::udp_destination_t>(dest)) { + Socket::InetAddress addr; + addr.resolveUdpDestination(udp_dest->dest_addr, m_conf.dest_port); + + udp_sockets.at(udp_dest.get())->send(edi_frag, addr); + } + else if (auto tcp_dest = dynamic_pointer_cast<edi::tcp_server_t>(dest)) { + tcp_dispatchers.at(tcp_dest.get())->write(edi_frag); + } + else if (auto tcp_dest = dynamic_pointer_cast<edi::tcp_client_t>(dest)) { + tcp_senders.at(tcp_dest.get())->sendall(edi_frag.data(), edi_frag.size()); + } + else { + throw logic_error("EDI destination not implemented"); + } + } + + if (m_conf.dump) { + ostream_iterator<uint8_t> debug_iterator(edi_debug_file); + copy(edi_frag.begin(), edi_frag.end(), debug_iterator); + } + } + + if (m_conf.verbose) { + fprintf(stderr, "EDI number of PFT fragments %zu\n", + edi_fragments.size()); + } + } + else { + // Send over ethernet + for (auto& dest : m_conf.destinations) { + if (const auto& udp_dest = dynamic_pointer_cast<edi::udp_destination_t>(dest)) { + Socket::InetAddress addr; + addr.resolveUdpDestination(udp_dest->dest_addr, m_conf.dest_port); + + udp_sockets.at(udp_dest.get())->send(af_packet, addr); + } + else if (auto tcp_dest = dynamic_pointer_cast<edi::tcp_server_t>(dest)) { + tcp_dispatchers.at(tcp_dest.get())->write(af_packet); + } + else if (auto tcp_dest = dynamic_pointer_cast<edi::tcp_client_t>(dest)) { + tcp_senders.at(tcp_dest.get())->sendall(af_packet.data(), af_packet.size()); + } + else { + throw logic_error("EDI destination not implemented"); + } + } + + if (m_conf.dump) { + ostream_iterator<uint8_t> debug_iterator(edi_debug_file); + copy(af_packet.begin(), af_packet.end(), debug_iterator); + } + } +} + +} diff --git a/contrib/edi/Transport.h b/contrib/edi/Transport.h new file mode 100644 index 0000000..db1adce --- /dev/null +++ b/contrib/edi/Transport.h @@ -0,0 +1,71 @@ +/* + Copyright (C) 2019 + Matthias P. Braendli, matthias.braendli@mpb.li + + http://www.opendigitalradio.org + + EDI output, + UDP and TCP transports and their configuration + + */ +/* + 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 <http://www.gnu.org/licenses/>. + */ + +#pragma once + +#include "config.h" +#include "edi/Config.h" +#include "edi/AFPacket.h" +#include "edi/PFT.h" +#include "edi/Interleaver.h" +#include "Socket.h" +#include <vector> +#include <unordered_map> +#include <stdexcept> +#include <fstream> +#include <cstdint> + +namespace edi { + +/** Configuration for EDI output */ + +class Sender { + public: + Sender(const configuration_t& conf); + + void write(const TagPacket& tagpacket); + + private: + configuration_t m_conf; + std::ofstream edi_debug_file; + + // The TagPacket will then be placed into an AFPacket + edi::AFPacketiser edi_afPacketiser; + + // The AF Packet will be protected with reed-solomon and split in fragments + edi::PFT edi_pft; + + // To mitigate for burst packet loss, PFT fragments can be sent out-of-order + edi::Interleaver edi_interleaver; + + std::unordered_map<udp_destination_t*, std::shared_ptr<Socket::UDPSocket>> udp_sockets; + std::unordered_map<tcp_server_t*, std::shared_ptr<Socket::TCPDataDispatcher>> tcp_dispatchers; + std::unordered_map<tcp_client_t*, std::shared_ptr<Socket::TCPSocket>> tcp_senders; +}; + +} + diff --git a/contrib/edi/crc.c b/contrib/edi/crc.c new file mode 100644 index 0000000..cc02473 --- /dev/null +++ b/contrib/edi/crc.c @@ -0,0 +1,266 @@ +/* + Copyright (C) 2003, 2004, 2005, 2006, 2007, 2008, 2009 Her Majesty the + Queen in Right of Canada (Communications Research Center Canada) + */ +/* + 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 <http://www.gnu.org/licenses/>. + */ + +#include "crc.h" +#ifndef _WIN32 +# include <unistd.h> +# include <netinet/in.h> +#endif +#include <stdio.h> +#include <fcntl.h> + +//#define CCITT 0x1021 + +uint8_t crc8tab[256] = { + 0x00, 0x07, 0x0e, 0x09, 0x1c, 0x1b, 0x12, 0x15, + 0x38, 0x3f, 0x36, 0x31, 0x24, 0x23, 0x2a, 0x2d, + 0x70, 0x77, 0x7e, 0x79, 0x6c, 0x6b, 0x62, 0x65, + 0x48, 0x4f, 0x46, 0x41, 0x54, 0x53, 0x5a, 0x5d, + 0xe0, 0xe7, 0xee, 0xe9, 0xfc, 0xfb, 0xf2, 0xf5, + 0xd8, 0xdf, 0xd6, 0xd1, 0xc4, 0xc3, 0xca, 0xcd, + 0x90, 0x97, 0x9e, 0x99, 0x8c, 0x8b, 0x82, 0x85, + 0xa8, 0xaf, 0xa6, 0xa1, 0xb4, 0xb3, 0xba, 0xbd, + 0xc7, 0xc0, 0xc9, 0xce, 0xdb, 0xdc, 0xd5, 0xd2, + 0xff, 0xf8, 0xf1, 0xf6, 0xe3, 0xe4, 0xed, 0xea, + 0xb7, 0xb0, 0xb9, 0xbe, 0xab, 0xac, 0xa5, 0xa2, + 0x8f, 0x88, 0x81, 0x86, 0x93, 0x94, 0x9d, 0x9a, + 0x27, 0x20, 0x29, 0x2e, 0x3b, 0x3c, 0x35, 0x32, + 0x1f, 0x18, 0x11, 0x16, 0x03, 0x04, 0x0d, 0x0a, + 0x57, 0x50, 0x59, 0x5e, 0x4b, 0x4c, 0x45, 0x42, + 0x6f, 0x68, 0x61, 0x66, 0x73, 0x74, 0x7d, 0x7a, + 0x89, 0x8e, 0x87, 0x80, 0x95, 0x92, 0x9b, 0x9c, + 0xb1, 0xb6, 0xbf, 0xb8, 0xad, 0xaa, 0xa3, 0xa4, + 0xf9, 0xfe, 0xf7, 0xf0, 0xe5, 0xe2, 0xeb, 0xec, + 0xc1, 0xc6, 0xcf, 0xc8, 0xdd, 0xda, 0xd3, 0xd4, + 0x69, 0x6e, 0x67, 0x60, 0x75, 0x72, 0x7b, 0x7c, + 0x51, 0x56, 0x5f, 0x58, 0x4d, 0x4a, 0x43, 0x44, + 0x19, 0x1e, 0x17, 0x10, 0x05, 0x02, 0x0b, 0x0c, + 0x21, 0x26, 0x2f, 0x28, 0x3d, 0x3a, 0x33, 0x34, + 0x4e, 0x49, 0x40, 0x47, 0x52, 0x55, 0x5c, 0x5b, + 0x76, 0x71, 0x78, 0x7f, 0x6a, 0x6d, 0x64, 0x63, + 0x3e, 0x39, 0x30, 0x37, 0x22, 0x25, 0x2c, 0x2b, + 0x06, 0x01, 0x08, 0x0f, 0x1a, 0x1d, 0x14, 0x13, + 0xae, 0xa9, 0xa0, 0xa7, 0xb2, 0xb5, 0xbc, 0xbb, + 0x96, 0x91, 0x98, 0x9f, 0x8a, 0x8d, 0x84, 0x83, + 0xde, 0xd9, 0xd0, 0xd7, 0xc2, 0xc5, 0xcc, 0xcb, + 0xe6, 0xe1, 0xe8, 0xef, 0xfa, 0xfd, 0xf4, 0xf3 +}; + + +uint16_t crc16tab[256] = { + 0x0000, 0x1021, 0x2042, 0x3063, 0x4084, 0x50a5, 0x60c6, 0x70e7, + 0x8108, 0x9129, 0xa14a, 0xb16b, 0xc18c, 0xd1ad, 0xe1ce, 0xf1ef, + 0x1231, 0x0210, 0x3273, 0x2252, 0x52b5, 0x4294, 0x72f7, 0x62d6, + 0x9339, 0x8318, 0xb37b, 0xa35a, 0xd3bd, 0xc39c, 0xf3ff, 0xe3de, + 0x2462, 0x3443, 0x0420, 0x1401, 0x64e6, 0x74c7, 0x44a4, 0x5485, + 0xa56a, 0xb54b, 0x8528, 0x9509, 0xe5ee, 0xf5cf, 0xc5ac, 0xd58d, + 0x3653, 0x2672, 0x1611, 0x0630, 0x76d7, 0x66f6, 0x5695, 0x46b4, + 0xb75b, 0xa77a, 0x9719, 0x8738, 0xf7df, 0xe7fe, 0xd79d, 0xc7bc, + 0x48c4, 0x58e5, 0x6886, 0x78a7, 0x0840, 0x1861, 0x2802, 0x3823, + 0xc9cc, 0xd9ed, 0xe98e, 0xf9af, 0x8948, 0x9969, 0xa90a, 0xb92b, + 0x5af5, 0x4ad4, 0x7ab7, 0x6a96, 0x1a71, 0x0a50, 0x3a33, 0x2a12, + 0xdbfd, 0xcbdc, 0xfbbf, 0xeb9e, 0x9b79, 0x8b58, 0xbb3b, 0xab1a, + 0x6ca6, 0x7c87, 0x4ce4, 0x5cc5, 0x2c22, 0x3c03, 0x0c60, 0x1c41, + 0xedae, 0xfd8f, 0xcdec, 0xddcd, 0xad2a, 0xbd0b, 0x8d68, 0x9d49, + 0x7e97, 0x6eb6, 0x5ed5, 0x4ef4, 0x3e13, 0x2e32, 0x1e51, 0x0e70, + 0xff9f, 0xefbe, 0xdfdd, 0xcffc, 0xbf1b, 0xaf3a, 0x9f59, 0x8f78, + 0x9188, 0x81a9, 0xb1ca, 0xa1eb, 0xd10c, 0xc12d, 0xf14e, 0xe16f, + 0x1080, 0x00a1, 0x30c2, 0x20e3, 0x5004, 0x4025, 0x7046, 0x6067, + 0x83b9, 0x9398, 0xa3fb, 0xb3da, 0xc33d, 0xd31c, 0xe37f, 0xf35e, + 0x02b1, 0x1290, 0x22f3, 0x32d2, 0x4235, 0x5214, 0x6277, 0x7256, + 0xb5ea, 0xa5cb, 0x95a8, 0x8589, 0xf56e, 0xe54f, 0xd52c, 0xc50d, + 0x34e2, 0x24c3, 0x14a0, 0x0481, 0x7466, 0x6447, 0x5424, 0x4405, + 0xa7db, 0xb7fa, 0x8799, 0x97b8, 0xe75f, 0xf77e, 0xc71d, 0xd73c, + 0x26d3, 0x36f2, 0x0691, 0x16b0, 0x6657, 0x7676, 0x4615, 0x5634, + 0xd94c, 0xc96d, 0xf90e, 0xe92f, 0x99c8, 0x89e9, 0xb98a, 0xa9ab, + 0x5844, 0x4865, 0x7806, 0x6827, 0x18c0, 0x08e1, 0x3882, 0x28a3, + 0xcb7d, 0xdb5c, 0xeb3f, 0xfb1e, 0x8bf9, 0x9bd8, 0xabbb, 0xbb9a, + 0x4a75, 0x5a54, 0x6a37, 0x7a16, 0x0af1, 0x1ad0, 0x2ab3, 0x3a92, + 0xfd2e, 0xed0f, 0xdd6c, 0xcd4d, 0xbdaa, 0xad8b, 0x9de8, 0x8dc9, + 0x7c26, 0x6c07, 0x5c64, 0x4c45, 0x3ca2, 0x2c83, 0x1ce0, 0x0cc1, + 0xef1f, 0xff3e, 0xcf5d, 0xdf7c, 0xaf9b, 0xbfba, 0x8fd9, 0x9ff8, + 0x6e17, 0x7e36, 0x4e55, 0x5e74, 0x2e93, 0x3eb2, 0x0ed1, 0x1ef0 +}; + + +uint32_t crc32tab[256] = { + 0x00000000, 0x04c11db7, 0x09823b6e, 0x0d4326d9, + 0x130476dc, 0x17c56b6b, 0x1a864db2, 0x1e475005, + 0x2608edb8, 0x22c9f00f, 0x2f8ad6d6, 0x2b4bcb61, + 0x350c9b64, 0x31cd86d3, 0x3c8ea00a, 0x384fbdbd, + 0x4c11db70, 0x48d0c6c7, 0x4593e01e, 0x4152fda9, + 0x5f15adac, 0x5bd4b01b, 0x569796c2, 0x52568b75, + 0x6a1936c8, 0x6ed82b7f, 0x639b0da6, 0x675a1011, + 0x791d4014, 0x7ddc5da3, 0x709f7b7a, 0x745e66cd, + 0x9823b6e0, 0x9ce2ab57, 0x91a18d8e, 0x95609039, + 0x8b27c03c, 0x8fe6dd8b, 0x82a5fb52, 0x8664e6e5, + 0xbe2b5b58, 0xbaea46ef, 0xb7a96036, 0xb3687d81, + 0xad2f2d84, 0xa9ee3033, 0xa4ad16ea, 0xa06c0b5d, + 0xd4326d90, 0xd0f37027, 0xddb056fe, 0xd9714b49, + 0xc7361b4c, 0xc3f706fb, 0xceb42022, 0xca753d95, + 0xf23a8028, 0xf6fb9d9f, 0xfbb8bb46, 0xff79a6f1, + 0xe13ef6f4, 0xe5ffeb43, 0xe8bccd9a, 0xec7dd02d, + 0x34867077, 0x30476dc0, 0x3d044b19, 0x39c556ae, + 0x278206ab, 0x23431b1c, 0x2e003dc5, 0x2ac12072, + 0x128e9dcf, 0x164f8078, 0x1b0ca6a1, 0x1fcdbb16, + 0x018aeb13, 0x054bf6a4, 0x0808d07d, 0x0cc9cdca, + 0x7897ab07, 0x7c56b6b0, 0x71159069, 0x75d48dde, + 0x6b93dddb, 0x6f52c06c, 0x6211e6b5, 0x66d0fb02, + 0x5e9f46bf, 0x5a5e5b08, 0x571d7dd1, 0x53dc6066, + 0x4d9b3063, 0x495a2dd4, 0x44190b0d, 0x40d816ba, + 0xaca5c697, 0xa864db20, 0xa527fdf9, 0xa1e6e04e, + 0xbfa1b04b, 0xbb60adfc, 0xb6238b25, 0xb2e29692, + 0x8aad2b2f, 0x8e6c3698, 0x832f1041, 0x87ee0df6, + 0x99a95df3, 0x9d684044, 0x902b669d, 0x94ea7b2a, + 0xe0b41de7, 0xe4750050, 0xe9362689, 0xedf73b3e, + 0xf3b06b3b, 0xf771768c, 0xfa325055, 0xfef34de2, + 0xc6bcf05f, 0xc27dede8, 0xcf3ecb31, 0xcbffd686, + 0xd5b88683, 0xd1799b34, 0xdc3abded, 0xd8fba05a, + 0x690ce0ee, 0x6dcdfd59, 0x608edb80, 0x644fc637, + 0x7a089632, 0x7ec98b85, 0x738aad5c, 0x774bb0eb, + 0x4f040d56, 0x4bc510e1, 0x46863638, 0x42472b8f, + 0x5c007b8a, 0x58c1663d, 0x558240e4, 0x51435d53, + 0x251d3b9e, 0x21dc2629, 0x2c9f00f0, 0x285e1d47, + 0x36194d42, 0x32d850f5, 0x3f9b762c, 0x3b5a6b9b, + 0x0315d626, 0x07d4cb91, 0x0a97ed48, 0x0e56f0ff, + 0x1011a0fa, 0x14d0bd4d, 0x19939b94, 0x1d528623, + 0xf12f560e, 0xf5ee4bb9, 0xf8ad6d60, 0xfc6c70d7, + 0xe22b20d2, 0xe6ea3d65, 0xeba91bbc, 0xef68060b, + 0xd727bbb6, 0xd3e6a601, 0xdea580d8, 0xda649d6f, + 0xc423cd6a, 0xc0e2d0dd, 0xcda1f604, 0xc960ebb3, + 0xbd3e8d7e, 0xb9ff90c9, 0xb4bcb610, 0xb07daba7, + 0xae3afba2, 0xaafbe615, 0xa7b8c0cc, 0xa379dd7b, + 0x9b3660c6, 0x9ff77d71, 0x92b45ba8, 0x9675461f, + 0x8832161a, 0x8cf30bad, 0x81b02d74, 0x857130c3, + 0x5d8a9099, 0x594b8d2e, 0x5408abf7, 0x50c9b640, + 0x4e8ee645, 0x4a4ffbf2, 0x470cdd2b, 0x43cdc09c, + 0x7b827d21, 0x7f436096, 0x7200464f, 0x76c15bf8, + 0x68860bfd, 0x6c47164a, 0x61043093, 0x65c52d24, + 0x119b4be9, 0x155a565e, 0x18197087, 0x1cd86d30, + 0x029f3d35, 0x065e2082, 0x0b1d065b, 0x0fdc1bec, + 0x3793a651, 0x3352bbe6, 0x3e119d3f, 0x3ad08088, + 0x2497d08d, 0x2056cd3a, 0x2d15ebe3, 0x29d4f654, + 0xc5a92679, 0xc1683bce, 0xcc2b1d17, 0xc8ea00a0, + 0xd6ad50a5, 0xd26c4d12, 0xdf2f6bcb, 0xdbee767c, + 0xe3a1cbc1, 0xe760d676, 0xea23f0af, 0xeee2ed18, + 0xf0a5bd1d, 0xf464a0aa, 0xf9278673, 0xfde69bc4, + 0x89b8fd09, 0x8d79e0be, 0x803ac667, 0x84fbdbd0, + 0x9abc8bd5, 0x9e7d9662, 0x933eb0bb, 0x97ffad0c, + 0xafb010b1, 0xab710d06, 0xa6322bdf, 0xa2f33668, + 0xbcb4666d, 0xb8757bda, 0xb5365d03, 0xb1f740b4 +}; + +// This function can be used to create a new table with a different polynom +void init_crc8tab(uint8_t l_code, uint8_t l_init) +{ + unsigned i, j, msb; + uint8_t nb; + uint8_t crc; + + for (i = 0; i < 256; ++i) { + crc = l_init; + nb = i ^ 0xff; + for (j = 0; j < 8; ++j) { + msb = (nb & (0x80 >> j)) && 1; + msb ^= (crc >> 7); + crc <<= 1; + if (msb) + crc ^= l_code; + } + crc8tab[i] = crc; + } +} + + +void init_crc16tab(uint16_t l_code, uint16_t l_init) +{ + unsigned i, j, msb; + uint8_t nb; + uint16_t crc; + + for (i = 0; i < 256; ++i) { + crc = l_init; + nb = i ^ 0xff; + for (j = 0; j < 8; ++j) { + msb = (nb & (0x80 >> j)) && 1; + msb ^= (crc >> 15); + crc <<= 1; + if (msb) + crc ^= l_code; + } + crc ^= 0xff00; + crc16tab[i] = crc; + } +} + + +void init_crc32tab(uint32_t l_code, uint32_t l_init) +{ + unsigned i, j, msb; + uint8_t nb; + uint32_t crc; + + for (i = 0; i < 256; ++i) { + crc = l_init; + nb = i ^ 0xff; + for (j = 0; j < 8; ++j) { + msb = (nb & (0x80 >> j)) && 1; + msb ^= (crc >> 31); + crc <<= 1; + if (msb) + crc ^= l_code; + } + crc ^= 0xffffff00; + crc32tab[i] = crc; + } +} + + +uint8_t crc8(uint8_t l_crc, const void *lp_data, unsigned l_nb) +{ + const uint8_t* data = (const uint8_t*)lp_data; + while (l_nb--) { + l_crc = crc8tab[l_crc ^ *(data++)]; + } + return (l_crc); +} + + +uint16_t crc16(uint16_t l_crc, const void *lp_data, unsigned l_nb) +{ + const uint8_t* data = (const uint8_t*)lp_data; + while (l_nb--) { + l_crc = + (l_crc << 8) ^ crc16tab[(l_crc >> 8) ^ *(data++)]; + } + return (l_crc); +} + + +uint32_t crc32(uint32_t l_crc, const void *lp_data, unsigned l_nb) +{ + const uint8_t* data = (const uint8_t*)lp_data; + while (l_nb--) { + l_crc = + (l_crc << 8) ^ crc32tab[((l_crc >> 24) ^ *(data++)) & 0xff]; + } + return (l_crc); +} diff --git a/contrib/edi/crc.h b/contrib/edi/crc.h new file mode 100644 index 0000000..b1785a1 --- /dev/null +++ b/contrib/edi/crc.h @@ -0,0 +1,59 @@ +/* + Copyright (C) 2003, 2004, 2005, 2006, 2007, 2008, 2009 Her Majesty the + Queen in Right of Canada (Communications Research Center Canada) + */ +/* + 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 <http://www.gnu.org/licenses/>. + */ + +#ifndef _CRC +#define _CRC + +#ifdef HAVE_CONFIG_H +# include "config.h" +#endif + +#ifndef _WIN32 + #include <stdint.h> +#else + #include <winsock2.h> // For types... + typedef BYTE uint8_t; + typedef WORD uint16_t; + typedef DWORD32 uint32_t; +#endif + + +#ifdef __cplusplus +extern "C" { // } +#endif + +void init_crc8tab(uint8_t l_code, uint8_t l_init); +uint8_t crc8(uint8_t l_crc, const void *lp_data, unsigned l_nb); +extern uint8_t crc8tab[]; + +void init_crc16tab(uint16_t l_code, uint16_t l_init); +uint16_t crc16(uint16_t l_crc, const void *lp_data, unsigned l_nb); +extern uint16_t crc16tab[]; + +void init_crc32tab(uint32_t l_code, uint32_t l_init); +uint32_t crc32(uint32_t l_crc, const void *lp_data, unsigned l_nb); +extern uint32_t crc32tab[]; + +#ifdef __cplusplus +} +#endif + +#endif //_CRC diff --git a/src/Outputs.cpp b/src/Outputs.cpp index a80ca08..be5b841 100644 --- a/src/Outputs.cpp +++ b/src/Outputs.cpp @@ -99,39 +99,6 @@ void ZMQ::set_encoder_type(encoder_selection_t& enc, int bitrate) bool ZMQ::write_frame(const uint8_t *buf, size_t len) { - switch (m_encoder) { - case encoder_selection_t::fdk_dabplus: - return send_frame(buf, len); - case encoder_selection_t::toolame_dab: - return write_toolame(buf, len); - } - throw logic_error("Unhandled encoder in ZMQ::write_frame"); -} - -bool ZMQ::write_toolame(const uint8_t *buf, size_t len) -{ - m_toolame_buffer.insert(m_toolame_buffer.end(), - buf, buf + len); - - // ODR-DabMux expects frames of length 3*bitrate - const auto frame_len = 3 * m_bitrate; - while (m_toolame_buffer.size() > frame_len) { - vec_u8 frame(frame_len); - // this is probably not very efficient - std::copy(m_toolame_buffer.begin(), m_toolame_buffer.begin() + frame_len, frame.begin()); - - bool success = send_frame(frame.data(), frame.size()); - if (not success) { - return false; - } - - m_toolame_buffer.erase(m_toolame_buffer.begin(), m_toolame_buffer.begin() + frame_len); - } - return true; -} - -bool ZMQ::send_frame(const uint8_t *buf, size_t len) -{ if (m_framebuf.size() != ZMQ_HEADER_SIZE + len) { m_framebuf.resize(ZMQ_HEADER_SIZE + len); } @@ -168,4 +135,72 @@ bool ZMQ::send_frame(const uint8_t *buf, size_t len) return true; } +EDI::EDI() { } + +EDI::~EDI() { } + +void EDI::add_udp_destination(const std::string& host, int port) +{ + auto dest = make_shared<edi::udp_destination_t>(); + dest->dest_addr = host; + m_edi_conf.dest_port = port; + m_edi_conf.destinations.push_back(dest); + + // We cannot carry AF packets over UDP, because they would be too large. + m_edi_conf.enable_pft = true; + + // TODO make FEC configurable +} + +void EDI::add_tcp_destination(const std::string& host, int port) +{ + auto dest = make_shared<edi::tcp_client_t>(); + dest->dest_addr = host; + if (dest->dest_port != 0 and dest->dest_port != port) { + throw runtime_error("All EDI UDP outputs must be to the same destination port"); + } + dest->dest_port = port; + m_edi_conf.destinations.push_back(dest); +} + +bool EDI::enabled() const +{ + return not m_edi_conf.destinations.empty(); +} + +bool EDI::write_frame(const uint8_t *buf, size_t len) +{ + if (not m_edi_sender) { + m_edi_sender = make_shared<edi::Sender>(m_edi_conf); + } + + edi::TagStarPTR edi_tagStarPtr; + edi_tagStarPtr.protocol = "DSTI"; + + edi::TagDSTI edi_tagDSTI; + + edi_tagDSTI.stihf = false; + edi_tagDSTI.atstf = false; + edi_tagDSTI.rfadf = false; + // DFCT is handled inside the TagDSTI + + edi::TagSSm edi_tagPayload; + // TODO make edi_tagPayload.stid configurable + edi_tagPayload.istd_data = buf; + edi_tagPayload.istd_length = len; + + // The above Tag Items will be assembled into a TAG Packet + edi::TagPacket edi_tagpacket(m_edi_conf.tagpacket_alignment); + + // put tags *ptr, DETI and all subchannels into one TagPacket + edi_tagpacket.tag_items.push_back(&edi_tagStarPtr); + edi_tagpacket.tag_items.push_back(&edi_tagDSTI); + edi_tagpacket.tag_items.push_back(&edi_tagPayload); + + m_edi_sender->write(edi_tagpacket); + + // TODO Handle TCP disconnect + return true; +} + } diff --git a/src/Outputs.h b/src/Outputs.h index 30b20c8..297f926 100644 --- a/src/Outputs.h +++ b/src/Outputs.h @@ -25,6 +25,10 @@ #include <cstdio> #include "common.h" #include "zmq.hpp" +#include "edi/TagItems.h" +#include "edi/TagPacket.h" +#include "edi/AFPacket.h" +#include "edi/Transport.h" extern "C" { #include "encryption.h" } @@ -109,9 +113,6 @@ class ZMQ: public Base { virtual bool write_frame(const uint8_t *buf, size_t len) override; private: - virtual bool write_toolame(const uint8_t *buf, size_t len); - virtual bool send_frame(const uint8_t *buf, size_t len); - zmq::context_t m_ctx; zmq::socket_t m_sock; @@ -120,7 +121,28 @@ class ZMQ: public Base { encoder_selection_t m_encoder = encoder_selection_t::fdk_dabplus; using vec_u8 = std::vector<uint8_t>; vec_u8 m_framebuf; - std::deque<uint8_t> m_toolame_buffer; +}; + + +class EDI: public Base { + public: + EDI(); + EDI(const EDI&) = delete; + EDI& operator=(const EDI&) = delete; + virtual ~EDI() override; + + void add_udp_destination(const std::string& host, int port); + void add_tcp_destination(const std::string& host, int port); + + bool enabled() const; + + virtual bool write_frame(const uint8_t *buf, size_t len) override; + + // TODO audio levels metadata + + private: + edi::configuration_t m_edi_conf; + std::shared_ptr<edi::Sender> m_edi_sender; }; } diff --git a/src/SampleQueue.h b/src/SampleQueue.h index aeeb8d4..f39dbd7 100644 --- a/src/SampleQueue.h +++ b/src/SampleQueue.h @@ -275,6 +275,11 @@ public: return ret; } + void set_max_size(size_t max_size) + { + m_max_size = max_size; + } + private: std::deque<T> m_queue; mutable std::mutex m_mutex; diff --git a/src/odr-audioenc.cpp b/src/odr-audioenc.cpp index 7f11edd..5f95538 100644 --- a/src/odr-audioenc.cpp +++ b/src/odr-audioenc.cpp @@ -1,6 +1,6 @@ /* ------------------------------------------------------------------ * Copyright (C) 2011 Martin Storsjo - * Copyright (C) 2018 Matthias P. Braendli + * Copyright (C) 2019 Matthias P. Braendli * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -36,7 +36,7 @@ * - \ref VLCInput.h VLC Input * - \ref AlsaInput.h Alsa Input * - \ref JackInput.h JACK Input - * - \ref Outputs.h ZeroMQ and file outputs + * - \ref Outputs.h ZeroMQ, file and EDI outputs * - \ref SampleQueue.h * - \ref charset.h Charset conversion * - \ref toolame.h libtolame API @@ -82,7 +82,7 @@ extern "C" { #include "fdk-aac/aacenc_lib.h" extern "C" { -#include "contrib/fec/fec.h" +#include "fec/fec.h" #include "libtoolame-dab/toolame.h" } @@ -94,31 +94,6 @@ using vec_u8 = std::vector<uint8_t>; using namespace std; -struct audioenc_settings_t { - int sample_rate=48000; - int channels=2; - - // For the ALSA input - string alsa_device; - - // For the file input - string infile; - bool continue_after_eof = false; - int raw_input = 0; - - // For the VLC input - string vlc_uri; - string vlc_icytext_file; - bool vlc_icytext_dlplus = false; - string vlc_gain; - string vlc_cache; - vector<string> vlc_additional_opts; - unsigned verbosity = 0; - - string jack_name; - - bool drift_compensation = false; -}; void usage(const char* name) @@ -211,6 +186,7 @@ void usage(const char* name) " -or- a single dash '-' to denote stdout\n" " If more than one ZMQ output is given, the socket\n" " will be connected to all listed endpoints.\n" + " -e, --edi=URI EDI output uri, (e.g. 'tcp://localhost:7000')\n" " -k, --secret-key=FILE Enable ZMQ encryption with the given secret key.\n" " -p, --pad=BYTES Enable PAD insertion and set PAD size in bytes.\n" " -P, --pad-fifo=FILENAME Set PAD data input fifo name" @@ -401,48 +377,6 @@ static void drift_compensation_delay(int sample_rate, int channels, size_t bytes timepoint_last_compensation += wait_time; } -static shared_ptr<InputInterface> initialise_input( - audioenc_settings_t& s, - SampleQueue<uint8_t>& queue) -{ - shared_ptr<InputInterface> input; - - if (not s.infile.empty()) { - input = make_shared<FileInput>(s.infile, s.raw_input, s.sample_rate, - s.continue_after_eof, queue); - } -#if HAVE_JACK - else if (not s.jack_name.empty()) { - input = make_shared<JackInput>(s.jack_name, s.channels, s.sample_rate, - queue); - } -#endif -#if HAVE_VLC - else if (not s.vlc_uri.empty()) { - input = make_shared<VLCInput>(s.vlc_uri, s.sample_rate, s.channels, - s.verbosity, s.vlc_gain, s.vlc_cache, s.vlc_additional_opts, - queue); - } -#endif -#if HAVE_ALSA - else if (s.drift_compensation) { - input = make_shared<AlsaInputThreaded>(s.alsa_device, s.channels, - s.sample_rate, queue); - } - else { - input = make_shared<AlsaInputDirect>(s.alsa_device, s.channels, - s.sample_rate, queue); - } -#endif - - if (not input) { - throw logic_error("Initialising input incomplete!"); - } - - input->prepare(); - - return input; -} #define no_argument 0 @@ -453,33 +387,58 @@ static shared_ptr<InputInterface> initialise_input( #define STATUS_OVERRUN 0x2 #define STATUS_UNDERRUN 0x4 -int main(int argc, char *argv[]) -{ - audioenc_settings_t settings; +struct AudioEnc { +public: + int sample_rate=48000; + int channels=2; - bool restart_on_fault = false; - int fault_counter = 0; + // For the ALSA input + string alsa_device; - int bitrate = 0; // 0 is default - int ch=0; + // For the file input + string infile; + bool continue_after_eof = false; + int raw_input = 0; + + // For the VLC input + string vlc_uri; + string vlc_icytext_file; + bool vlc_icytext_dlplus = false; + string vlc_gain; + string vlc_cache; + vector<string> vlc_additional_opts; + unsigned verbosity = 0; + + string jack_name; + + bool drift_compensation = false; encoder_selection_t selected_encoder = encoder_selection_t::fdk_dabplus; + bool afterburner = true; + uint32_t bandwidth = 0; + int bitrate = 0; // 0 means default bitrate + + int dab_psy_model = 1; + + bool restart_on_fault = false; + int fault_counter = 0; + + std::deque<uint8_t> toolame_buffer; shared_ptr<Output::File> file_output; shared_ptr<Output::ZMQ> zmq_output; + Output::EDI edi_output; vector<string> output_uris; + vector<string> edi_output_uris; void *rs_handler = nullptr; - bool afterburner = true; - uint32_t bandwidth = 0; AACENC_InfoStruct info = { 0 }; int aot = AOT_NONE; string decode_wavfilename; string dab_channel_mode; - int dab_psy_model = 1; /* Keep track of peaks */ int peak_left = 0; @@ -505,214 +464,33 @@ int main(int argc, char *argv[]) char* keyfile = nullptr; char secretkey[CURVE_KEYLEN+1]; - const struct option longopts[] = { - {"bitrate", required_argument, 0, 'b'}, - {"bandwidth", required_argument, 0, 'B'}, - {"channels", required_argument, 0, 'c'}, - {"dabmode", required_argument, 0, 4 }, - {"dabpsy", required_argument, 0, 5 }, - {"device", required_argument, 0, 'd'}, - {"decode", required_argument, 0, 6 }, - {"format", required_argument, 0, 'f'}, - {"input", required_argument, 0, 'i'}, - {"jack", required_argument, 0, 'j'}, - {"output", required_argument, 0, 'o'}, - {"pad", required_argument, 0, 'p'}, - {"pad-fifo", required_argument, 0, 'P'}, - {"rate", required_argument, 0, 'r'}, - {"secret-key", required_argument, 0, 'k'}, - {"silence", required_argument, 0, 's'}, - {"vlc-cache", required_argument, 0, 'C'}, - {"vlc-gain", required_argument, 0, 'g'}, - {"vlc-uri", required_argument, 0, 'v'}, - {"vlc-opt", required_argument, 0, 'L'}, - {"write-icy-text", required_argument, 0, 'w'}, - {"write-icy-text-dl-plus", no_argument, 0, 'W'}, - {"aaclc", no_argument, 0, 0 }, - {"dab", no_argument, 0, 'a'}, - {"drift-comp", no_argument, 0, 'D'}, - {"fifo-silence", no_argument, 0, 3 }, - {"help", no_argument, 0, 'h'}, - {"level", no_argument, 0, 'l'}, - {"no-afterburner", no_argument, 0, 'A'}, - {"ps", no_argument, 0, 2 }, - {"restart", no_argument, 0, 'R'}, - {"sbr", no_argument, 0, 1 }, - {"verbosity", no_argument, 0, 'V'}, - {0, 0, 0, 0}, - }; - - fprintf(stderr, - "Welcome to %s %s, compiled at %s, %s", - PACKAGE_NAME, -#if defined(GITVERSION) - GITVERSION, -#else - PACKAGE_VERSION, -#endif - __DATE__, __TIME__); - fprintf(stderr, "\n"); - fprintf(stderr, " http://opendigitalradio.org\n\n"); + SampleQueue<uint8_t> queue; + HANDLE_AACENCODER encoder; + unique_ptr<AACDecoder> decoder; - if (argc < 2) { - usage(argv[0]); - return 1; - } + AudioEnc() : queue(BYTES_PER_SAMPLE, channels, 0, drift_compensation) { } + AudioEnc(const AudioEnc&) = delete; + AudioEnc& operator=(const AudioEnc&) = delete; + ~AudioEnc(); - int index; - while(ch != -1) { - ch = getopt_long(argc, argv, "aAhDlRVb:B:c:f:i:j:k:L:o:r:d:p:P:s:v:w:Wg:C:", longopts, &index); - switch (ch) { - case 0: // AAC-LC - aot = AOT_DABPLUS_AAC_LC; - break; - case 1: // SBR - aot = AOT_DABPLUS_SBR; - break; - case 2: // PS - aot = AOT_DABPLUS_PS; - break; - case 3: // FIFO Silence - settings.continue_after_eof = true; - // Enable drift compensation, otherwise we would block instead of inserting silence. - settings.drift_compensation = true; - break; - case 4: // DAB channel mode - dab_channel_mode = optarg; - if (not( dab_channel_mode == "s" or - dab_channel_mode == "d" or - dab_channel_mode == "j" or - dab_channel_mode == "m")) { - fprintf(stderr, "Invalid DAB channel mode\n"); - usage(argv[0]); - return 1; - } - break; - case 5: // DAB psy model - dab_psy_model = std::stoi(optarg); - break; - case 6: // Enable loopback decoder for AAC - decode_wavfilename = optarg; - break; - case 'a': - selected_encoder = encoder_selection_t::toolame_dab; - break; - case 'A': - afterburner = false; - break; - case 'b': - bitrate = std::stoi(optarg); - break; - case 'B': - bandwidth = std::stoi(optarg); - break; - case 'c': - settings.channels = std::stoi(optarg); - break; - case 'd': - settings.alsa_device = optarg; - break; - case 'D': - settings.drift_compensation = true; - break; - case 'f': - if (strcmp(optarg, "raw") == 0) { - settings.raw_input = 1; - } - else if (strcmp(optarg, "wav") != 0) { - usage(argv[0]); - return 1; - } - break; - case 'i': - settings.infile = optarg; - break; - case 'j': -#if HAVE_JACK - settings.jack_name = optarg; -#else - fprintf(stderr, "JACK disabled at compile time!\n"); - return 1; -#endif - break; - case 'k': - keyfile = optarg; - break; - case 'l': - show_level = 1; - break; - case 'o': - output_uris.push_back(optarg); - break; - case 'p': - padlen = std::stoi(optarg); - break; - case 'P': - pad_fifo = optarg; - break; - case 'r': - settings.sample_rate = std::stoi(optarg); - break; - case 'R': - restart_on_fault = true; - break; - case 's': - silence_timeout = std::stoi(optarg); - if (silence_timeout > 0 && silence_timeout < 3600*24*30) { - die_on_silence = true; - } - else { - fprintf(stderr, "Invalid silence timeout (%d) given!\n", silence_timeout); - return 1; - } - - break; -#ifdef HAVE_VLC - case 'v': - settings.vlc_uri = optarg; - break; - case 'w': - settings.vlc_icytext_file = optarg; - break; - case 'W': - settings.vlc_icytext_dlplus = true; - break; - case 'g': - settings.vlc_gain = optarg; - break; - case 'C': - settings.vlc_cache = optarg; - break; - case 'L': - settings.vlc_additional_opts.push_back(optarg); - break; -#else - case 'v': - case 'w': - fprintf(stderr, "VLC input not enabled at compile time!\n"); - return 1; -#endif - case 'V': - settings.verbosity++; - break; - case '?': - case 'h': - usage(argv[0]); - return 1; - } - } + int run(); + bool send_frame(const uint8_t *buf, size_t len); + shared_ptr<InputInterface> initialise_input(); +}; +int AudioEnc::run() +{ int num_inputs = 0; #if HAVE_ALSA - if (not settings.alsa_device.empty()) num_inputs++; + if (not alsa_device.empty()) num_inputs++; #endif - if (not settings.infile.empty()) num_inputs++; + if (not infile.empty()) num_inputs++; #if HAVE_JACK - if (not settings.jack_name.empty()) num_inputs++; + if (not jack_name.empty()) num_inputs++; #endif #if HAVE_VLC - if (not settings.vlc_uri.empty()) num_inputs++; + if (not vlc_uri.empty()) num_inputs++; #endif if (num_inputs == 0) { @@ -737,7 +515,7 @@ int main(int argc, char *argv[]) return 1; } - if ( ! (settings.sample_rate == 32000 || settings.sample_rate == 48000)) { + if ( ! (sample_rate == 32000 || sample_rate == 48000)) { fprintf(stderr, "Invalid sample rate. Possible values are: 32000, 48000.\n"); return 1; } @@ -747,7 +525,7 @@ int main(int argc, char *argv[]) bitrate = 192; } - if ( ! (settings.sample_rate == 24000 || settings.sample_rate == 48000)) { + if ( ! (sample_rate == 24000 || sample_rate == 48000)) { fprintf(stderr, "Invalid sample rate. Possible values are: 24000, 48000.\n"); return 1; } @@ -758,8 +536,8 @@ int main(int argc, char *argv[]) return 1; } - if (output_uris.empty()) { - fprintf(stderr, "No output URI defined\n"); + if (output_uris.empty() and edi_output_uris.empty()) { + fprintf(stderr, "No output defined\n"); return 1; } @@ -791,6 +569,34 @@ int main(int argc, char *argv[]) } } + for (const auto& uri : edi_output_uris) { + if (uri.compare(0, 6, "tcp://") == 0 or + uri.compare(0, 6, "udp://") == 0) { + auto host_port_sep_ix = uri.find(':', 6); + if (host_port_sep_ix != string::npos) { + auto host = uri.substr(6, host_port_sep_ix - 6); + auto port = std::stoi(uri.substr(host_port_sep_ix + 1)); + + auto proto = uri.substr(0, 3); + if (proto == "tcp") { + edi_output.add_tcp_destination(host, port); + } + else if (proto == "udp") { + edi_output.add_udp_destination(host, port); + } + else { + throw logic_error("unhandled proto"); + } + } + else { + fprintf(stderr, "Invalid EDI URL host!\n"); + } + } + else { + fprintf(stderr, "Invalid EDI protocol!\n"); + } + } + if (padlen != 0) { int flags; if (mkfifo(pad_fifo, S_IWUSR | S_IRUSR | S_IRGRP | S_IROTH) != 0) { @@ -814,13 +620,10 @@ int main(int argc, char *argv[]) vec_u8 input_buf; - HANDLE_AACENCODER encoder; - unique_ptr<AACDecoder> decoder; - if (selected_encoder == encoder_selection_t::fdk_dabplus) { int subchannel_index = bitrate / 8; - if (prepare_aac_encoder(&encoder, subchannel_index, settings.channels, - settings.sample_rate, afterburner, bandwidth, &aot) != 0) { + if (prepare_aac_encoder(&encoder, subchannel_index, channels, + sample_rate, afterburner, bandwidth, &aot) != 0) { fprintf(stderr, "Encoder preparation failed\n"); return 1; } @@ -831,7 +634,7 @@ int main(int argc, char *argv[]) } // Each DAB+ frame will need input_size audio bytes - const int input_size = settings.channels * BYTES_PER_SAMPLE * info.frameLength; + const int input_size = channels * BYTES_PER_SAMPLE * info.frameLength; fprintf(stderr, "DAB+ Encoding: framelen=%d (%dB)\n", info.frameLength, input_size); @@ -846,7 +649,7 @@ int main(int argc, char *argv[]) int err = toolame_init(); if (err == 0) { - err = toolame_set_samplerate(settings.sample_rate); + err = toolame_set_samplerate(sample_rate); } if (err == 0) { @@ -854,15 +657,15 @@ int main(int argc, char *argv[]) } if (dab_channel_mode.empty()) { - if (settings.channels == 2) { + if (channels == 2) { dab_channel_mode = 'j'; // Default to joint-stereo } - else if (settings.channels == 1) { + else if (channels == 1) { dab_channel_mode = 'm'; // Default to mono } else { fprintf(stderr, "Unsupported channels number %d\n", - settings.channels); + channels); return 1; } } @@ -885,7 +688,7 @@ int main(int argc, char *argv[]) return err; } - input_buf.resize(settings.channels * 1152 * BYTES_PER_SAMPLE); + input_buf.resize(channels * 1152 * BYTES_PER_SAMPLE); if (not decode_wavfilename.empty()) { fprintf(stderr, "--decode not supported for DAB\n"); @@ -899,15 +702,15 @@ int main(int argc, char *argv[]) * is active. This is only valid for FDK-AAC. */ const int enc_calls_per_output = (aot == AOT_DABPLUS_AAC_LC) ? - settings.sample_rate / 8000 : - settings.sample_rate / 16000; + sample_rate / 8000 : + sample_rate / 16000; int max_size = 32*input_buf.size() + NUM_SAMPLES_PER_CALL; /*! The SampleQueue \c queue is given to the inputs, so that they * can fill it. */ - SampleQueue<uint8_t> queue(BYTES_PER_SAMPLE, settings.channels, max_size, settings.drift_compensation); + queue.set_max_size(max_size); /* symsize=8, gfpoly=0x11d, fcr=0, prim=1, nroots=10, pad=135 */ rs_handler = init_rs_char(8, 0x11d, 0, 1, 10, 135); @@ -918,14 +721,16 @@ int main(int argc, char *argv[]) shared_ptr<InputInterface> input; try { - input = initialise_input(settings, queue); + input = initialise_input(); } catch (const runtime_error& e) { fprintf(stderr, "Initialising input triggered exception: %s\n", e.what()); return 1; } - zmq_output->set_encoder_type(selected_encoder, bitrate); + if (zmq_output) { + zmq_output->set_encoder_type(selected_encoder, bitrate); + } int outbuf_size; vec_u8 zmqframebuf; @@ -1029,7 +834,7 @@ int main(int argc, char *argv[]) } try { - input = initialise_input(settings, queue); + input = initialise_input(); } catch (const runtime_error& e) { fprintf(stderr, "Initialising input triggered exception: %s\n", e.what()); @@ -1051,14 +856,14 @@ int main(int argc, char *argv[]) break; } - if (settings.drift_compensation) { + if (drift_compensation) { size_t overruns = 0; size_t bytes_from_queue = queue.pop(&input_buf[0], input_buf.size(), &overruns); // returns bytes if (bytes_from_queue != input_buf.size()) { - expand_missing_samples(input_buf, settings.channels, bytes_from_queue); + expand_missing_samples(input_buf, channels, bytes_from_queue); } read_bytes = input_buf.size(); - drift_compensation_delay(settings.sample_rate, settings.channels, read_bytes); + drift_compensation_delay(sample_rate, channels, read_bytes); if (bytes_from_queue != input_buf.size()) { status |= STATUS_UNDERRUN; @@ -1096,7 +901,7 @@ int main(int argc, char *argv[]) } try { - input = initialise_input(settings, queue); + input = initialise_input(); } catch (const runtime_error& e) { fprintf(stderr, "Initialising input triggered exception: %s\n", e.what()); @@ -1117,10 +922,10 @@ int main(int argc, char *argv[]) * we can hand over to ODR-PadEnc. */ #if HAVE_VLC - if (not settings.vlc_uri.empty() and not settings.vlc_icytext_file.empty()) { + if (not vlc_uri.empty() and not vlc_icytext_file.empty()) { // Using std::dynamic_pointer_cast would be safer, but is C++17 VLCInput *vlc_input = (VLCInput*)(input.get()); - vlc_input->write_icy_text(settings.vlc_icytext_file, settings.vlc_icytext_dlplus); + vlc_input->write_icy_text(vlc_icytext_file, vlc_icytext_dlplus); } #endif @@ -1147,7 +952,7 @@ int main(int argc, char *argv[]) */ if (die_on_silence && MAX(peak_left, peak_right) == 0) { const unsigned int frame_time_msec = 1000ul * - read_bytes / (BYTES_PER_SAMPLE * settings.channels * settings.sample_rate); + read_bytes / (BYTES_PER_SAMPLE * channels * sample_rate); measured_silence_ms += frame_time_msec; @@ -1223,10 +1028,10 @@ int main(int argc, char *argv[]) */ short input_buffers[2][1152]; - if (settings.channels == 1) { + if (channels == 1) { memcpy(input_buffers[0], &input_buf[0], 1152 * BYTES_PER_SAMPLE); } - else if (settings.channels == 2) { + else if (channels == 2) { for (int i = 0; i < 1152; i++) { int16_t l = input_buf[4*i] | (input_buf[4*i+1] << 8); int16_t r = input_buf[4*i+2] | (input_buf[4*i+3] << 8); @@ -1292,36 +1097,48 @@ int main(int argc, char *argv[]) numOutBytes = outbuf_size; } - if (numOutBytes != 0) { - if (file_output) { - file_output->write_frame(outbuf.data(), numOutBytes); - } - else if (zmq_output) { - bool success = zmq_output->write_frame(outbuf.data(), numOutBytes); + if (numOutBytes > 0 and selected_encoder == encoder_selection_t::toolame_dab) { + toolame_buffer.insert(toolame_buffer.end(), outbuf.begin(), outbuf.begin() + numOutBytes); + + // ODR-DabMux expects frames of length 3*bitrate + const auto frame_len = 3 * bitrate; + while (toolame_buffer.size() > frame_len) { + vec_u8 frame(frame_len); + // this is probably not very efficient + std::copy(toolame_buffer.begin(), toolame_buffer.begin() + frame_len, frame.begin()); + toolame_buffer.erase(toolame_buffer.begin(), toolame_buffer.begin() + frame_len); + bool success = send_frame(frame.data(), frame.size()); if (not success) { - fprintf(stderr, "ZeroMQ send error !\n"); + fprintf(stderr, "Send error !\n"); send_error_count ++; } - - if (send_error_count > 10) { - fprintf(stderr, "ZeroMQ send failed ten times, aborting!\n"); - retval = 4; - break; - } + } + } + else if (numOutBytes > 0 and selected_encoder == encoder_selection_t::fdk_dabplus) { + bool success = send_frame(outbuf.data(), numOutBytes); + if (not success) { + fprintf(stderr, "Send error !\n"); + send_error_count ++; } } + if (send_error_count > 10) { + fprintf(stderr, "Send failed ten times, aborting!\n"); + retval = 4; + break; + } + if (numOutBytes != 0) { if (show_level) { - if (settings.channels == 1) { + if (channels == 1) { fprintf(stderr, "\rIn: [%-6s] %1s %1s %1s", level(1, MAX(peak_right, peak_left)), status & STATUS_PAD_INSERTED ? "P" : " ", status & STATUS_UNDERRUN ? "U" : " ", status & STATUS_OVERRUN ? "O" : " "); } - else if (settings.channels == 2) { + else if (channels == 2) { fprintf(stderr, "\rIn: [%6s|%-6s] %1s %1s %1s", level(0, peak_left), level(1, peak_right), @@ -1351,7 +1168,45 @@ int main(int argc, char *argv[]) } while (read_bytes > 0); fprintf(stderr, "\n"); + return retval; +} + +bool AudioEnc::send_frame(const uint8_t *buf, size_t len) +{ + if (file_output) { + return file_output->write_frame(buf, len); + } + else if (zmq_output) { + return zmq_output->write_frame(buf, len); + } + else if (edi_output.enabled()) { + switch (selected_encoder) { + case encoder_selection_t::fdk_dabplus: + { + // STI/EDI specifies that one AF packet must contain 24ms worth of data, + // therefore we must split the superframe into five parts + if (len % 5 != 0) { + throw logic_error("Superframe size not multiple of 5"); + } + + const size_t blocksize = len/5; + for (size_t i = 0; i < 5; i++) { + bool success = edi_output.write_frame(buf + i * blocksize, blocksize); + if (not success) { + return false; + } + } + return true; + } + case encoder_selection_t::toolame_dab: + return edi_output.write_frame(buf, len); + } + } + return false; +} +AudioEnc::~AudioEnc() +{ file_output.reset(); zmq_output.reset(); @@ -1360,7 +1215,256 @@ int main(int argc, char *argv[]) if (selected_encoder == encoder_selection_t::fdk_dabplus) { aacEncClose(&encoder); } +} - return retval; +shared_ptr<InputInterface> AudioEnc::initialise_input() +{ + shared_ptr<InputInterface> input; + + if (not infile.empty()) { + input = make_shared<FileInput>(infile, raw_input, sample_rate, + continue_after_eof, queue); + } +#if HAVE_JACK + else if (not jack_name.empty()) { + input = make_shared<JackInput>(jack_name, channels, sample_rate, + queue); + } +#endif +#if HAVE_VLC + else if (not vlc_uri.empty()) { + input = make_shared<VLCInput>(vlc_uri, sample_rate, channels, + verbosity, vlc_gain, vlc_cache, vlc_additional_opts, + queue); + } +#endif +#if HAVE_ALSA + else if (drift_compensation) { + input = make_shared<AlsaInputThreaded>(alsa_device, channels, + sample_rate, queue); + } + else { + input = make_shared<AlsaInputDirect>(alsa_device, channels, + sample_rate, queue); + } +#endif + + if (not input) { + throw logic_error("Initialising input incomplete!"); + } + + input->prepare(); + + return input; +} + +int main(int argc, char *argv[]) +{ + AudioEnc audio_enc; + + const struct option longopts[] = { + {"bitrate", required_argument, 0, 'b'}, + {"bandwidth", required_argument, 0, 'B'}, + {"channels", required_argument, 0, 'c'}, + {"dabmode", required_argument, 0, 4 }, + {"dabpsy", required_argument, 0, 5 }, + {"device", required_argument, 0, 'd'}, + {"edi", required_argument, 0, 'e'}, + {"decode", required_argument, 0, 6 }, + {"format", required_argument, 0, 'f'}, + {"input", required_argument, 0, 'i'}, + {"jack", required_argument, 0, 'j'}, + {"output", required_argument, 0, 'o'}, + {"pad", required_argument, 0, 'p'}, + {"pad-fifo", required_argument, 0, 'P'}, + {"rate", required_argument, 0, 'r'}, + {"secret-key", required_argument, 0, 'k'}, + {"silence", required_argument, 0, 's'}, + {"vlc-cache", required_argument, 0, 'C'}, + {"vlc-gain", required_argument, 0, 'g'}, + {"vlc-uri", required_argument, 0, 'v'}, + {"vlc-opt", required_argument, 0, 'L'}, + {"write-icy-text", required_argument, 0, 'w'}, + {"write-icy-text-dl-plus", no_argument, 0, 'W'}, + {"aaclc", no_argument, 0, 0 }, + {"dab", no_argument, 0, 'a'}, + {"drift-comp", no_argument, 0, 'D'}, + {"fifo-silence", no_argument, 0, 3 }, + {"help", no_argument, 0, 'h'}, + {"level", no_argument, 0, 'l'}, + {"no-afterburner", no_argument, 0, 'A'}, + {"ps", no_argument, 0, 2 }, + {"restart", no_argument, 0, 'R'}, + {"sbr", no_argument, 0, 1 }, + {"verbosity", no_argument, 0, 'V'}, + {0, 0, 0, 0}, + }; + + fprintf(stderr, + "Welcome to %s %s, compiled at %s, %s", + PACKAGE_NAME, +#if defined(GITVERSION) + GITVERSION, +#else + PACKAGE_VERSION, +#endif + __DATE__, __TIME__); + fprintf(stderr, "\n"); + fprintf(stderr, " http://opendigitalradio.org\n\n"); + + + if (argc < 2) { + usage(argv[0]); + return 1; + } + + int ch=0; + int index; + while(ch != -1) { + ch = getopt_long(argc, argv, "aAhDlRVb:B:c:e:f:i:j:k:L:o:r:d:p:P:s:v:w:Wg:C:", longopts, &index); + switch (ch) { + case 0: // AAC-LC + audio_enc.aot = AOT_DABPLUS_AAC_LC; + break; + case 1: // SBR + audio_enc.aot = AOT_DABPLUS_SBR; + break; + case 2: // PS + audio_enc.aot = AOT_DABPLUS_PS; + break; + case 3: // FIFO Silence + audio_enc.continue_after_eof = true; + // Enable drift compensation, otherwise we would block instead of inserting silence. + audio_enc.drift_compensation = true; + break; + case 4: // DAB channel mode + audio_enc.dab_channel_mode = optarg; + if (not( audio_enc.dab_channel_mode == "s" or + audio_enc.dab_channel_mode == "d" or + audio_enc.dab_channel_mode == "j" or + audio_enc.dab_channel_mode == "m")) { + fprintf(stderr, "Invalid DAB channel mode\n"); + usage(argv[0]); + return 1; + } + break; + case 5: // DAB psy model + audio_enc.dab_psy_model = std::stoi(optarg); + break; + case 6: // Enable loopback decoder for AAC + audio_enc.decode_wavfilename = optarg; + break; + case 'a': + audio_enc.selected_encoder = encoder_selection_t::toolame_dab; + break; + case 'A': + audio_enc.afterburner = false; + break; + case 'b': + audio_enc.bitrate = std::stoi(optarg); + break; + case 'B': + audio_enc.bandwidth = std::stoi(optarg); + break; + case 'c': + audio_enc.channels = std::stoi(optarg); + break; + case 'd': + audio_enc.alsa_device = optarg; + break; + case 'D': + audio_enc.drift_compensation = true; + break; + case 'e': + audio_enc.edi_output_uris.push_back(optarg); + break; + case 'f': + if (strcmp(optarg, "raw") == 0) { + audio_enc.raw_input = 1; + } + else if (strcmp(optarg, "wav") != 0) { + usage(argv[0]); + return 1; + } + break; + case 'i': + audio_enc.infile = optarg; + break; + case 'j': +#if HAVE_JACK + audio_enc.jack_name = optarg; +#else + fprintf(stderr, "JACK disabled at compile time!\n"); + return 1; +#endif + break; + case 'k': + audio_enc.keyfile = optarg; + break; + case 'l': + audio_enc.show_level = 1; + break; + case 'o': + audio_enc.output_uris.push_back(optarg); + break; + case 'p': + audio_enc.padlen = std::stoi(optarg); + break; + case 'P': + audio_enc.pad_fifo = optarg; + break; + case 'r': + audio_enc.sample_rate = std::stoi(optarg); + break; + case 'R': + audio_enc.restart_on_fault = true; + break; + case 's': + audio_enc.silence_timeout = std::stoi(optarg); + if (audio_enc.silence_timeout > 0 && audio_enc.silence_timeout < 3600*24*30) { + audio_enc.die_on_silence = true; + } + else { + fprintf(stderr, "Invalid silence timeout (%d) given!\n", audio_enc.silence_timeout); + return 1; + } + + break; +#ifdef HAVE_VLC + case 'v': + audio_enc.vlc_uri = optarg; + break; + case 'w': + audio_enc.vlc_icytext_file = optarg; + break; + case 'W': + audio_enc.vlc_icytext_dlplus = true; + break; + case 'g': + audio_enc.vlc_gain = optarg; + break; + case 'C': + audio_enc.vlc_cache = optarg; + break; + case 'L': + audio_enc.vlc_additional_opts.push_back(optarg); + break; +#else + case 'v': + case 'w': + fprintf(stderr, "VLC input not enabled at compile time!\n"); + return 1; +#endif + case 'V': + audio_enc.verbosity++; + break; + case '?': + case 'h': + usage(argv[0]); + return 1; + } + } + + return audio_enc.run(); } |