Reconnect on all transport layer errors and log them.

Summary: Ref #38

Reviewers: ivica

Reviewed By: ivica

Subscribers: miljen

Differential Revision: https://repo.mireo.local/D37743

Author: biljazovic

doc/qbk/reference/concepts/LoggerType.qbk

@@ -72,6 +72,13 @@ A type satisfies the `LoggerType` concept if it defines [*any number (including
         ]
         [Invoked when the __DISCONNECT__ packet is received, indicating that the Broker wants to close this connection. ]
     ]
+    [
+        [`void at_transport_error(error_code ec);`]
+        [
+            [*`ec`] is the `error_code` returned by the `async_read_some` or `async_write_some` operation on the underlying __StreamType__ stream.
+        ]
+        [Invoked when a read or write operation on the underlying __StreamType__ stream fails, triggering a reconnection to the broker.]
+    ]
 ]
 
 For example, a type `T` that defines `at_connack` and `at_disconnect` functions with their respective arguments is considered a valid `LoggerType`.

include/boost/mqtt5/detail/log_invoke.hpp

@@ -68,6 +68,10 @@ class log_invoke {
             _logger.at_disconnect(rc, dc_props);
     }
 
+    void at_transport_error(error_code ec) {
+        if constexpr (has_at_transport_error<LoggerType>)
+            _logger.at_transport_error(ec);
+    }
 };
 
 } // end namespace boost::mqtt5::detail

include/boost/mqtt5/impl/connect_op.hpp

