From e1e83dfba6aabcbdb3b0d8752a932b67ac432c5c Mon Sep 17 00:00:00 2001 From: Emiel Bruijntjes Date: Mon, 20 Sep 2021 12:47:56 +0200 Subject: [PATCH 1/2] when a very short heartbeat timer is installed, the timer did not expire soon enough (the default 60-second timeout was still used for the timer) PLUS when we detect a heartbeat-timeout, we now close the connection with immediate effect (because we do not trust the regular AMQP handshake to do anything, because it is also not respecting heartbeats --- include/amqpcpp/libev.h | 13 +++++++++++-- 1 file changed, 11 insertions(+), 2 deletions(-) diff --git a/include/amqpcpp/libev.h b/include/amqpcpp/libev.h index 76eea2f..17c6c34 100644 --- a/include/amqpcpp/libev.h +++ b/include/amqpcpp/libev.h @@ -255,8 +255,9 @@ private: // the server was inactive for a too long period of time, reset state _next = _expire = 0.0; _timeout = 0; - // close the connection because server was inactive - return (void)_connection->close(); + // close the connection because server was inactive (we close it with immediate effect, + // because it was inactive so we cannot trust it to respect the AMQP close handshake) + return (void)_connection->close(true); } else if (now >= _next) { @@ -365,8 +366,16 @@ private: // because the server has just sent us some data, we will update the expire time too _expire = now + _timeout * 1.5; + // stop the existing timer (we have to stop it and restart it, because ev_timer_set() + // on its own does not change the running timer) (note that we assume that the timer + // is already running and keeps on running, so no calls to ev_ref()/en_unref() here) + ev_timer_stop(_loop, &_timer); + // find the earliest thing that expires ev_timer_set(&_timer, std::min(_next, _expire) - now, 0.0); + + // and start it again + ev_timer_start(_loop, &_timer); // expose the accepted interval return _timeout; From 0f596d7a6e5766ab60e05e437a01f58a724950b8 Mon Sep 17 00:00:00 2001 From: Emiel Bruijntjes Date: Mon, 20 Sep 2021 12:48:30 +0200 Subject: [PATCH 2/2] added KEEPALIVE option to the TCP socket so that at least we are notified if the peer turns out to be non-connected --- src/linux_tcp/tcpresolver.h | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/src/linux_tcp/tcpresolver.h b/src/linux_tcp/tcpresolver.h index d4af8de..0b0801c 100644 --- a/src/linux_tcp/tcpresolver.h +++ b/src/linux_tcp/tcpresolver.h @@ -120,6 +120,12 @@ private: // turn socket into a non-blocking socket and set the close-on-exec bit fcntl(_socket, F_SETFL, O_NONBLOCK | O_CLOEXEC); + + // we set the 'keepalive' option so that we automatically detect if the peer is dead + int keepalive = 1; + + // set the keepalive option + setsockopt(_socket, SOL_SOCKET, SO_KEEPALIVE, &keepalive, sizeof(keepalive)); // try to connect non-blocking if (connect(_socket, addresses[i]->ai_addr, addresses[i]->ai_addrlen) == 0) break;