Hi I've got a question regarding the behavior I've observed. It may not be a bug or sth, maybe I just use the endpoint in the wrong way. Anyway, I have an endpoint that I connect to a broker, keep alive is set to 15 sec. Once I establish a connection with the broker I move the endpoint variable into the other variable let's call it endpoint2 (it is a simplified example based on my code, just to ease reproduction). Then, using endpoint2 variable I try to receive some packets. After about 15 seconds (the same as keep alive) the code gets crashed.
The "stack trace" is pretty long thus I'm showing only the relevant parts:
endpoint.hpp:138:31: runtime error: member call on null pointer of type 'struct element_type'
pc_0x557f6e551b55###func_async_mqtt::basic_endpoint<(async_mqtt::role)1, 2ul, boost::asio::basic_stream_socket<boost::asio::ip::tcp, boost::asio::io_context::basic_executor_type<std::allocator, 0ul> > >::strand()###file_/home/bielpa/.priv/repo/hhctrl-mgmt/cmake-build-linux_gcc_debug_integration_tests_sanitizers/deps/async_mqtt-src-src/include/async_mqtt/endpoint.hpp###line_138###obj(home_assistant_it_v2+0x110db55)
pc_0x557f6e859bb9###func_void async_mqtt::basic_endpoint<(async_mqtt::role)1, 2ul, boost::asio::basic_stream_socket<boost::asio::ip::tcp, boost::asio::io_context::basic_executor_type<std::allocator, 0ul> > >::send_impl<async_mqtt::v3_1_1::pingreq_packet>::operator()<boost::asio::detail::composed_op<async_mqtt::basic_endpoint<(async_mqtt::role)1, 2ul, boost::asio::basic_stream_socket<boost::asio::ip::tcp, boost::asio::io_context::basic_executor_type<std::allocator, 0ul> > >::send_impl<async_mqtt::v3_1_1::pingreq_packet>, boost::asio::detail::composed_work<void ()>, async_mqtt::basic_endpoint<(async_mqtt::role)1, 2ul, boost::asio::basic_stream_socket<boost::asio::ip::tcp, boost::asio::io_context::basic_executor_type<std::allocator, 0ul> > >::reset_pingreq_send_timer()::{lambda(boost::system::error_code const&)#1}::operator()(boost::system::error_code const&) const::{lambda(async_mqtt::system_error const&)#1}, void (async_mqtt::system_error)> >(boost::asio::detail::composed_op<async_mqtt::basic_endpoint<(async_mqtt::role)1, 2ul, boost::asio::basic_stream_socket<boost::asio::ip::tcp, boost::asio::io_context::basic_executor_type<std::allocator, 0ul> > >::send_impl<async_mqtt::v3_1_1::pingreq_packet>, boost::asio::detail::composed_work<void ()>, async_mqtt::basic_endpoint<(async_mqtt::role)1, 2ul, boost::asio::basic_stream_socket<boost::asio::ip::tcp, boost::asio::io_context::basic_executor_type<std::allocator, 0ul> > >::reset_pingreq_send_timer()::{lambda(boost::system::error_code const&)#1}::operator()(boost::system::error_code const&) const::{lambda(async_mqtt::system_error const&)#1}, void (async_mqtt::system_error)>&, boost::system::error_code const&, unsigned long)###file_/home/bielpa/.priv/repo/hhctrl-mgmt/cmake-build-linux_gcc_debug_integration_tests_sanitizers/deps/async_mqtt-src-src/include/async_mqtt/endpoint.hpp###line_860###obj(home_assistant_it_v2+0x1415bb9)
pc_0x557f6e83b430###func_void boost::asio::detail::composed_op<async_mqtt::basic_endpoint<(async_mqtt::role)1, 2ul, boost::asio::basic_stream_socket<boost::asio::ip::tcp, boost::asio::io_context::basic_executor_type<std::allocator, 0ul> > >::send_impl<async_mqtt::v3_1_1::pingreq_packet>, boost::asio::detail::composed_work<void ()>, async_mqtt::basic_endpoint<(async_mqtt::role)1, 2ul, boost::asio::basic_stream_socket<boost::asio::ip::tcp, boost::asio::io_context::basic_executor_type<std::allocator, 0ul> > >::reset_pingreq_send_timer()::{lambda(boost::system::error_code const&)#1}::operator()(boost::system::error_code const&) const::{lambda(async_mqtt::system_error const&)#1}, void (async_mqtt::system_error)>::operator()<>()###file_/home/bielpa/.conan/data/boost/1.82.0///package/8cc3305c27e5ff838d1c7590662e309638310dfc/include/boost/asio/compose.hpp###line_107###obj_(home_assistant_it_v2+0x13f7430)
pc_0x557f6e823011###func_void boost::asio::detail::initiate_composed_op<void (async_mqtt::system_error), void ()>::operator()<async_mqtt::basic_endpoint<(async_mqtt::role)1, 2ul, boost::asio::basic_stream_socket<boost::asio::ip::tcp, boost::asio::io_context::basic_executor_type<std::allocator, 0ul> > >::reset_pingreq_send_timer()::{lambda(boost::system::error_code const&)#1}::operator()(boost::system::error_code const&) const::{lambda(async_mqtt::system_error const&)#1}, async_mqtt::basic_endpoint<(async_mqtt::role)1, 2ul, boost::asio::basic_stream_socket<boost::asio::ip::tcp, boost::asio::io_context::basic_executor_type<std::allocator, 0ul> > >::send_impl<async_mqtt::v3_1_1::pingreq_packet> >(async_mqtt::basic_endpoint<(async_mqtt::role)1, 2ul, boost::asio::basic_stream_socket<boost::asio::ip::tcp, boost::asio::io_context::basic_executor_type<std::allocator, 0ul> > >::reset_pingreq_send_timer()::{lambda(boost::system::error_code const&)#1}::operator()(boost::system::error_code const&) const::{lambda(async_mqtt::system_error const&)#1}&&, async_mqtt::basic_endpoint<(async_mqtt::role)1, 2ul, boost::asio::basic_stream_socket<boost::asio::ip::tcp, boost::asio::io_context::basic_executor_type<std::allocator, 0ul> > >::send_impl<async_mqtt::v3_1_1::pingreq_packet>&&) const###file_/home/bielpa/.conan/data/boost/1.82.0///package/8cc3305c27e5ff838d1c7590662e309638310dfc/include/boost/asio/compose.hpp###line_275###obj_(home_assistant_it_v2+0x13df011)
I guess that after those 15 seconds the ping timer expires and the endpoint wants to send a ping req, however, due to the previous move operation on the endpoint variable something goes wrong. Thus my question is - Is the move operation on the endpoint that was already connected allowed? Of course, it can be easily mitigated by e.g. allocating the endpoint on the heap, however I would like to hear your opinion.
Below is the code you can use to reproduce this "issue".
#include <catch2/catch_all.hpp>
#include <boost/asio/io_context.hpp>
#include <boost/asio/use_awaitable.hpp>
#include <boost/asio/ip/tcp.hpp>
#include <boost/asio/co_spawn.hpp>
#include <boost/asio/connect.hpp>
#include <async_mqtt/all.hpp>
#include <spdlog/spdlog.h>
void rethrow(std::exception_ptr ptr)
{
if (ptr) {
std::rethrow_exception(ptr);
}
}
TEST_CASE("Client can connect to broker 2")
{
auto ioc = boost::asio::io_context{};
auto ep = async_mqtt::endpoint<async_mqtt::role::client, async_mqtt::protocol::mqtt> {
async_mqtt::protocol_version::v3_1_1, ioc.get_executor()
};
// NOLINTBEGIN
boost::asio::co_spawn(ioc.get_executor(), [&ep, &ioc]() mutable -> boost::asio::awaitable<void> {
// Resolve
auto resolver = boost::asio::ip::tcp::resolver{ep.next_layer().get_executor()};
auto eps = co_await resolver.async_resolve("127.0.0.1", "1883", boost::asio::use_awaitable);
// Connect
co_await boost::asio::async_connect(
ep.next_layer(),
eps,
boost::asio::use_awaitable);
// Send connect packet
auto packet = async_mqtt::v3_1_1::connect_packet {
true,
15,
async_mqtt::allocate_buffer("some_unique_id_1234567"),
std::nullopt,
async_mqtt::allocate_buffer("test_user"), // credentials for my broker
async_mqtt::allocate_buffer("test")
};
if (auto system_error = co_await ep.send(std::move(packet), boost::asio::use_awaitable)) {
throw std::runtime_error{system_error.what()};
}
// Recv connack packet
if (async_mqtt::packet_variant packet_variant = co_await ep.recv(boost::asio::use_awaitable)) {
packet_variant.visit(
async_mqtt::overload{
[](const auto&) {
}
});
} else {
throw std::runtime_error{packet_variant.get<async_mqtt::system_error>().message()};
}
auto new_ep = std::move(ep);
// Receive
while (true) {
if (auto recv_packet = co_await new_ep.recv(boost::asio::use_awaitable)) {
auto type = recv_packet.type();
if (!type) {
throw std::runtime_error{"Invalid packet"};
}
spdlog::get("async_mqtt_client")->debug("Packet received: {}", async_mqtt::control_packet_type_to_str(type.value()));
} else {
throw std::runtime_error{recv_packet.template get<async_mqtt::system_error>().message()};
}
}
}, rethrow);
// NOLINTEND
ioc.run();
}