diff options
| -rw-r--r-- | contrib/Socket.cpp | 17 | ||||
| -rw-r--r-- | contrib/Socket.h | 19 | ||||
| -rw-r--r-- | contrib/edioutput/Transport.cpp | 10 |
3 files changed, 41 insertions, 5 deletions
diff --git a/contrib/Socket.cpp b/contrib/Socket.cpp index 2df1559..a85b98b 100644 --- a/contrib/Socket.cpp +++ b/contrib/Socket.cpp @@ -1222,7 +1222,7 @@ TCPSendClient::~TCPSendClient() } } -void TCPSendClient::sendall(const std::vector<uint8_t>& buffer) +TCPSendClient::ErrorStats TCPSendClient::sendall(const std::vector<uint8_t>& buffer) { if (not m_running) { throw runtime_error(m_exception_data); @@ -1234,6 +1234,17 @@ void TCPSendClient::sendall(const std::vector<uint8_t>& buffer) vector<uint8_t> discard; m_queue.try_pop(discard); } + + TCPSendClient::ErrorStats es; + es.num_reconnects = m_num_reconnects.load(); + + es.has_seen_new_errors = es.num_reconnects != m_num_reconnects_prev; + m_num_reconnects_prev = es.num_reconnects; + + auto lock = unique_lock<mutex>(m_error_mutex); + es.last_error = m_last_error; + + return es; } void TCPSendClient::process() @@ -1255,12 +1266,16 @@ void TCPSendClient::process() } else { try { + m_num_reconnects.fetch_add(1, std::memory_order_seq_cst); m_sock.connect(m_hostname, m_port); m_is_connected = true; } catch (const runtime_error& e) { m_is_connected = false; this_thread::sleep_for(chrono::seconds(1)); + + auto lock = unique_lock<mutex>(m_error_mutex); + m_last_error = e.what(); } } } diff --git a/contrib/Socket.h b/contrib/Socket.h index 1320a64..ab2a14a 100644 --- a/contrib/Socket.h +++ b/contrib/Socket.h @@ -329,10 +329,18 @@ class TCPSendClient { public: TCPSendClient(const std::string& hostname, int port); ~TCPSendClient(); + TCPSendClient(const TCPSendClient&) = delete; + TCPSendClient& operator=(const TCPSendClient&) = delete; - /* Throws a runtime_error on error - */ - void sendall(const std::vector<uint8_t>& buffer); + + struct ErrorStats { + std::string last_error = ""; + size_t num_reconnects = 0; + bool has_seen_new_errors = false; + }; + + /* Throws a runtime_error when the process thread isn't running */ + ErrorStats sendall(const std::vector<uint8_t>& buffer); private: void process(); @@ -349,6 +357,11 @@ class TCPSendClient { std::string m_exception_data; std::thread m_sender_thread; TCPSocket m_listener_socket; + + std::atomic<size_t> m_num_reconnects = ATOMIC_VAR_INIT(0); + size_t m_num_reconnects_prev = 0; + std::mutex m_error_mutex; + std::string m_last_error = ""; }; } diff --git a/contrib/edioutput/Transport.cpp b/contrib/edioutput/Transport.cpp index 8ebb9fc..cbff6d4 100644 --- a/contrib/edioutput/Transport.cpp +++ b/contrib/edioutput/Transport.cpp @@ -193,7 +193,15 @@ void Sender::write(const AFPacket& af_packet) 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); + const auto error_stats = tcp_senders.at(tcp_dest.get())->sendall(af_packet); + + if (error_stats.has_seen_new_errors) { + fprintf(stderr, "TCP output %s:%d has %zu reconnects: most recent error: %s\n", + tcp_dest->dest_addr.c_str(), + tcp_dest->dest_port, + error_stats.num_reconnects, + error_stats.last_error.c_str()); + } } else { throw logic_error("EDI destination not implemented"); |