diff --git a/src/rmq/rmqa/rmqa_rabbitcontextimpl.cpp b/src/rmq/rmqa/rmqa_rabbitcontextimpl.cpp index d84fff9..3ffc5b6 100644 --- a/src/rmq/rmqa/rmqa_rabbitcontextimpl.cpp +++ b/src/rmq/rmqa/rmqa_rabbitcontextimpl.cpp @@ -214,6 +214,7 @@ RabbitContextImpl::RabbitContextImpl( d_connectionMonitor, options.clientProperties(), options.connectionErrorThreshold(), + options.connectionEstablishmentTimeout(), isHostHealthMonitoringEnabled); if (!d_threadPool) { diff --git a/src/rmq/rmqa/rmqa_rabbitcontextoptions.cpp b/src/rmq/rmqa/rmqa_rabbitcontextoptions.cpp index 1402125..8c1810a 100644 --- a/src/rmq/rmqa/rmqa_rabbitcontextoptions.cpp +++ b/src/rmq/rmqa/rmqa_rabbitcontextoptions.cpp @@ -21,9 +21,14 @@ #include #include #include +#include +#include namespace BloombergLP { namespace rmqa { + +BALL_LOG_SET_NAMESPACE_CATEGORY("RMQA.RABBITCONTEXTOPTIONS") + namespace { void populateUsefulInformation(rmqt::FieldTable* propertiesPtr) @@ -66,6 +71,7 @@ RabbitContextOptions::RabbitContextOptions() , d_messageProcessingTimeout(DEFAULT_MESSAGE_PROCESSING_TIMEOUT) , d_tunables() , d_connectionErrorThreshold() +, d_connectionEstablishmentTimeout() , d_shuffleConnectionEndpoints() , d_hostHealthConfig() { @@ -115,6 +121,30 @@ RabbitContextOptions& RabbitContextOptions::setConnectionErrorThreshold( return *this; } +RabbitContextOptions& RabbitContextOptions::setConnectionEstablishmentTimeout( + const bsl::optional& timeout) +{ + // The hung timer applies this at millisecond granularity (it uses + // bsls::TimeInterval::totalMilliseconds()), so any value below 1ms -- + // including a positive sub-millisecond interval -- rounds down to zero and + // would expire the bound immediately on every attempt, producing a + // connect -> hung -> retry loop that never establishes. Require at least + // 1ms rather than merely positive; log as well as assert so the + // misconfiguration is diagnosable even in builds where assertions are + // compiled out. + if (timeout.has_value() && *timeout < bsls::TimeInterval(0, 1000 * 1000)) { + BALL_LOG_ERROR << "setConnectionEstablishmentTimeout given " + << (timeout->totalSecondsAsDouble() * 1000.0) + << "ms; values below 1ms round down to zero at the " + "timer's millisecond granularity and prevent any " + "connection from establishing"; + } + BSLS_ASSERT(!timeout.has_value() || + *timeout >= bsls::TimeInterval(0, 1000 * 1000)); + d_connectionEstablishmentTimeout = timeout; + return *this; +} + RabbitContextOptions& RabbitContextOptions::setConsumerTracing( const bsl::shared_ptr& consumerTracing) { diff --git a/src/rmq/rmqa/rmqa_rabbitcontextoptions.h b/src/rmq/rmqa/rmqa_rabbitcontextoptions.h index 4ab70eb..d382e7f 100644 --- a/src/rmq/rmqa/rmqa_rabbitcontextoptions.h +++ b/src/rmq/rmqa/rmqa_rabbitcontextoptions.h @@ -113,6 +113,27 @@ class RabbitContextOptions { RabbitContextOptions& setConnectionErrorThreshold( const bsl::optional& timeout); + /// \brief Set the maximum time a single connection-establishment attempt + /// (TCP connect + TLS handshake + AMQP handshake) may take before it is + /// aborted and retried. The library already bounds establishment with a + /// default; this makes that bound configurable. + /// \param timeout the establishment timeout, as a `bsls::TimeInterval` so + /// that any resolution (down to nanoseconds) can be expressed -- e.g. + /// `bsls::TimeInterval(5.0)` for five seconds, or + /// `bsls::TimeInterval(0, 500 * 1000 * 1000)` for 500 milliseconds. If not + /// set (`bsl::nullopt`), the library default is used. This only ever + /// applies to the establishment phase -- it is cancelled once a connection + /// is established -- so it does not affect long-lived connections. Lowering + /// it makes the client give up on a stalled attempt and retry sooner; + /// raising it tolerates slower networks/handshakes. + /// \note The timeout is applied at millisecond granularity: the behavior + /// is undefined unless `timeout`, when set, is at least one millisecond. A + /// smaller (sub-millisecond) interval rounds down to zero, which would + /// expire the bound immediately and prevent any connection from + /// establishing. + RabbitContextOptions& setConnectionEstablishmentTimeout( + const bsl::optional& timeout); + /// \brief will be called back to create a context which spans for the /// lifetime of the messageguard _before_ it is passed to its consumer /// message processor if there has @@ -183,6 +204,12 @@ class RabbitContextOptions { return d_connectionErrorThreshold; } + const bsl::optional& + connectionEstablishmentTimeout() const + { + return d_connectionEstablishmentTimeout; + } + const rmqt::Tunables& tunables() const { return d_tunables; } const bsl::shared_ptr& consumerTracing() const @@ -221,6 +248,7 @@ class RabbitContextOptions { bsls::TimeInterval d_messageProcessingTimeout; rmqt::Tunables d_tunables; bsl::optional d_connectionErrorThreshold; + bsl::optional d_connectionEstablishmentTimeout; bsl::shared_ptr d_consumerTracing; bsl::shared_ptr d_producerTracing; bsl::optional d_shuffleConnectionEndpoints; diff --git a/src/rmq/rmqamqp/rmqamqp_connection.cpp b/src/rmq/rmqamqp/rmqamqp_connection.cpp index ac3bc46..9ce9513 100644 --- a/src/rmq/rmqamqp/rmqamqp_connection.cpp +++ b/src/rmq/rmqamqp/rmqamqp_connection.cpp @@ -48,6 +48,7 @@ #include #include #include +#include #include #include #include @@ -80,7 +81,7 @@ const bsl::uint16_t k_MAX_CHANNEL_NUM = bsl::numeric_limits::max(); // k_MAX_HEARTBEAT_TIMEOUT_SEC cannot be zero as the HeartbeatManager does // not support this const bsl::uint16_t k_MAX_HEARTBEAT_TIMEOUT_SEC = 60; -const bsl::uint16_t k_HUNG_TIMER_SEC = 60; +const bsl::uint16_t k_DEFAULT_HUNG_TIMER_SEC = 60; BALL_LOG_SET_NAMESPACE_CATEGORY("RMQAMQP.CONNECTION") @@ -220,6 +221,7 @@ Connection::Connection( const bsl::shared_ptr& endpoint, const bsl::shared_ptr& credentials, const rmqt::FieldTable& clientProperties, + const bsl::optional& connectionEstablishmentTimeout, bsl::string_view name) : d_resolver(resolver) , d_retryHandler(retryHandler) @@ -233,8 +235,9 @@ Connection::Connection( , d_state(Connection::DISCONNECTED) , d_clientProperties(clientProperties) , d_channels() -, d_hungTimer( - timerFactory->createWithTimeout(bsls::TimeInterval(k_HUNG_TIMER_SEC))) +, d_connectionEstablishmentTimeout(connectionEstablishmentTimeout.value_or( + bsls::TimeInterval(k_DEFAULT_HUNG_TIMER_SEC))) +, d_hungTimer(timerFactory->createWithTimeout(d_connectionEstablishmentTimeout)) , d_timerFactory(timerFactory) , d_firstConnectCb() , d_closeCb() @@ -601,8 +604,32 @@ void Connection::socketShutdown(DisconnectType dcType, d_state = Connection::DISCONNECTED; BALL_LOG_TRACE << "State now set to: " << d_state; if (dcType == WILL_RETRY) { + // The connection is down and will be retried: either an establishment + // attempt failed or an established connection was lost. Schedule the + // retry before invoking the observational hook: the hook is a + // caller-supplied override, and if it were to throw, the retry must + // already have been scheduled so the connection is not left dead. + // Ordering is otherwise immaterial -- retry only arms a timer, so the + // next initiateConnect() (which reads the endpoint) runs well after + // this returns. d_retryHandler->retry( bdlf::BindUtil::bind(&Connection::retry, weak_from_this())); + // The hook is a caller-supplied override; catch any exception so it + // cannot propagate into the event loop and terminate the process. The + // retry above is already scheduled, so the connection is unaffected. + try { + d_endpoint->onConnectFailed(); + } + catch (const bsl::exception& e) { + BALL_LOG_ERROR << "rmqt::Endpoint::onConnectFailed threw (" + << e.what() << "); ignoring for " + << connectionDebugName(); + } + catch (...) { + BALL_LOG_ERROR << "rmqt::Endpoint::onConnectFailed threw a " + "non-standard exception; ignoring for " + << connectionDebugName(); + } } else { if (d_firstConnectCb) { @@ -670,6 +697,26 @@ class Connection::ConnectionMethodProcessor { conn.d_firstConnectCb(result); conn.d_firstConnectCb = ConnectedCallback(); } + + // Invoke the observational hook last: it is a caller-supplied + // override, so the critical connect path (timer cancel, retry + // success, channel open, connect callback) must complete first. By + // this point the connection is fully established, and any exception + // is caught so it cannot propagate into the event loop and + // terminate the process. + try { + conn.d_endpoint->onConnectSuccess(); + } + catch (const bsl::exception& e) { + BALL_LOG_ERROR << "rmqt::Endpoint::onConnectSuccess threw (" + << e.what() << "); ignoring for " + << conn.connectionDebugName(); + } + catch (...) { + BALL_LOG_ERROR << "rmqt::Endpoint::onConnectSuccess threw a " + "non-standard exception; ignoring for " + << conn.connectionDebugName(); + } } } @@ -1010,9 +1057,9 @@ void Connection::connectionHung(rmqio::Timer::InterruptReason reason) d_metricPublisher->publishCounter( "hung_connection_reset", 1, d_vhostTags); - BALL_LOG_FATAL << "Timed out after " << k_HUNG_TIMER_SEC - << " seconds while connecting: " - << connectionDebugName(); + BALL_LOG_ERROR << "Timed out after " + << d_connectionEstablishmentTimeout.totalMilliseconds() + << " ms while connecting: " << connectionDebugName(); closeSocket(WILL_RETRY); } } @@ -1056,6 +1103,7 @@ Connection::Factory::Factory( const bsl::shared_ptr& connectionMonitor, const rmqt::FieldTable& clientProperties, const bsl::optional& connectionErrorThreshold, + const bsl::optional& connectionEstablishmentTimeout, const bool isHostHealthMonitoringEnabled) : d_errorCb(errorCb) , d_clientProperties(clientProperties) @@ -1064,6 +1112,7 @@ Connection::Factory::Factory( , d_timerFactory(timerFactory) , d_connectionMonitor(connectionMonitor) , d_connectionErrorThreshold(connectionErrorThreshold) +, d_connectionEstablishmentTimeout(connectionEstablishmentTimeout) , d_isHostHealthMonitoringEnabled(isHostHealthMonitoringEnabled) { } @@ -1083,6 +1132,7 @@ bsl::shared_ptr Connection::Factory::create( endpoint, credentials, generateClientProperties(d_clientProperties, name), + d_connectionEstablishmentTimeout, name)); d_connectionMonitor->addConnection(bsl::weak_ptr(result)); diff --git a/src/rmq/rmqamqp/rmqamqp_connection.h b/src/rmq/rmqamqp/rmqamqp_connection.h index f5dd857..fca1916 100644 --- a/src/rmq/rmqamqp/rmqamqp_connection.h +++ b/src/rmq/rmqamqp/rmqamqp_connection.h @@ -172,16 +172,18 @@ class Connection : public bsl::enable_shared_from_this, /// \param endpoint references the vhost this connection is /// joining /// \param credentials Credentials used to connect to the broker - Connection(const bsl::shared_ptr& resolver, - const bsl::shared_ptr& retryHandler, - const bsl::shared_ptr& hbManager, - const bsl::shared_ptr& timerFactory, - const bsl::shared_ptr& channelFactory, - const bsl::shared_ptr& metricPublisher, - const bsl::shared_ptr& endpoint, - const bsl::shared_ptr& credentials, - const rmqt::FieldTable& clientProperties, - bsl::string_view name); + Connection( + const bsl::shared_ptr& resolver, + const bsl::shared_ptr& retryHandler, + const bsl::shared_ptr& hbManager, + const bsl::shared_ptr& timerFactory, + const bsl::shared_ptr& channelFactory, + const bsl::shared_ptr& metricPublisher, + const bsl::shared_ptr& endpoint, + const bsl::shared_ptr& credentials, + const rmqt::FieldTable& clientProperties, + const bsl::optional& connectionEstablishmentTimeout, + bsl::string_view name); private: bsl::shared_ptr d_resolver; @@ -199,6 +201,7 @@ class Connection : public bsl::enable_shared_from_this, State d_state; rmqt::FieldTable d_clientProperties; ChannelMap d_channels; + bsls::TimeInterval d_connectionEstablishmentTimeout; bsl::shared_ptr d_hungTimer; bsl::shared_ptr d_timerFactory; @@ -318,14 +321,16 @@ class Connection::Factory { public: // CREATORS - Factory(const bsl::shared_ptr& resolver, - const bsl::shared_ptr& timerFactory, - const rmqt::ErrorCallback& errorCb, - const bsl::shared_ptr& metricPublisher, - const bsl::shared_ptr& connectionMonitor, - const rmqt::FieldTable& clientProperties, - const bsl::optional& connectionErrorThreshold, - const bool isHostHealthMonitoringEnabled); + Factory( + const bsl::shared_ptr& resolver, + const bsl::shared_ptr& timerFactory, + const rmqt::ErrorCallback& errorCb, + const bsl::shared_ptr& metricPublisher, + const bsl::shared_ptr& connectionMonitor, + const rmqt::FieldTable& clientProperties, + const bsl::optional& connectionErrorThreshold, + const bsl::optional& connectionEstablishmentTimeout, + const bool isHostHealthMonitoringEnabled); virtual ~Factory() {} @@ -350,6 +355,7 @@ class Connection::Factory { const bsl::shared_ptr d_timerFactory; const bsl::shared_ptr d_connectionMonitor; const bsl::optional d_connectionErrorThreshold; + const bsl::optional d_connectionEstablishmentTimeout; const bool d_isHostHealthMonitoringEnabled; }; // class Connection::Factory } // namespace rmqamqp diff --git a/src/rmq/rmqt/rmqt_endpoint.h b/src/rmq/rmqt/rmqt_endpoint.h index 7705273..94c66e4 100644 --- a/src/rmq/rmqt/rmqt_endpoint.h +++ b/src/rmq/rmqt/rmqt_endpoint.h @@ -36,6 +36,37 @@ class Endpoint { virtual bsl::uint16_t port() const = 0; virtual bsl::shared_ptr securityParameters() const; + + /// \brief Invoked each time an AMQP connection to this endpoint is + /// established (Connection.Open-Ok received). Defaults to a no-op; + /// implementations may override it to observe the connection lifecycle, + /// e.g. for monitoring or metrics. + /// + /// \warning Invoked synchronously on the connection's event-loop thread + /// (the same context as the connection's other callbacks). Overrides must + /// be cheap and non-blocking and must not call back into the connection. + /// A single `Endpoint` may be shared (via `shared_ptr`) by more than one + /// connection, and those connections may run on different event-loop + /// threads; an override must therefore synchronise any state it mutates. + virtual void onConnectSuccess() {} + + /// \brief Invoked each time the connection to this endpoint is not + /// established and will be retried -- either an establishment attempt + /// failed, or a previously-established connection was lost and will + /// reconnect. Defaults to a no-op; implementations may override it to + /// observe the connection lifecycle, e.g. for monitoring or metrics. + /// + /// \warning Invoked synchronously on the connection's event-loop thread + /// (the same context as the connection's other callbacks). Overrides must + /// be cheap and non-blocking and must not call back into the connection. + /// A single `Endpoint` may be shared (via `shared_ptr`) by more than one + /// connection, and those connections may run on different event-loop + /// threads; an override must therefore synchronise any state it mutates. + /// + /// \note Only invoked when the connection will be retried. It is not + /// invoked on a terminal failure (e.g. when the connection error threshold + /// is breached and the failure is surfaced to the application). + virtual void onConnectFailed() {} }; } // namespace rmqt diff --git a/src/tests/rmqa/rmqa_connectionimpl.t.cpp b/src/tests/rmqa/rmqa_connectionimpl.t.cpp index b9ab6ee..d3238f0 100644 --- a/src/tests/rmqa/rmqa_connectionimpl.t.cpp +++ b/src/tests/rmqa/rmqa_connectionimpl.t.cpp @@ -73,6 +73,7 @@ class MockConnection : public rmqamqp::Connection { endpoint, credentials, clientProperties, + bsl::optional(), "Connection Name") , d_ackQueue(bsl::make_shared()) , d_receiveChannel( diff --git a/src/tests/rmqa/rmqa_rabbitcontextoptions.t.cpp b/src/tests/rmqa/rmqa_rabbitcontextoptions.t.cpp index be5a477..eabd469 100644 --- a/src/tests/rmqa/rmqa_rabbitcontextoptions.t.cpp +++ b/src/tests/rmqa/rmqa_rabbitcontextoptions.t.cpp @@ -17,6 +17,11 @@ #include +#include +#include + +#include + #include using namespace BloombergLP; @@ -25,6 +30,11 @@ using namespace ::testing; namespace { bool alwaysHealthy() { return true; } + +/// `bsls_asserttest`'s `BSLS_ASSERTTEST_ASSERT_{PASS,FAIL}` macros report their +/// outcome by calling a driver-supplied `ASSERT(bool)` (which its header allows +/// to be a function, not just a macro). Route that into a gtest expectation. +void ASSERT(bool result) { EXPECT_TRUE(result); } } // namespace TEST(RabbitContextOptions, Constructs) { rmqa::RabbitContextOptions t; } @@ -53,3 +63,48 @@ TEST(RabbitContextOptions, SetHostHealthConfig) // Now it should be set EXPECT_TRUE(options.hostHealthConfig().has_value()); } + +TEST(RabbitContextOptions, SetConnectionEstablishmentTimeoutAcceptsValid) +{ + rmqa::RabbitContextOptions options; + bsls::AssertTestHandlerGuard guard; + + // Unset is allowed (falls back to the library default). + BSLS_ASSERTTEST_ASSERT_PASS( + options.setConnectionEstablishmentTimeout(bsl::nullopt)); + + // A whole-second positive value is allowed. + BSLS_ASSERTTEST_ASSERT_PASS( + options.setConnectionEstablishmentTimeout(bsls::TimeInterval(5, 0))); + EXPECT_EQ(options.connectionEstablishmentTimeout(), + bsls::TimeInterval(5, 0)); + + // A sub-second value must be accepted (500ms), as must the 1ms granularity + // floor the hung timer supports. + BSLS_ASSERTTEST_ASSERT_PASS(options.setConnectionEstablishmentTimeout( + bsls::TimeInterval(0, 500 * 1000 * 1000))); + EXPECT_EQ(options.connectionEstablishmentTimeout(), + bsls::TimeInterval(0, 500 * 1000 * 1000)); + + BSLS_ASSERTTEST_ASSERT_PASS(options.setConnectionEstablishmentTimeout( + bsls::TimeInterval(0, 1000 * 1000))); // exactly 1ms +} + +TEST(RabbitContextOptions, SetConnectionEstablishmentTimeoutRejectsTooSmall) +{ + rmqa::RabbitContextOptions options; + bsls::AssertTestHandlerGuard guard; + + // Zero would expire the establishment bound immediately. + BSLS_ASSERTTEST_ASSERT_FAIL( + options.setConnectionEstablishmentTimeout(bsls::TimeInterval(0, 0))); + + // Negative is likewise invalid. + BSLS_ASSERTTEST_ASSERT_FAIL( + options.setConnectionEstablishmentTimeout(bsls::TimeInterval(-1, 0))); + + // A positive but sub-millisecond interval (500us) rounds down to 0ms in the + // hung timer, so it is rejected too. + BSLS_ASSERTTEST_ASSERT_FAIL(options.setConnectionEstablishmentTimeout( + bsls::TimeInterval(0, 500 * 1000))); +} diff --git a/src/tests/rmqamqp/rmqamqp_connection.t.cpp b/src/tests/rmqamqp/rmqamqp_connection.t.cpp index 23f75bf..e9ad783 100644 --- a/src/tests/rmqamqp/rmqamqp_connection.t.cpp +++ b/src/tests/rmqamqp/rmqamqp_connection.t.cpp @@ -319,6 +319,17 @@ class MockConnectionMonitor : public ConnectionMonitor { void addConnection(const bsl::weak_ptr&) {} }; +class MockEndpoint : public rmqt::SimpleEndpoint { + public: + MockEndpoint(const bsl::string& address, const bsl::string& vhost) + : rmqt::SimpleEndpoint(address, vhost) + { + } + + MOCK_METHOD0(onConnectSuccess, void()); + MOCK_METHOD0(onConnectFailed, void()); +}; + class ConnectionFactory : public rmqamqp::Connection::Factory { const bsl::shared_ptr d_retryHandler; const bsl::shared_ptr d_hbManager; @@ -333,7 +344,9 @@ class ConnectionFactory : public rmqamqp::Connection::Factory { const rmqt::FieldTable& clientProperties, const bsl::shared_ptr& retryHandler, const bsl::shared_ptr& hbManager, - const bsl::shared_ptr& channelFactory) + const bsl::shared_ptr& channelFactory, + const bsl::optional& establishmentTimeout = + bsl::optional()) : rmqamqp::Connection::Factory(resolver, timerFactory, errorCb, @@ -341,6 +354,7 @@ class ConnectionFactory : public rmqamqp::Connection::Factory { bsl::make_shared(), clientProperties, bsls::TimeInterval(), + establishmentTimeout, false) , d_retryHandler(retryHandler) , d_hbManager(hbManager) @@ -1698,3 +1712,207 @@ TEST_F(ConnectionHungTests, HungConnectionSendsMetric) // 3. Process shutdown d_eventLoop.run(); } + +// Tests for the Endpoint connection-lifecycle hooks (onConnectSuccess / +// onConnectFailed) and the configurable connection-establishment timeout. +class ConnectionHookTests : public ConnectionTests {}; + +TEST_F(ConnectionHookTests, OnConnectSuccessFiresOnHandshake) +{ + bsl::shared_ptr endpoint = + bsl::make_shared("127.0.0.1", TEST_VHOST); + EXPECT_CALL(*endpoint, onConnectSuccess()).Times(1); + EXPECT_CALL(*endpoint, onConnectFailed()).Times(0); + + expectFirstHandshakeFrames(); + + { + bsl::shared_ptr conn = + d_factory->create(endpoint, d_credentials, "test-connection"); + conn->startFirstConnection(d_onConnectCb); + + d_eventLoop.run(); + + expectShutdownCalls(); + } + d_eventLoop.run(); +} + +TEST_F(ConnectionHookTests, OnConnectFailedThenSuccessAcrossReconnect) +{ + // Models the sequence where an establishment attempt stalls, the hung + // timer fires (onConnectFailed), the connection retries and then succeeds + // (onConnectSuccess). + bsl::shared_ptr endpoint = + bsl::make_shared("127.0.0.1", TEST_VHOST); + EXPECT_CALL(*endpoint, onConnectFailed()).Times(AtLeast(1)); + EXPECT_CALL(*endpoint, onConnectSuccess()).Times(1); + + expectHeaderAndStartFrames(); + + { + bsl::shared_ptr conn = + d_factory->create(endpoint, d_credentials, "test-connection"); + conn->startFirstConnection(d_onConnectCb); + + // 1. Partial handshake (stalls before Tune/Open) + d_eventLoop.run(); + d_eventLoop.restart(); + + resetExpectations(); // retry -> reconnect + expectFirstHandshakeFrames(); + + // 2. Hung timer fires -> onConnectFailed -> retry -> reconnect succeeds + d_timerFactory->step_time(bsls::TimeInterval(120)); + d_eventLoop.run(); + d_eventLoop.restart(); + + expectShutdownCalls(); + } + d_eventLoop.run(); +} + +TEST_F(ConnectionHookTests, ConfigurableEstablishmentTimeoutIsHonoured) +{ + // A short (5s) establishment timeout must fire the hung timer well before + // the 60s default -- proving the configured value is threaded through. + d_factory = bsl::make_shared(d_resolver, + d_timerFactory, + d_errorCallback, + d_metricPublisher, + d_clientProperties, + d_retryHandler, + d_heartbeat, + d_channelFactory, + bsls::TimeInterval(5)); + + bsl::shared_ptr endpoint = + bsl::make_shared("127.0.0.1", TEST_VHOST); + EXPECT_CALL(*endpoint, onConnectFailed()).Times(AtLeast(1)); + EXPECT_CALL(*endpoint, onConnectSuccess()).Times(1); + + expectHeaderAndStartFrames(); + + { + bsl::shared_ptr conn = + d_factory->create(endpoint, d_credentials, "test-connection"); + conn->startFirstConnection(d_onConnectCb); + + d_eventLoop.run(); + d_eventLoop.restart(); + + resetExpectations(); + expectFirstHandshakeFrames(); + + // Only 6s elapse -- enough for a 5s timeout, far short of the 60s + // default -- yet the hung timer fires and drives the failure hook. + d_timerFactory->step_time(bsls::TimeInterval(6)); + d_eventLoop.run(); + d_eventLoop.restart(); + + expectShutdownCalls(); + } + d_eventLoop.run(); +} + +TEST_F(ConnectionHookTests, HooksFireAcrossServerInitiatedReconnect) +{ + // A server-initiated close after a successful connection is a retriable + // failure: onConnectFailed fires even though the connection had already + // succeeded, and onConnectSuccess fires again on the reconnect. This pins + // the symmetric "each establishment / each retriable failure" contract. + bsl::shared_ptr endpoint = + bsl::make_shared("127.0.0.1", TEST_VHOST); + EXPECT_CALL(*endpoint, onConnectSuccess()).Times(2); + EXPECT_CALL(*endpoint, onConnectFailed()).Times(1); + + expectFirstHandshakeFrames(); + + { + bsl::shared_ptr conn = + d_factory->create(endpoint, d_credentials, "test-connection"); + conn->startFirstConnection(d_onConnectCb); + + // 1. Establish the connection. + d_eventLoop.run(); + d_eventLoop.restart(); + + expectCloseFrame(); + expectCloseOkFrame(); + resetExpectations(); + expectHandshakeFrames(); + + feedNextFrame(); // Queue the server Close frame + + // 2. Server closes -> onConnectFailed -> reconnect -> onConnectSuccess. + d_eventLoop.run(); + d_eventLoop.restart(); + + EXPECT_THAT(conn->state(), Eq(rmqamqp::Connection::CONNECTED)); + expectShutdownCalls(); + } + d_eventLoop.run(); +} + +TEST_F(ConnectionHookTests, OnConnectSuccessExceptionIsContained) +{ + // A throwing onConnectSuccess override must not escape into the event loop + // (which would terminate the process). The connection still establishes -- + // the hook runs after onConnect(true) -- and the exception is swallowed. + bsl::shared_ptr endpoint = + bsl::make_shared("127.0.0.1", TEST_VHOST); + EXPECT_CALL(*endpoint, onConnectSuccess()) + .WillOnce(Throw(bsl::runtime_error("boom"))); + EXPECT_CALL(*endpoint, onConnectFailed()).Times(0); + + expectFirstHandshakeFrames(); // asserts onConnect(true) still fires + + { + bsl::shared_ptr conn = + d_factory->create(endpoint, d_credentials, "test-connection"); + conn->startFirstConnection(d_onConnectCb); + + d_eventLoop.run(); + + EXPECT_THAT(conn->state(), Eq(rmqamqp::Connection::CONNECTED)); + expectShutdownCalls(); + } + d_eventLoop.run(); +} + +TEST_F(ConnectionHookTests, OnConnectFailedExceptionIsContained) +{ + // A throwing onConnectFailed override must not escape into the event loop, + // and must not prevent the retry (which is scheduled before the hook): the + // connection still reconnects and onConnectSuccess fires. + bsl::shared_ptr endpoint = + bsl::make_shared("127.0.0.1", TEST_VHOST); + EXPECT_CALL(*endpoint, onConnectFailed()) + .WillOnce(Throw(bsl::runtime_error("boom"))) + .WillRepeatedly(Return()); + EXPECT_CALL(*endpoint, onConnectSuccess()).Times(1); + + expectHeaderAndStartFrames(); + + { + bsl::shared_ptr conn = + d_factory->create(endpoint, d_credentials, "test-connection"); + conn->startFirstConnection(d_onConnectCb); + + // 1. Partial handshake (stalls before Tune/Open). + d_eventLoop.run(); + d_eventLoop.restart(); + + resetExpectations(); // retry -> reconnect + expectFirstHandshakeFrames(); + + // 2. Hung timer fires -> onConnectFailed throws (contained) -> retry -> + // reconnect succeeds. + d_timerFactory->step_time(bsls::TimeInterval(120)); + d_eventLoop.run(); + d_eventLoop.restart(); + + expectShutdownCalls(); + } + d_eventLoop.run(); +} diff --git a/src/tests/rmqamqp/rmqamqp_hosthealthmonitor.t.cpp b/src/tests/rmqamqp/rmqamqp_hosthealthmonitor.t.cpp index f8df67d..5e172c5 100644 --- a/src/tests/rmqamqp/rmqamqp_hosthealthmonitor.t.cpp +++ b/src/tests/rmqamqp/rmqamqp_hosthealthmonitor.t.cpp @@ -64,6 +64,7 @@ class MockConnection : public rmqamqp::Connection { endpoint, credentials, clientProperties, + bsl::optional(), connectionName) { }