feedc0de/boost_mqtt5
1
0
Fork 0
forked from boostorg/mqtt5
Code Pull Requests Activity

resend terminal disconnect

Browse Source 
Summary:
- if we are in the process of connecting when user invokes async_disconnect we should send a disconnect packet after connect completes
- move integration disconnect tests to integration folder

Reviewers: ivica

Reviewed By: ivica

Subscribers: korina

Differential Revision: https://repo.mireo.local/D28119
This commit is contained in:
Bruno Iljazovic
2024-02-23 14:38:57 +01:00
parent 2f055e8894
commit 701698e5ec
4 changed files with 319 additions and 236 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
include/async_mqtt5/impl/disconnect_op.hpp
View File

@ -98,14 +98,17 @@ public:
_svc_ptr->async_send(
wire_data,
no_serial, send_flag::terminal,
asio::consign(
asio::prepend(std::move(*this), on_disconnect {}),
std::move(disconnect)
asio::prepend(
std::move(*this),
on_disconnect {}, std::move(disconnect)
)
);
}
void operator()(on_disconnect, error_code ec) {
void operator()(
on_disconnect,
control_packet<allocator_type> disconnect, error_code ec
) {
// The connection must be closed even
// if we failed to send the DISCONNECT packet
// with Reason Code of 0x80 or greater.
@ -116,14 +119,17 @@ public:
)
return complete(asio::error::operation_aborted);
if (ec == asio::error::try_again) {
if (_context.terminal)
return send_disconnect(std::move(disconnect));
return complete(error_code {});
}
if (_context.terminal) {
_svc_ptr->cancel();
return complete(error_code {});
}
if (ec == asio::error::try_again)
return complete(error_code {});
_svc_ptr->close_stream();
_svc_ptr->open_stream();

4
include/async_mqtt5/mqtt_client.hpp
View File

@ -872,8 +872,8 @@ public:
*
* \par Completion condition
* The asynchronous operation will complete when one of the following conditions is true:\n
* - The Client has attempted to send a \__DISCONNECT\__ packet, regardless of whether
* the sending was successful or not.\n
* - The Client has sent a \__DISCONNECT\__ packet.\n
* - 5 seconds have elapsed without a successful send.\n
* - An error occurred. This is indicated by an associated \__ERROR_CODE\__ in the handler.\n
*
* \par Error codes

304
test/integration/disconnect.cpp Normal file
View File

@ -0,0 +1,304 @@
#include <boost/test/unit_test.hpp>
#include <boost/asio/detached.hpp>
#include <boost/asio/io_context.hpp>
#include <boost/asio/steady_timer.hpp>
#include <async_mqtt5/mqtt_client.hpp>
#include "test_common/message_exchange.hpp"
#include "test_common/test_broker.hpp"
#include "test_common/test_stream.hpp"
using namespace async_mqtt5;
BOOST_AUTO_TEST_SUITE(disconnect/*, *boost::unit_test::disabled()*/)
struct shared_test_data {
error_code success {};
error_code fail = asio::error::not_connected;
const std::string connect = encoders::encode_connect(
"", std::nullopt, std::nullopt, 60, false, {}, std::nullopt
);
const std::string connack = encoders::encode_connack(
true, reason_codes::success.value(), {}
);
const std::string disconnect = encoders::encode_disconnect(
reason_codes::normal_disconnection.value(), disconnect_props {}
);
};
using test::after;
using namespace std::chrono_literals;
using client_type = mqtt_client<test::test_stream>;
template <typename TestCase>
void run_test(test::msg_exchange broker_side, TestCase&& test_case) {
asio::io_context ioc;
auto executor = ioc.get_executor();
auto& broker = asio::make_service<test::test_broker>(
ioc, executor, std::move(broker_side)
);
asio::steady_timer timer(executor);
client_type c(executor);
c.brokers("127.0.0.1,127.0.0.1") // to avoid reconnect backoff
.async_run(asio::detached);
test_case(c, timer);
ioc.run();
BOOST_TEST(broker.received_all_expected());
}
BOOST_FIXTURE_TEST_CASE(successful_disconnect, shared_test_data) {
constexpr int expected_handlers_called = 1;
int handlers_called = 0;
test::msg_exchange broker_side;
broker_side
.expect(connect)
.complete_with(success, after(0ms))
.reply_with(connack, after(0ms))
.expect(disconnect)
.complete_with(success, after(0ms));
run_test(
std::move(broker_side),
[&](client_type& c, asio::steady_timer& timer) {
timer.expires_after(100ms);
timer.async_wait([&](error_code) {
c.async_disconnect(
[&](error_code ec) {
handlers_called++;
BOOST_TEST(!ec);
}
);
});
}
);
BOOST_TEST(handlers_called == expected_handlers_called);
}
BOOST_FIXTURE_TEST_CASE(successful_disconnect_in_queue, shared_test_data) {
constexpr int expected_handlers_called = 2;
int handlers_called = 0;
// packets
auto publish_qos0 = 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(success, after(1ms))
.reply_with(connack, after(2ms))
.expect(publish_qos0)
.complete_with(success, after(1ms))
.expect(disconnect)
.complete_with(success, after(0ms));
run_test(
std::move(broker_side),
[&](client_type& c, asio::steady_timer& timer) {
timer.expires_after(50ms);
timer.async_wait([&](error_code) {
c.async_publish<qos_e::at_most_once>(
"topic", "payload", retain_e::no, {},
[&handlers_called](error_code ec) {
BOOST_TEST(handlers_called == 0);
handlers_called++;
BOOST_TEST(!ec);
}
);
c.async_disconnect(
[&](error_code ec) {
BOOST_TEST(handlers_called == 1);
handlers_called++;
BOOST_TEST(!ec);
}
);
});
}
);
BOOST_TEST(handlers_called == expected_handlers_called);
}
BOOST_FIXTURE_TEST_CASE(disconnect_on_disconnected_client, shared_test_data) {
constexpr int expected_handlers_called = 1;
int handlers_called = 0;
test::msg_exchange broker_side;
broker_side
.expect(connect)
.expect(connect);
run_test(
std::move(broker_side),
[&](client_type& c, asio::steady_timer& timer) {
timer.expires_after(50ms);
timer.async_wait([&](error_code) {
c.async_disconnect(
[&](error_code ec) {
handlers_called++;
BOOST_TEST(ec == asio::error::operation_aborted);
}
);
});
}
);
BOOST_TEST(handlers_called == expected_handlers_called);
}
BOOST_FIXTURE_TEST_CASE(disconnect_in_queue_on_disconnected_client, shared_test_data) {
constexpr int expected_handlers_called = 2;
int handlers_called = 0;
test::msg_exchange broker_side;
broker_side
.expect(connect)
.expect(connect);
run_test(
std::move(broker_side),
[&](client_type& c, asio::steady_timer& timer) {
timer.expires_after(50ms);
timer.async_wait([&](error_code) {
c.async_publish<qos_e::at_most_once>(
"topic", "payload", retain_e::no, {},
[&handlers_called](error_code ec) {
BOOST_TEST(handlers_called == 1);
handlers_called++;
BOOST_TEST(ec == asio::error::operation_aborted);
}
);
c.async_disconnect(
[&](error_code ec) {
BOOST_TEST(handlers_called == 0);
handlers_called++;
BOOST_TEST(ec == asio::error::operation_aborted);
}
);
});
}
);
BOOST_TEST(handlers_called == expected_handlers_called);
}
BOOST_FIXTURE_TEST_CASE(resend_terminal_disconnect, shared_test_data) {
constexpr int expected_handlers_called = 1;
int handlers_called = 0;
test::msg_exchange broker_side;
broker_side
.expect(connect)
.complete_with(success, after(0ms))
.reply_with(connack, after(0ms))
.expect(disconnect)
.complete_with(success, after(0ms));
run_test(
std::move(broker_side),
[&](client_type& c, asio::steady_timer&) {
c.async_disconnect(
[&](error_code ec) {
handlers_called++;
BOOST_TEST(!ec);
}
);
}
);
BOOST_TEST(handlers_called == expected_handlers_called);
}
BOOST_FIXTURE_TEST_CASE(dont_resend_non_terminal_disconnect, shared_test_data) {
auto malformed_publish_1 = encoders::encode_publish(
1, "malformed topic", "malformed payload",
static_cast<qos_e>(0b11), retain_e::yes, dup_e::no, {}
);
auto malformed_publish_2 = encoders::encode_publish(
2, "malformed topic", "malformed payload",
static_cast<qos_e>(0b11), retain_e::yes, dup_e::no, {}
);
auto disconnect_malformed_publish = encoders::encode_disconnect(
reason_codes::malformed_packet.value(),
test::dprops_with_reason_string("Malformed PUBLISH received: QoS bits set to 0b11")
);
test::msg_exchange broker_side;
broker_side
.expect(connect)
.complete_with(success, after(0ms))
.reply_with(connack, after(0ms))
.send(malformed_publish_1, malformed_publish_2, after(10ms))
.expect(disconnect_malformed_publish)
.complete_with(success, after(0ms))
.expect(connect)
.complete_with(success, after(0ms))
.reply_with(connack, after(0ms));
run_test(
std::move(broker_side),
[&](client_type& c, asio::steady_timer& timer) {
timer.expires_after(50ms);
timer.async_wait([&](error_code) {
c.cancel();
});
}
);
}
BOOST_FIXTURE_TEST_CASE(omit_props, shared_test_data) {
constexpr int expected_handlers_called = 1;
int handlers_called = 0;
connack_props co_props;
co_props[prop::maximum_packet_size] = 20;
// packets
auto connack_with_max_packet = encoders::encode_connack(
false, reason_codes::success.value(), co_props
);
disconnect_props props;
props[prop::reason_string] = std::string(50, 'a');
auto big_disconnect = encoders::encode_disconnect(
reason_codes::normal_disconnection.value(), props
);
test::msg_exchange broker_side;
broker_side
.expect(connect)
.complete_with(success, after(0ms))
.reply_with(connack_with_max_packet, after(0ms))
.expect(disconnect)
.complete_with(success, after(0ms));
run_test(
std::move(broker_side),
[&](client_type& c, asio::steady_timer& timer) {
timer.expires_after(50ms);
timer.async_wait([&](error_code) {
c.async_disconnect(
disconnect_rc_e::normal_disconnection, props,
[&](error_code ec) {
handlers_called++;
BOOST_TEST(!ec);
}
);
});
}
);
BOOST_TEST(handlers_called == expected_handlers_called);
}
BOOST_AUTO_TEST_SUITE_END()

227
test/unit/disconnect_op.cpp
View File

@ -1,17 +1,10 @@
#include <boost/test/unit_test.hpp>
#include <boost/asio/io_context.hpp>
#include <boost/asio/steady_timer.hpp>
#include <async_mqtt5/mqtt_client.hpp>
#include <async_mqtt5/impl/disconnect_op.hpp>
#include <async_mqtt5/detail/internal_types.hpp>
#include "test_common/message_exchange.hpp"
#include "test_common/test_service.hpp"
#include "test_common/test_stream.hpp"
using namespace async_mqtt5;
@ -63,224 +56,4 @@ BOOST_AUTO_TEST_CASE(malformed_user_property_value) {
run_malformed_props_test(dprops);
}
struct shared_test_data {
error_code success {};
error_code fail = asio::error::not_connected;
const std::string connect = encoders::encode_connect(
"", std::nullopt, std::nullopt, 60, false, {}, std::nullopt
);
const std::string connack = encoders::encode_connack(
true, reason_codes::success.value(), {}
);
const std::string disconnect = encoders::encode_disconnect(
reason_codes::normal_disconnection.value(), disconnect_props {}
);
};
using test::after;
using namespace std::chrono_literals;
using client_type = mqtt_client<test::test_stream>;
template <typename TestCase>
void run_test(test::msg_exchange broker_side, TestCase&& test_case) {
asio::io_context ioc;
auto executor = ioc.get_executor();
auto& broker = asio::make_service<test::test_broker>(
ioc, executor, std::move(broker_side)
);
asio::steady_timer timer(executor);
client_type c(executor);
c.brokers("127.0.0.1")
.async_run(asio::detached);
test_case(c, timer);
ioc.run();
BOOST_TEST(broker.received_all_expected());
}
BOOST_FIXTURE_TEST_CASE(successful_disconnect, shared_test_data) {
constexpr int expected_handlers_called = 1;
int handlers_called = 0;
test::msg_exchange broker_side;
broker_side
.expect(connect)
.complete_with(success, after(0ms))
.reply_with(connack, after(0ms))
.expect(disconnect)
.complete_with(success, after(0ms));
run_test(
std::move(broker_side),
[&](client_type& c, asio::steady_timer& timer) {
timer.expires_after(100ms);
timer.async_wait([&](error_code) {
c.async_disconnect(
[&](error_code ec) {
handlers_called++;
BOOST_TEST(!ec);
}
);
});
}
);
BOOST_TEST(handlers_called == expected_handlers_called);
}
BOOST_FIXTURE_TEST_CASE(successful_disconnect_in_queue, shared_test_data) {
constexpr int expected_handlers_called = 2;
int handlers_called = 0;
// packets
auto publish_qos0 = 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(success, after(1ms))
.reply_with(connack, after(2ms))
.expect(publish_qos0)
.complete_with(success, after(1ms))
.expect(disconnect)
.complete_with(success, after(0ms));
run_test(
std::move(broker_side),
[&](client_type& c, asio::steady_timer& timer) {
timer.expires_after(50ms);
timer.async_wait([&](error_code) {
c.async_publish<qos_e::at_most_once>(
"topic", "payload", retain_e::no, {},
[&handlers_called](error_code ec) {
BOOST_TEST(handlers_called == 0);
handlers_called++;
BOOST_TEST(!ec);
}
);
c.async_disconnect(
[&](error_code ec) {
BOOST_TEST(handlers_called == 1);
handlers_called++;
BOOST_TEST(!ec);
}
);
});
}
);
BOOST_TEST(handlers_called == expected_handlers_called);
}
BOOST_FIXTURE_TEST_CASE(disconnect_on_disconnected_client, shared_test_data) {
constexpr int expected_handlers_called = 1;
int handlers_called = 0;
test::msg_exchange broker_side;
broker_side
.expect(connect);
run_test(
std::move(broker_side),
[&](client_type& c, asio::steady_timer& timer) {
timer.expires_after(50ms);
timer.async_wait([&](error_code) {
c.async_disconnect(
[&](error_code ec) {
handlers_called++;
BOOST_TEST(ec == asio::error::operation_aborted);
}
);
});
}
);
BOOST_TEST(handlers_called == expected_handlers_called);
}
BOOST_FIXTURE_TEST_CASE(disconnect_in_queue_on_disconnected_client, shared_test_data) {
constexpr int expected_handlers_called = 2;
int handlers_called = 0;
test::msg_exchange broker_side;
broker_side
.expect(connect);
run_test(
std::move(broker_side),
[&](client_type& c, asio::steady_timer& timer) {
timer.expires_after(50ms);
timer.async_wait([&](error_code) {
c.async_publish<qos_e::at_most_once>(
"topic", "payload", retain_e::no, {},
[&handlers_called](error_code ec) {
BOOST_TEST(handlers_called == 1);
handlers_called++;
BOOST_TEST(ec == asio::error::operation_aborted);
}
);
c.async_disconnect(
[&](error_code ec) {
BOOST_TEST(handlers_called == 0);
handlers_called++;
BOOST_TEST(ec == asio::error::operation_aborted);
}
);
});
}
);
BOOST_TEST(handlers_called == expected_handlers_called);
}
BOOST_FIXTURE_TEST_CASE(omit_props, shared_test_data) {
constexpr int expected_handlers_called = 1;
int handlers_called = 0;
connack_props co_props;
co_props[prop::maximum_packet_size] = 20;
// packets
auto connack_with_max_packet = encoders::encode_connack(
false, reason_codes::success.value(), co_props
);
disconnect_props props;
props[prop::reason_string] = std::string(50, 'a');
auto big_disconnect = encoders::encode_disconnect(
reason_codes::normal_disconnection.value(), props
);
test::msg_exchange broker_side;
broker_side
.expect(connect)
.complete_with(success, after(0ms))
.reply_with(connack_with_max_packet, after(0ms))
.expect(disconnect)
.complete_with(success, after(0ms));
run_test(
std::move(broker_side),
[&](client_type& c, asio::steady_timer& timer) {
timer.expires_after(50ms);
timer.async_wait([&](error_code) {
c.async_disconnect(
disconnect_rc_e::normal_disconnection, props,
[&](error_code ec) {
handlers_called++;
BOOST_TEST(!ec);
}
);
});
}
);
BOOST_TEST(handlers_called == expected_handlers_called);
}
BOOST_AUTO_TEST_SUITE_END()