@@ -240,8 +240,10 @@ class connect_op {
         if (is_cancelled())
             return complete(asio::error::operation_aborted);
 
-        if (ec)
+        if (ec) {
+            _log.at_transport_error(ec);
             return do_shutdown(ec);
+        }
 
         _buffer_ptr = std::make_unique<std::string>(min_packet_sz, char(0));
 
@@ -258,8 +260,10 @@ class connect_op {
         if (is_cancelled())
             return complete(asio::error::operation_aborted);
 
-        if (ec)
+        if (ec) {
+            _log.at_transport_error(ec);
             return do_shutdown(ec);
+        }
 
         auto code = control_code_e((*_buffer_ptr)[0] & 0b11110000);
 
@@ -301,8 +305,10 @@ class connect_op {
         if (is_cancelled())
             return complete(asio::error::operation_aborted);
 
-        if (ec)
+        if (ec) {
+            _log.at_transport_error(ec);
             return do_shutdown(ec);
+        }
 
         if (code == control_code_e::connack)
             return on_connack(first, last);
@@ -402,8 +408,10 @@ class connect_op {
         if (is_cancelled())
             return complete(asio::error::operation_aborted);
 
-        if (ec)
+        if (ec) {
+            _log.at_transport_error(ec);
             return do_shutdown(ec);
+        }
 
         auto buff = asio::buffer(_buffer_ptr->data(), min_packet_sz);
         asio::async_read(

include/boost/mqtt5/impl/read_op.hpp

@@ -76,13 +76,8 @@ class read_op {
             );
         }
         else
-            asio::post(
-                _owner.get_executor(),
-                asio::prepend(
-                    std::move(*this), on_read {}, stream_ptr,
-                    std::array<size_t, 2> { 0, 1 },
-                    asio::error::not_connected, 0, error_code {}
-                )
+            _owner.async_reconnect(
+                stream_ptr, asio::prepend(std::move(*this), on_reconnect {})
             );
     }
 
@@ -100,13 +95,10 @@ class read_op {
         if (!ec)
             return complete(ec, bytes_read);
 
-        // websocket returns operation_aborted if disconnected
-        if (should_reconnect(ec) || ec == asio::error::operation_aborted)
-            return _owner.async_reconnect(
-                stream_ptr, asio::prepend(std::move(*this), on_reconnect {})
-            );
-
-        return complete(asio::error::no_recovery, bytes_read);
+        _owner.log().at_transport_error(ec);
+        _owner.async_reconnect(
+            stream_ptr, asio::prepend(std::move(*this), on_reconnect {})
+        );
     }
 
     void operator()(on_reconnect, error_code ec) {
@@ -120,17 +112,6 @@ class read_op {
     void complete(error_code ec, size_t bytes_read) {
         std::move(_handler)(ec, bytes_read);
     }
-
-    static bool should_reconnect(error_code ec) {
-        using namespace asio::error;
-        // note: Win ERROR_SEM_TIMEOUT == Posix ENOLINK (Reserved)
-        return ec.value() == 1236L || /* Win ERROR_CONNECTION_ABORTED */
-            ec.value() == 121L || /* Win ERROR_SEM_TIMEOUT */
-            ec == connection_aborted || ec == not_connected ||
-            ec == timed_out || ec == connection_reset ||
-            ec == broken_pipe || ec == asio::error::eof;
-    }
-
 };
 
 

include/boost/mqtt5/impl/write_op.hpp

@@ -61,12 +61,8 @@ class write_op {
                 asio::prepend(std::move(*this), on_write {}, stream_ptr)
             );
         else
-            asio::post(
-                _owner.get_executor(),
-                asio::prepend(
-                    std::move(*this), on_write {},
-                    stream_ptr, asio::error::not_connected, 0
-                )
+            _owner.async_reconnect(
+                stream_ptr, asio::prepend(std::move(*this), on_reconnect {})
             );
     }
 
@@ -80,13 +76,10 @@ class write_op {
         if (!ec)
             return complete(ec, bytes_written);
 
-        // websocket returns operation_aborted if disconnected
-        if (should_reconnect(ec) || ec == asio::error::operation_aborted)
-            return _owner.async_reconnect(
-                stream_ptr, asio::prepend(std::move(*this), on_reconnect {})
-            );
-
-        return complete(asio::error::no_recovery, 0);
+        _owner.log().at_transport_error(ec);
+        _owner.async_reconnect(
+            stream_ptr, asio::prepend(std::move(*this), on_reconnect {})
+        );
     }
 
     void operator()(on_reconnect, error_code ec) {
@@ -100,17 +93,6 @@ class write_op {
     void complete(error_code ec, size_t bytes_written) {
         std::move(_handler)(ec, bytes_written);
     }
-
-    static bool should_reconnect(error_code ec) {
-        using namespace asio::error;
-        // note: Win ERROR_SEM_TIMEOUT == Posix ENOLINK (Reserved)
-        return ec.value() == 1236L || /* Win ERROR_CONNECTION_ABORTED */
-            ec.value() == 121L || /* Win ERROR_SEM_TIMEOUT */
-            ec == connection_aborted || ec == not_connected ||
-            ec == timed_out || ec == connection_reset ||
-            ec == broken_pipe || ec == asio::error::eof;
-    }
-
 };
 
 } // end namespace boost::mqtt5::detail

include/boost/mqtt5/logger.hpp

@@ -198,6 +198,22 @@ class logger {
         std::clog << std::endl;
     }
 
+    /**
+     * \brief Outputs the error message when it occurs at the transport layer
+     * read or write operations.
+     *
+     * \param ec The error code returned by the `async_read_some` or
+     * `async_write_some` operation on the underlying \__StreamType\__ stream.
+     */
+    void at_transport_error(error_code ec) {
+        if (_level < log_level::info)
+            return;
+
+        output_prefix();
+        std::clog
+            << "transport layer error: " << ec.message() << "." << std::endl;
+    }
+
 private:
     void output_prefix() {
         std::clog << prefix << " ";
@@ -251,6 +267,7 @@ static_assert(has_at_tls_handshake<logger>);
 static_assert(has_at_ws_handshake<logger>);
 static_assert(has_at_connack<logger>);
 static_assert(has_at_disconnect<logger>);
+static_assert(has_at_transport_error<logger>);
 
 } // end namespace boost::mqtt5
 

include/boost/mqtt5/logger_traits.hpp

@@ -97,6 +97,15 @@ using at_disconnect_sig = decltype(
 template <typename T>
 constexpr bool has_at_disconnect = boost::is_detected<at_disconnect_sig, T>::value;
 
+// at_transport_error
+
+template <typename T>
+using at_transport_error = decltype(
+    std::declval<T&>().at_transport_error(std::declval<error_code>())
+);
+template <typename T>
+constexpr bool has_at_transport_error = boost::is_detected<at_transport_error, T>::value;
+
 } // end namespace boost::mqtt5
 
 

test/unit/logger.cpp

@@ -40,6 +40,7 @@ void logger_test() {
     BOOST_STATIC_ASSERT(has_at_ws_handshake<logger>);
     BOOST_STATIC_ASSERT(has_at_connack<logger>);
     BOOST_STATIC_ASSERT(has_at_disconnect<logger>);
+    BOOST_STATIC_ASSERT(has_at_transport_error<logger>);
 }
 
 BOOST_AUTO_TEST_SUITE(logger_tests)
@@ -417,6 +418,19 @@ BOOST_FIXTURE_TEST_CASE(at_disconnect_debug, disconnect_test_data) {
     test_logger_output(std::move(test_fun), expected_output);
 }
 
+// at_transport_error
+
+BOOST_AUTO_TEST_CASE(at_transport_error_info) {
+    const auto expected_output = "[Boost.MQTT5] transport layer error: End of file.\n";
+
+    auto test_fun = [] {
+        logger l(log_level::info);
+        l.at_transport_error(asio::error::eof);
+    };
+
+    test_logger_output(std::move(test_fun), expected_output);
+}
+
 // Test that the mqtt_client calls logger functions as expected.
 
 BOOST_AUTO_TEST_CASE(client_disconnect) {
@@ -477,6 +491,65 @@ BOOST_AUTO_TEST_CASE(client_disconnect) {
     BOOST_TEST(log == expected_msg);
 }
 
+BOOST_AUTO_TEST_CASE(client_transport_error) {
+    using test::after;
+    using namespace std::chrono_literals;
+
+    const auto& success = success_msg();
+    const auto expected_msg =
+        "[Boost.MQTT5] resolve: 127.0.0.1:1883 - " + success + ".\n"
+        "[Boost.MQTT5] TCP connect: 127.0.0.1:1883 - " + success + ".\n"
+        "[Boost.MQTT5] transport layer error: End of file.\n"
+        "[Boost.MQTT5] resolve: 127.0.0.1:1883 - " + success + ".\n"
+        "[Boost.MQTT5] TCP connect: 127.0.0.1:1883 - " + success + ".\n"
+        "[Boost.MQTT5] connack: The operation completed successfully.\n"
+    ;
+
+    boost::test_tools::output_test_stream output;
+    {
+        clog_redirect guard(output.rdbuf());
+
+        // packets
+        auto connect = encoders::encode_connect(
+            "", std::nullopt, std::nullopt, 60, false, test::dflt_cprops, std::nullopt
+        );
+        auto connack = encoders::encode_connack(false, uint8_t(0x00), {});
+
+        const std::string topic = "topic", payload = "payload0";
+        const std::string publish = encoders::encode_publish(
+            0, topic, payload, qos_e::at_most_once, retain_e::no, dup_e::no, {}
+        );
+
+        test::msg_exchange broker_side;
+        broker_side
+            .expect(connect)
+                .complete_with(asio::error::eof, after(0ms))
+            .expect(connect)
+                .complete_with(error_code{}, after(0ms))
+                .reply_with(connack, after(10ms));
+
+        asio::io_context ioc;
+        auto executor = ioc.get_executor();
+        auto& broker = asio::make_service<test::test_broker>(
+            ioc, executor, std::move(broker_side)
+        );
+
+        mqtt_client<test::test_stream, std::monostate, logger> c(executor);
+        c.brokers("127.0.0.1,127.0.0.1") // to avoid reconnect backoff
+            .async_run(asio::detached);
+
+        asio::steady_timer timer(c.get_executor());
+        timer.expires_after(100ms);
+        timer.async_wait([&c](error_code) { c.cancel(); });
+
+        ioc.run();
+        BOOST_TEST(broker.received_all_expected());
+    }
+
+    std::string log = output.rdbuf()->str();
+    BOOST_TEST(log == expected_msg);
+}
+
 #ifdef BOOST_MQTT5_EXTRA_DEPS
 using stream_type = boost::beast::websocket::stream<
     asio::ssl::stream<asio::ip::tcp::socket>