From 15a3d62727f5be6c4822e716a3b039e43a57f1ad Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Korina=20=C5=A0imi=C4=8Devi=C4=87?= Date: Thu, 5 Oct 2023 10:19:37 +0200 Subject: [PATCH] [mqtt-test] cancellation tests from long time ago Summary: related to T12015 - added some cancellation tests (testing ioc.stop() & client.cancel() - some tests fail and some hang (on windows specifically) Reviewers: ivica Reviewed By: ivica Subscribers: miljen, iljazovic Differential Revision: https://repo.mireo.local/D26317 --- SConscript | 1 + include/async_mqtt5/impl/client_service.hpp | 6 + test/unit/test/cancellation.cpp | 246 ++++++++++++++++++++ win/test/mqtt-test.vcxproj | 1 + win/test/mqtt-test.vcxproj.filters | 3 + 5 files changed, 257 insertions(+) create mode 100644 test/unit/test/cancellation.cpp diff --git a/SConscript b/SConscript index a99589e..0fb7a67 100644 --- a/SConscript +++ b/SConscript @@ -23,6 +23,7 @@ test_sources = [ 'test/unit/test/publish_send_op.cpp', 'test/unit/test/client_broker.cpp', 'test/unit/test/coroutine.cpp', + 'test/unit/test/cancellation.cpp', 'test/unit/src/run_tests.cpp' ] diff --git a/include/async_mqtt5/impl/client_service.hpp b/include/async_mqtt5/impl/client_service.hpp index 65ce9f2..0345d5e 100644 --- a/include/async_mqtt5/impl/client_service.hpp +++ b/include/async_mqtt5/impl/client_service.hpp @@ -198,6 +198,12 @@ public: _cancel_ping.emit(asio::cancellation_type::terminal); _cancel_sentry.emit(asio::cancellation_type::terminal); + // cancelling the receive channel invokes all pending handlers with + // ec = asio::experimental::error::channel_cancelled + // adding another ec to the list of the possible client ecs + + // TODO: close() the channel instead, and open() it on the next run() + _rec_channel.cancel(); _replies.cancel_unanswered(); _async_sender.cancel(); _stream.close(); diff --git a/test/unit/test/cancellation.cpp b/test/unit/test/cancellation.cpp new file mode 100644 index 0000000..832d0f2 --- /dev/null +++ b/test/unit/test/cancellation.cpp @@ -0,0 +1,246 @@ +#include + +#include +#include +#include + +#include + +using namespace async_mqtt5; + +namespace async_mqtt5::test { + +enum cancellation_type { + ioc_stop = 1, + client_cancel +}; + +} // end namespace async_mqtt5::test + +template +void cancel_async_receive() { + using enum test::cancellation_type; + + constexpr int num_handlers = 3; + constexpr int expected_handlers_called = type == ioc_stop ? 0 : num_handlers; + int handlers_called = 0; + + asio::io_context ioc; + + using stream_type = asio::ip::tcp::socket; + using client_type = mqtt_client; + client_type c(ioc, ""); + + c.brokers("mqtt.mireo.local", 1883) + .run(); + + for (auto i = 0; i < num_handlers; ++i) + c.async_receive([&handlers_called]( + error_code ec, std::string, std::string, publish_props + ) { + BOOST_CHECK_EQUAL(ec, asio::experimental::error::channel_cancelled); + handlers_called++; + }); + + asio::steady_timer timer(c.get_executor()); + timer.expires_after(std::chrono::seconds(1)); + timer.async_wait([&](auto) { + if constexpr (type == ioc_stop) + ioc.stop(); + else + c.cancel(); + }); + + ioc.run(); + BOOST_CHECK_EQUAL( + handlers_called, expected_handlers_called + ); +} + +template +void cancel_async_publish() { + using enum test::cancellation_type; + + constexpr int expected_handlers_called = type == ioc_stop ? 0 : 3; + int handlers_called = 0; + + asio::io_context ioc; + + using stream_type = asio::ip::tcp::socket; + using client_type = mqtt_client; + client_type c(ioc, ""); + + c.brokers("mqtt.mireo.local", 1883) + .run(); + + c.async_publish( + "topic", "payload", retain_e::yes, {}, + [&handlers_called](error_code ec) { + BOOST_CHECK_EQUAL(ec, asio::error::operation_aborted); + handlers_called++; + } + ); + + c.async_publish( + "topic", "payload", retain_e::yes, {}, + [&handlers_called](error_code ec, reason_code rc, puback_props) { + BOOST_CHECK_EQUAL(ec, asio::error::operation_aborted); + handlers_called++; + } + ); + + + c.async_publish( + "topic", "payload", retain_e::yes, {}, + [&handlers_called](error_code ec, reason_code rc, pubcomp_props) { + BOOST_CHECK_EQUAL(ec, asio::error::operation_aborted); + handlers_called++; + } + ); + + asio::steady_timer timer(c.get_executor()); + timer.expires_after(std::chrono::seconds(1)); + timer.async_wait([&](auto) { + if constexpr (type == ioc_stop) + ioc.stop(); + else + c.cancel(); + }); + + ioc.run(); + BOOST_CHECK_EQUAL( + handlers_called, expected_handlers_called + ); +} + +template +void cancel_during_connecting() { + using enum test::cancellation_type; + + constexpr int expected_handlers_called = type == ioc_stop ? 0 : 1; + int handlers_called = 0; + + asio::io_context ioc; + + using stream_type = asio::ip::tcp::socket; + using client_type = mqtt_client; + client_type c(ioc, ""); + + c.brokers("127.0.0.1", 1883) + .run(); + + c.async_publish( + "topic", "payload", retain_e::yes, {}, + [&handlers_called](error_code ec) { + BOOST_CHECK_EQUAL(ec, asio::error::operation_aborted); + handlers_called++; + } + ); + + asio::steady_timer timer(c.get_executor()); + timer.expires_after(std::chrono::seconds(2)); + timer.async_wait([&](auto) { + if constexpr (type == ioc_stop) + ioc.stop(); + else + c.cancel(); + }); + + ioc.run(); + BOOST_CHECK_EQUAL( + handlers_called, expected_handlers_called + ); +} + + +BOOST_AUTO_TEST_SUITE(cancellation) + + +BOOST_AUTO_TEST_CASE(ioc_stop_async_receive) { + cancel_async_receive(); +} + + +BOOST_AUTO_TEST_CASE(client_cancel_async_receive) { + cancel_async_receive(); +} + +// passes on debian, hangs on windows in io_context destructor +BOOST_AUTO_TEST_CASE(ioc_stop_async_publish) { + cancel_async_publish(); +} + +BOOST_AUTO_TEST_CASE(client_cancel_async_publish) { + cancel_async_publish(); +} + +// passes on debian, hangs on windows +BOOST_AUTO_TEST_CASE(ioc_stop_cancel_during_connecting) { + cancel_during_connecting(); +} + +BOOST_AUTO_TEST_CASE(client_cancel_during_connecting) { + cancel_during_connecting(); +} + + +// last two ec checks fail, they finish with no_recovery when +// you don't call c.cancel() +BOOST_AUTO_TEST_CASE(rerunning_the_client) { + constexpr int expected_handlers_called = 3; + int handlers_called = 0; + + asio::io_context ioc; + + using stream_type = asio::ip::tcp::socket; + using client_type = mqtt_client; + client_type c(ioc, ""); + + c.brokers("mqtt.mireo.local", 1883) + .run(); + + c.async_publish( + "cancelled_topic", "cancelled_payload", retain_e::yes, {}, + [&handlers_called](error_code ec, reason_code rc, pubcomp_props) { + BOOST_CHECK_EQUAL(ec, asio::error::operation_aborted); + BOOST_CHECK_EQUAL(rc, reason_codes::empty); + handlers_called++; + } + ); + + asio::post(ioc, [&c] { c.cancel(); }); + + asio::steady_timer cancel_timer(c.get_executor()); + cancel_timer.expires_after(std::chrono::seconds(1)); + cancel_timer.async_wait( + [&](auto) { + c.run(); + + c.async_publish( + "test/mqtt-test", "payload", retain_e::yes, {}, + [&handlers_called](error_code ec) { + BOOST_CHECK(!ec); + handlers_called++; + } + ); + + c.async_receive([&handlers_called]( + error_code ec, std::string, std::string, publish_props + ) { + BOOST_CHECK(!ec); + handlers_called++; + }); + } + ); + + asio::steady_timer timer(c.get_executor()); + timer.expires_after(std::chrono::seconds(2)); + timer.async_wait([&](auto) { c.cancel(); }); + + ioc.run(); + BOOST_CHECK_EQUAL( + handlers_called, expected_handlers_called + ); +} + +BOOST_AUTO_TEST_SUITE_END(); diff --git a/win/test/mqtt-test.vcxproj b/win/test/mqtt-test.vcxproj index 9080c95..f57067d 100644 --- a/win/test/mqtt-test.vcxproj +++ b/win/test/mqtt-test.vcxproj @@ -29,6 +29,7 @@ + diff --git a/win/test/mqtt-test.vcxproj.filters b/win/test/mqtt-test.vcxproj.filters index e70f94c..49a4334 100644 --- a/win/test/mqtt-test.vcxproj.filters +++ b/win/test/mqtt-test.vcxproj.filters @@ -59,5 +59,8 @@ Source Files + + Source Files\test + \ No newline at end of file