From d14814a92377084177753c7a60d83a9307ad0672 Mon Sep 17 00:00:00 2001 From: "Matthias P. Braendli" Date: Sat, 16 Jan 2021 08:06:09 +0100 Subject: Update common code to latest, update zmq.hpp and adapt --- src/EtiReader.cpp | 59 ++++- src/EtiReader.h | 8 +- src/InputZeroMQReader.cpp | 2 +- src/OutputZeroMQ.cpp | 4 +- src/zmq.hpp | 602 ---------------------------------------------- 5 files changed, 53 insertions(+), 622 deletions(-) delete mode 100644 src/zmq.hpp (limited to 'src') diff --git a/src/EtiReader.cpp b/src/EtiReader.cpp index 33194b2..51266af 100644 --- a/src/EtiReader.cpp +++ b/src/EtiReader.cpp @@ -28,6 +28,7 @@ #include "Log.h" #include "PcDebug.h" #include "TimestampDecoder.h" +#include "edi/common.hpp" #include #include @@ -544,6 +545,10 @@ void EdiTransport::Open(const std::string& uri) const string proto = uri.substr(0, 3); if (proto == "udp") { + if (m_proto == Proto::TCP) { + throw std::invalid_argument("Cannot specify both TCP and UDP urls"); + } + size_t found_port = uri.find_first_of(":", 6); if (found_port == string::npos) { throw std::invalid_argument("EDI UDP input port must be provided"); @@ -565,17 +570,15 @@ void EdiTransport::Open(const std::string& uri) etiLog.level(info) << "EDI UDP input: host:" << m_bindto << ", source:" << m_mcastaddr << ", port:" << m_port; - // The max_fragments_queued is only a protection against a runaway - // memory usage. - // Rough calculation: - // 300 seconds, 24ms per frame, up to 20 fragments per frame - const size_t max_fragments_queued = 20 * 300 * 1000 / 24; - - m_udp_rx.start(m_port, m_bindto, m_mcastaddr, max_fragments_queued); + m_udp_rx.add_receive_port(m_port, m_bindto, m_mcastaddr); m_proto = Proto::UDP; m_enabled = true; } else if (proto == "tcp") { + if (m_proto != Proto::Unspecified) { + throw std::invalid_argument("Cannot call Open several times with TCP"); + } + size_t found_port = uri.find_first_of(":", 6); if (found_port == string::npos) { throw std::invalid_argument("EDI TCP input port must be provided"); @@ -598,16 +601,47 @@ void EdiTransport::Open(const std::string& uri) bool EdiTransport::rxPacket() { switch (m_proto) { + case Proto::Unspecified: + { + etiLog.level(warn) << "EDI receiving from uninitialised socket"; + return false; + } case Proto::UDP: { - auto udp_data = m_udp_rx.get_packet_buffer(); - - if (udp_data.empty()) { + Socket::InetAddress received_from; + try { + auto received_packets = m_udp_rx.receive(100); + for (auto rp : received_packets) { + received_from = rp.received_from; + + EdiDecoder::Packet p; + p.buf = move(rp.packetdata); + p.received_on_port = rp.port_received_on; + m_decoder.push_packet(p); + } + return true; + } + catch (const Socket::UDPReceiver::Timeout&) { return false; } + catch (const Socket::UDPReceiver::Interrupted&) { + return false; + } + catch (const invalid_argument& e) { + try { + fprintf(stderr, "Invalid argument receiving EDI from %s: %s\n", + received_from.to_string().c_str(), e.what()); + } + catch (const invalid_argument& ee) { + fprintf(stderr, "Invalid argument receiving EDI %s\n", e.what()); + fprintf(stderr, "Invalid argument converting source address %s\n", ee.what()); + } + } + catch (const runtime_error& e) { + fprintf(stderr, "Runtime error UDP Receive: %s\n", e.what()); + } - m_decoder.push_packet(udp_data); - return true; + return false; } case Proto::TCP: { @@ -648,4 +682,3 @@ EdiInput::EdiInput(double& tist_offset_s, float edi_max_delay_ms) : decoder.setMaxDelay(lroundf(edi_max_delay_ms / 24.0f)); } } - diff --git a/src/EtiReader.h b/src/EtiReader.h index be3dd27..d97acf6 100644 --- a/src/EtiReader.h +++ b/src/EtiReader.h @@ -201,13 +201,13 @@ private: }; /* The EDI input does not use the inputs defined in InputReader.h, as they were - * designed for ETI. It uses the EdiTransport which in turn uses a threaded - * receiver. + * designed for ETI. */ class EdiTransport { public: EdiTransport(EdiDecoder::ETIDecoder& decoder); + /* Can be called once when using TCP, or several times when using UDP */ void Open(const std::string& uri); bool isEnabled(void) const { return m_enabled; } @@ -224,8 +224,8 @@ class EdiTransport { std::string m_bindto; std::string m_mcastaddr; - enum class Proto { UDP, TCP }; - Proto m_proto; + enum class Proto { Unspecified, UDP, TCP }; + Proto m_proto = Proto::Unspecified; Socket::UDPReceiver m_udp_rx; std::vector m_tcpbuffer; Socket::TCPClient m_tcpclient; diff --git a/src/InputZeroMQReader.cpp b/src/InputZeroMQReader.cpp index 995e4b0..40a07d4 100644 --- a/src/InputZeroMQReader.cpp +++ b/src/InputZeroMQReader.cpp @@ -198,7 +198,7 @@ void InputZeroMQReader::RecvProcess() continue; } - subscriber.recv(&incoming); + subscriber.recv(incoming); if (queue_size < m_max_queued_frames) { if (incoming.size() < ZMQ_DAB_MESSAGE_T_HEADERSIZE) { diff --git a/src/OutputZeroMQ.cpp b/src/OutputZeroMQ.cpp index 69f4aa1..373081b 100644 --- a/src/OutputZeroMQ.cpp +++ b/src/OutputZeroMQ.cpp @@ -68,10 +68,10 @@ int OutputZeroMQ::process(Buffer* dataIn) if (m_type == ZMQ_REP) { // A ZMQ_REP socket requires a request first zmq::message_t msg; - m_zmq_sock.recv(&msg); + m_zmq_sock.recv(msg, zmq::recv_flags::none); } - m_zmq_sock.send(dataIn->getData(), dataIn->getLength()); + m_zmq_sock.send(zmq::const_buffer{dataIn->getData(), dataIn->getLength()}); return dataIn->getLength(); } diff --git a/src/zmq.hpp b/src/zmq.hpp deleted file mode 100644 index eb5416e..0000000 --- a/src/zmq.hpp +++ /dev/null @@ -1,602 +0,0 @@ -/* - Copyright (c) 2009-2011 250bpm s.r.o. - Copyright (c) 2011 Botond Ballo - Copyright (c) 2007-2009 iMatix Corporation - - Permission is hereby granted, free of charge, to any person obtaining a copy - of this software and associated documentation files (the "Software"), to - deal in the Software without restriction, including without limitation the - rights to use, copy, modify, merge, publish, distribute, sublicense, and/or - sell copies of the Software, and to permit persons to whom the Software is - furnished to do so, subject to the following conditions: - - The above copyright notice and this permission notice shall be included in - all copies or substantial portions of the Software. - - THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR - IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, - FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE - AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER - LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING - FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS - IN THE SOFTWARE. -*/ - -#ifndef __ZMQ_HPP_INCLUDED__ -#define __ZMQ_HPP_INCLUDED__ - -#include - -#include -#include -#include -#include -#include - -// Detect whether the compiler supports C++11 rvalue references. -#if (defined(__GNUC__) && (__GNUC__ > 4 || \ - (__GNUC__ == 4 && __GNUC_MINOR__ > 2)) && \ - defined(__GXX_EXPERIMENTAL_CXX0X__)) - #define ZMQ_HAS_RVALUE_REFS - #define ZMQ_DELETED_FUNCTION = delete -#elif defined(__clang__) - #if __has_feature(cxx_rvalue_references) - #define ZMQ_HAS_RVALUE_REFS - #endif - - #if __has_feature(cxx_deleted_functions) - #define ZMQ_DELETED_FUNCTION = delete - #else - #define ZMQ_DELETED_FUNCTION - #endif -#elif defined(_MSC_VER) && (_MSC_VER >= 1600) - #define ZMQ_HAS_RVALUE_REFS - #define ZMQ_DELETED_FUNCTION -#else - #define ZMQ_DELETED_FUNCTION -#endif - -#if ZMQ_VERSION >= ZMQ_MAKE_VERSION(3, 3, 0) -#define ZMQ_NEW_MONITOR_EVENT_LAYOUT -#endif - -#if ZMQ_VERSION >= ZMQ_MAKE_VERSION(4, 1, 0) -#define ZMQ_HAS_PROXY_STEERABLE -/* Socket event data */ -typedef struct { - uint16_t event; // id of the event as bitfield - int32_t value ; // value is either error code, fd or reconnect interval -} zmq_event_t; -#endif - -// In order to prevent unused variable warnings when building in non-debug -// mode use this macro to make assertions. -#ifndef NDEBUG -# define ZMQ_ASSERT(expression) assert(expression) -#else -# define ZMQ_ASSERT(expression) (void)(expression) -#endif - -namespace zmq -{ - - typedef zmq_free_fn free_fn; - typedef zmq_pollitem_t pollitem_t; - - class error_t : public std::exception - { - public: - - error_t () : errnum (zmq_errno ()) {} - - virtual const char *what () const throw () - { - return zmq_strerror (errnum); - } - - int num () const - { - return errnum; - } - - private: - - int errnum; - }; - - inline int poll (zmq_pollitem_t *items_, int nitems_, long timeout_ = -1) - { - int rc = zmq_poll (items_, nitems_, timeout_); - if (rc < 0) - throw error_t (); - return rc; - } - - inline void proxy (void *frontend, void *backend, void *capture) - { - int rc = zmq_proxy (frontend, backend, capture); - if (rc != 0) - throw error_t (); - } - -#ifdef ZMQ_HAS_PROXY_STEERABLE - inline void proxy_steerable (void *frontend, void *backend, void *capture, void *control) - { - int rc = zmq_proxy_steerable (frontend, backend, capture, control); - if (rc != 0) - throw error_t (); - } -#endif - - inline void version (int *major_, int *minor_, int *patch_) - { - zmq_version (major_, minor_, patch_); - } - - class message_t - { - friend class socket_t; - - public: - - inline message_t () - { - int rc = zmq_msg_init (&msg); - if (rc != 0) - throw error_t (); - } - - inline explicit message_t (size_t size_) - { - int rc = zmq_msg_init_size (&msg, size_); - if (rc != 0) - throw error_t (); - } - - inline message_t (void *data_, size_t size_, free_fn *ffn_, - void *hint_ = NULL) - { - int rc = zmq_msg_init_data (&msg, data_, size_, ffn_, hint_); - if (rc != 0) - throw error_t (); - } - -#ifdef ZMQ_HAS_RVALUE_REFS - inline message_t (message_t &&rhs) : msg (rhs.msg) - { - int rc = zmq_msg_init (&rhs.msg); - if (rc != 0) - throw error_t (); - } - - inline message_t &operator = (message_t &&rhs) - { - std::swap (msg, rhs.msg); - return *this; - } -#endif - - inline ~message_t () - { - int rc = zmq_msg_close (&msg); - ZMQ_ASSERT (rc == 0); - } - - inline void rebuild () - { - int rc = zmq_msg_close (&msg); - if (rc != 0) - throw error_t (); - rc = zmq_msg_init (&msg); - if (rc != 0) - throw error_t (); - } - - inline void rebuild (size_t size_) - { - int rc = zmq_msg_close (&msg); - if (rc != 0) - throw error_t (); - rc = zmq_msg_init_size (&msg, size_); - if (rc != 0) - throw error_t (); - } - - inline void rebuild (void *data_, size_t size_, free_fn *ffn_, - void *hint_ = NULL) - { - int rc = zmq_msg_close (&msg); - if (rc != 0) - throw error_t (); - rc = zmq_msg_init_data (&msg, data_, size_, ffn_, hint_); - if (rc != 0) - throw error_t (); - } - - inline void move (message_t *msg_) - { - int rc = zmq_msg_move (&msg, &(msg_->msg)); - if (rc != 0) - throw error_t (); - } - - inline void copy (message_t *msg_) - { - int rc = zmq_msg_copy (&msg, &(msg_->msg)); - if (rc != 0) - throw error_t (); - } - - inline bool more () - { - int rc = zmq_msg_more (&msg); - return rc != 0; - } - - inline void *data () - { - return zmq_msg_data (&msg); - } - - inline const void* data () const - { - return zmq_msg_data (const_cast(&msg)); - } - - inline size_t size () const - { - return zmq_msg_size (const_cast(&msg)); - } - - private: - - // The underlying message - zmq_msg_t msg; - - // Disable implicit message copying, so that users won't use shared - // messages (less efficient) without being aware of the fact. - message_t (const message_t&); - void operator = (const message_t&); - }; - - class context_t - { - friend class socket_t; - - public: - inline context_t () - { - ptr = zmq_ctx_new (); - if (ptr == NULL) - throw error_t (); - } - - - inline explicit context_t (int io_threads_, int max_sockets_ = ZMQ_MAX_SOCKETS_DFLT) - { - ptr = zmq_ctx_new (); - if (ptr == NULL) - throw error_t (); - - int rc = zmq_ctx_set (ptr, ZMQ_IO_THREADS, io_threads_); - ZMQ_ASSERT (rc == 0); - - rc = zmq_ctx_set (ptr, ZMQ_MAX_SOCKETS, max_sockets_); - ZMQ_ASSERT (rc == 0); - } - -#ifdef ZMQ_HAS_RVALUE_REFS - inline context_t (context_t &&rhs) : ptr (rhs.ptr) - { - rhs.ptr = NULL; - } - inline context_t &operator = (context_t &&rhs) - { - std::swap (ptr, rhs.ptr); - return *this; - } -#endif - - inline ~context_t () - { - close(); - } - - inline void close() - { - if (ptr == NULL) - return; - int rc = zmq_ctx_destroy (ptr); - ZMQ_ASSERT (rc == 0); - ptr = NULL; - } - - // Be careful with this, it's probably only useful for - // using the C api together with an existing C++ api. - // Normally you should never need to use this. - inline operator void* () - { - return ptr; - } - - private: - - void *ptr; - - context_t (const context_t&); - void operator = (const context_t&); - }; - - class socket_t - { - friend class monitor_t; - public: - - inline socket_t (context_t &context_, int type_) - { - ctxptr = context_.ptr; - ptr = zmq_socket (context_.ptr, type_); - if (ptr == NULL) - throw error_t (); - } - -#ifdef ZMQ_HAS_RVALUE_REFS - inline socket_t(socket_t&& rhs) : ptr(rhs.ptr) - { - rhs.ptr = NULL; - } - inline socket_t& operator=(socket_t&& rhs) - { - std::swap(ptr, rhs.ptr); - return *this; - } -#endif - - inline ~socket_t () - { - close(); - } - - inline operator void* () - { - return ptr; - } - - inline void close() - { - if(ptr == NULL) - // already closed - return ; - int rc = zmq_close (ptr); - ZMQ_ASSERT (rc == 0); - ptr = 0 ; - } - - inline void setsockopt (int option_, const void *optval_, - size_t optvallen_) - { - int rc = zmq_setsockopt (ptr, option_, optval_, optvallen_); - if (rc != 0) - throw error_t (); - } - - inline void getsockopt (int option_, void *optval_, - size_t *optvallen_) - { - int rc = zmq_getsockopt (ptr, option_, optval_, optvallen_); - if (rc != 0) - throw error_t (); - } - - inline void bind (const char *addr_) - { - int rc = zmq_bind (ptr, addr_); - if (rc != 0) - throw error_t (); - } - - inline void unbind (const char *addr_) - { - int rc = zmq_unbind (ptr, addr_); - if (rc != 0) - throw error_t (); - } - - inline void connect (const char *addr_) - { - int rc = zmq_connect (ptr, addr_); - if (rc != 0) - throw error_t (); - } - - inline void disconnect (const char *addr_) - { - int rc = zmq_disconnect (ptr, addr_); - if (rc != 0) - throw error_t (); - } - - inline bool connected() - { - return(ptr != NULL); - } - - inline size_t send (const void *buf_, size_t len_, int flags_ = 0) - { - int nbytes = zmq_send (ptr, buf_, len_, flags_); - if (nbytes >= 0) - return (size_t) nbytes; - if (zmq_errno () == EAGAIN) - return 0; - throw error_t (); - } - - inline bool send (message_t &msg_, int flags_ = 0) - { - int nbytes = zmq_msg_send (&(msg_.msg), ptr, flags_); - if (nbytes >= 0) - return true; - if (zmq_errno () == EAGAIN) - return false; - throw error_t (); - } - -#ifdef ZMQ_HAS_RVALUE_REFS - inline bool send (message_t &&msg_, int flags_ = 0) - { - return send(msg_, flags_); - } -#endif - - inline size_t recv (void *buf_, size_t len_, int flags_ = 0) - { - int nbytes = zmq_recv (ptr, buf_, len_, flags_); - if (nbytes >= 0) - return (size_t) nbytes; - if (zmq_errno () == EAGAIN) - return 0; - throw error_t (); - } - - inline bool recv (message_t *msg_, int flags_ = 0) - { - int nbytes = zmq_msg_recv (&(msg_->msg), ptr, flags_); - if (nbytes >= 0) - return true; - if (zmq_errno () == EAGAIN) - return false; - throw error_t (); - } - - private: - void *ptr; - void *ctxptr; - - socket_t (const socket_t&) ZMQ_DELETED_FUNCTION; - void operator = (const socket_t&) ZMQ_DELETED_FUNCTION; - }; - - class monitor_t - { - public: - monitor_t() : socketPtr(NULL) {} - virtual ~monitor_t() {} - - void monitor(socket_t &socket, const char *addr_, int events = ZMQ_EVENT_ALL) - { - int rc = zmq_socket_monitor(socket.ptr, addr_, events); - if (rc != 0) - throw error_t (); - - socketPtr = socket.ptr; - void *s = zmq_socket (socket.ctxptr, ZMQ_PAIR); - assert (s); - - rc = zmq_connect (s, addr_); - assert (rc == 0); - - on_monitor_started(); - - while (true) { - zmq_msg_t eventMsg; - zmq_msg_init (&eventMsg); - rc = zmq_recvmsg (s, &eventMsg, 0); - if (rc == -1 && zmq_errno() == ETERM) - break; - assert (rc != -1); -#if ZMQ_VERSION_MAJOR >= 4 - const char* data = static_cast(zmq_msg_data(&eventMsg)); - zmq_event_t msgEvent; - memcpy(&msgEvent.event, data, sizeof(uint16_t)); data += sizeof(uint16_t); - memcpy(&msgEvent.value, data, sizeof(int32_t)); - zmq_event_t* event = &msgEvent; -#else - zmq_event_t* event = static_cast(zmq_msg_data(&eventMsg)); -#endif - -#ifdef ZMQ_NEW_MONITOR_EVENT_LAYOUT - zmq_msg_t addrMsg; - zmq_msg_init (&addrMsg); - rc = zmq_recvmsg (s, &addrMsg, 0); - if (rc == -1 && zmq_errno() == ETERM) - break; - assert (rc != -1); - const char* str = static_cast(zmq_msg_data (&addrMsg)); - std::string address(str, str + zmq_msg_size(&addrMsg)); - zmq_msg_close (&addrMsg); -#else - // Bit of a hack, but all events in the zmq_event_t union have the same layout so this will work for all event types. - std::string address = event->data.connected.addr; -#endif - -#ifdef ZMQ_EVENT_MONITOR_STOPPED - if (event->event == ZMQ_EVENT_MONITOR_STOPPED) - break; -#endif - - switch (event->event) { - case ZMQ_EVENT_CONNECTED: - on_event_connected(*event, address.c_str()); - break; - case ZMQ_EVENT_CONNECT_DELAYED: - on_event_connect_delayed(*event, address.c_str()); - break; - case ZMQ_EVENT_CONNECT_RETRIED: - on_event_connect_retried(*event, address.c_str()); - break; - case ZMQ_EVENT_LISTENING: - on_event_listening(*event, address.c_str()); - break; - case ZMQ_EVENT_BIND_FAILED: - on_event_bind_failed(*event, address.c_str()); - break; - case ZMQ_EVENT_ACCEPTED: - on_event_accepted(*event, address.c_str()); - break; - case ZMQ_EVENT_ACCEPT_FAILED: - on_event_accept_failed(*event, address.c_str()); - break; - case ZMQ_EVENT_CLOSED: - on_event_closed(*event, address.c_str()); - break; - case ZMQ_EVENT_CLOSE_FAILED: - on_event_close_failed(*event, address.c_str()); - break; - case ZMQ_EVENT_DISCONNECTED: - on_event_disconnected(*event, address.c_str()); - break; - default: - on_event_unknown(*event, address.c_str()); - break; - } - zmq_msg_close (&eventMsg); - } - zmq_close (s); - socketPtr = NULL; - } - -#ifdef ZMQ_EVENT_MONITOR_STOPPED - void abort() - { - if (socketPtr) - zmq_socket_monitor(socketPtr, NULL, 0); - } -#endif - virtual void on_monitor_started() {} - virtual void on_event_connected(const zmq_event_t &event_, const char* addr_) { (void)event_; (void)addr_; } - virtual void on_event_connect_delayed(const zmq_event_t &event_, const char* addr_) { (void)event_; (void)addr_; } - virtual void on_event_connect_retried(const zmq_event_t &event_, const char* addr_) { (void)event_; (void)addr_; } - virtual void on_event_listening(const zmq_event_t &event_, const char* addr_) { (void)event_; (void)addr_; } - virtual void on_event_bind_failed(const zmq_event_t &event_, const char* addr_) { (void)event_; (void)addr_; } - virtual void on_event_accepted(const zmq_event_t &event_, const char* addr_) { (void)event_; (void)addr_; } - virtual void on_event_accept_failed(const zmq_event_t &event_, const char* addr_) { (void)event_; (void)addr_; } - virtual void on_event_closed(const zmq_event_t &event_, const char* addr_) { (void)event_; (void)addr_; } - virtual void on_event_close_failed(const zmq_event_t &event_, const char* addr_) { (void)event_; (void)addr_; } - virtual void on_event_disconnected(const zmq_event_t &event_, const char* addr_) { (void)event_; (void)addr_; } - virtual void on_event_unknown(const zmq_event_t &event_, const char* addr_) { (void)event_; (void)addr_; } - private: - void* socketPtr; - }; -} - -#endif -- cgit v1.2.3